[Spark] Fix resource leaking when a deletion vector file is not found

delta-io · Sep 27, 2023 · 9190330 · 9190330
1 parent 7f8fe8b
commit 9190330
Showing 1 changed file with 20 additions and 8 deletions.
diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/DeltaParquetFileFormat.scala b/spark/src/main/scala/org/apache/spark/sql/delta/DeltaParquetFileFormat.scala
@@ -164,14 +164,26 @@ case class DeltaParquetFileFormat(
     val useOffHeapBuffers = sparkSession.sessionState.conf.offHeapColumnVectorEnabled
     (partitionedFile: PartitionedFile) => {
       val rowIteratorFromParquet = parquetDataReader(partitionedFile)
-      val iterToReturn =
-        iteratorWithAdditionalMetadataColumns(
-          partitionedFile,
-          rowIteratorFromParquet,
-          isRowDeletedColumn,
-          useOffHeapBuffers = useOffHeapBuffers,
-          rowIndexColumn = rowIndexColumn)
-      iterToReturn.asInstanceOf[Iterator[InternalRow]]
+      try {
+        val iterToReturn =
+          iteratorWithAdditionalMetadataColumns(
+            partitionedFile,
+            rowIteratorFromParquet,
+            isRowDeletedColumn,
+            useOffHeapBuffers = useOffHeapBuffers,
+            rowIndexColumn = rowIndexColumn)
+        iterToReturn.asInstanceOf[Iterator[InternalRow]]
+      } catch {
+        case NonFatal(e) =>
+          // Close the iterator if it is a closeable resource. The `ParquetFileFormat` opens
+          // the file and returns `RecordReaderIterator` (which implements `AutoCloseable` and
+          // `Iterator`) instance as a `Iterator`.
+          rowIteratorFromParquet match {
+            case resource: AutoCloseable => closeQuietly(resource)
+            case _ => // do nothing
+          }
+          throw e
+      }
     }
   }