[SPARK-23457] Register task completion listeners first for ParquetFileFormat

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 2.3.0
• Fix Version/s: 2.3.1, 2.4.0
• Component/s: SQL
• Assignee: Dongjoon Hyun
• Reporter: Dongjoon Hyun
• Created: 17/Feb/18 17:15
• Updated: 05/Mar/18 19:54
• Resolved: 20/Feb/18 05:36

Description

ParquetFileFormat leaks open files in some cases. This issue aims to register task completion listener first.

  test("SPARK-23390 Register task completion listeners first in ParquetFileFormat") {
    withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_BATCH_SIZE.key -> s"${Int.MaxValue}") {
      withTempDir { dir =>
        val basePath = dir.getCanonicalPath
        Seq(0).toDF("a").write.format("parquet").save(new Path(basePath, "first").toString)
        Seq(1).toDF("a").write.format("parquet").save(new Path(basePath, "second").toString)
        val df = spark.read.parquet(
          new Path(basePath, "first").toString,
          new Path(basePath, "second").toString)
        val e = intercept[SparkException] {
          df.collect()
        }
        assert(e.getCause.isInstanceOf[OutOfMemoryError])
      }
    }
  }

Issue Links

• links to: [Github] Pull Request #20619 (dongjoon-hyun)
• links to: [Github] Pull Request #20714 (dongjoon-hyun)