fix(amber): close Iceberg reader streams leaked by iterator probes and bounded reads

amber/src/test/integration/org/apache/texera/amber/storage/iceberg/IcebergRestCatalogIntegrationSpec.scala

@@ -47,6 +47,17 @@ class IcebergRestCatalogIntegrationSpec extends AnyFlatSpec with BeforeAndAfterA
     )
   }
 
+  override def afterAll(): Unit = {
+    // RESTCatalog is Closeable: closing releases its HTTP client and the
+    // S3FileIO instances it created for table operations. Left unclosed,
+    // the finalizer reclaims them and logs "Unclosed S3FileIO instance"
+    // warnings with full stack traces after the spec finishes.
+    if (restCatalog != null) {
+      restCatalog.close()
+    }
+    super.afterAll()
+  }
+
   behavior of "Iceberg REST catalog"
 
   it should "round-trip table metadata via the REST catalog" in {

common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala

@@ -186,6 +186,53 @@ private[storage] class IcebergDocument[T >: Null <: AnyRef](
         private var currentRecordIterator: Iterator[Record] = Iterator.empty
         private var currentRecordIteratorCloser: AutoCloseable = () => ()
 
+        // Next file to read, claimed from usableFileIterator by hasNext but
+        // not opened until next() actually needs a record. hasNext answers
+        // from the file's recordCount metadata alone, so probes such as
+        // isEmpty/nonEmpty (which abandon the iterator right after hasNext)
+        // never leave a Parquet reader / S3 stream open behind them.
+        private var pendingFile: Option[FileScanTask] = None
+
+        // Idempotent release of the active file's reader (and the S3 stream
+        // beneath it). Also resets the record iterator so a closed reader is
+        // never polled again.
+        private def closeCurrentReader(): Unit = {
+          currentRecordIteratorCloser.close()
+          currentRecordIteratorCloser = () => ()
+          currentRecordIterator = Iterator.empty
+        }
+
+        // Open the file claimed by hasNext and point currentRecordIterator at
+        // its records, applying the one-off partial skip for the first file.
+        // Only next() calls this, keeping hasNext free of any Parquet/S3
+        // resource acquisition.
+        private def openPendingFile(): Unit = {
+          val task = pendingFile.getOrElse(
+            throw new IllegalStateException("no pending file to open")
+          )
+          pendingFile = None
+          val schemaToUse = columns match {
+            case Some(cols) => tableSchema.select(cols.asJava)
+            case None       => tableSchema
+          }
+          // Release the prior file's reader before opening the next.
+          closeCurrentReader()
+          val nextIter = IcebergUtil.readDataFileAsIterator(
+            task.file(),
+            schemaToUse,
+            table.get
+          )
+          currentRecordIteratorCloser = nextIter
+          currentRecordIterator = nextIter.asScala
+
+          // Skip records within the file if necessary
+          val recordsToSkipInFile = from - numOfSkippedRecords
+          if (recordsToSkipInFile > 0) {
+            currentRecordIterator = currentRecordIterator.drop(recordsToSkipInFile)
+            numOfSkippedRecords += recordsToSkipInFile
+          }
+        }
+
         // Util function to load the table's metadata
         private def loadTableMetadata(): Option[Table] = {
           IcebergUtil.loadTableMetadata(
@@ -266,8 +313,7 @@ private[storage] class IcebergDocument[T >: Null <: AnyRef](
         override def hasNext: Boolean = {
           if (numOfReturnedRecords >= totalRecordsToReturn) {
             // Caller-imposed limit reached; release the active file's reader.
-            currentRecordIteratorCloser.close()
-            currentRecordIteratorCloser = () => ()
+            closeCurrentReader()
             return false
           }
 
@@ -277,48 +323,54 @@ private[storage] class IcebergDocument[T >: Null <: AnyRef](
             return true
           }
 
-          if (!usableFileIterator.hasNext) {
-            usableFileIterator = seekToUsableFile()
-          }
+          // The active file (if any) is exhausted; release its reader before
+          // deciding from metadata whether more records exist.
+          closeCurrentReader()
 
-          while (!currentRecordIterator.hasNext && usableFileIterator.hasNext) {
-            val nextFile = usableFileIterator.next()
-            val schemaToUse = columns match {
-              case Some(cols) => tableSchema.select(cols.asJava)
-              case None       => tableSchema
+          if (pendingFile.isEmpty) {
+            if (!usableFileIterator.hasNext) {
+              usableFileIterator = seekToUsableFile()
             }
-            // Release the prior file's reader before opening the next.
-            currentRecordIteratorCloser.close()
-            val nextIter = IcebergUtil.readDataFileAsIterator(
-              nextFile.file(),
-              schemaToUse,
-              table.get
-            )
-            currentRecordIteratorCloser = nextIter
-            currentRecordIterator = nextIter.asScala
-
-            // Skip records within the file if necessary
-            val recordsToSkipInFile = from - numOfSkippedRecords
-            if (recordsToSkipInFile > 0) {
-              currentRecordIterator = currentRecordIterator.drop(recordsToSkipInFile)
-              numOfSkippedRecords += recordsToSkipInFile
+            // Claim the next file that still has usable records, judging by
+            // recordCount metadata alone (the same field the whole-file skip
+            // in seekToUsableFile already relies on). The partial skip
+            // `from - numOfSkippedRecords` can only be non-zero for the first
+            // claimed file: seekToUsableFile's dropWhile guarantees that file
+            // satisfies recordCount > partial skip, and openPendingFile zeroes
+            // the skip before any later file is judged here.
+            while (pendingFile.isEmpty && usableFileIterator.hasNext) {
+              val task = usableFileIterator.next()
+              if (task.file().recordCount() > from - numOfSkippedRecords) {
+                pendingFile = Some(task)
+              }
             }
           }
 
-          val hasMore = currentRecordIterator.hasNext
-          if (!hasMore) {
-            // All files exhausted; release the last file's reader.
-            currentRecordIteratorCloser.close()
-            currentRecordIteratorCloser = () => ()
-          }
-          hasMore
+          pendingFile.nonEmpty
         }
 
         override def next(): T = {
           if (!hasNext) throw new NoSuchElementException("No more records available")
 
+          // hasNext only claims files by their metadata; the Parquet reader is
+          // opened lazily here. The loop is defensive: should a claimed file
+          // yield nothing after the partial skip, hasNext claims the next one
+          // (or reports exhaustion).
+          while (!currentRecordIterator.hasNext) {
+            openPendingFile()
+            if (!currentRecordIterator.hasNext && !hasNext) {
+              throw new NoSuchElementException("No more records available")
+            }
+          }
+
           val record = currentRecordIterator.next()
           numOfReturnedRecords += 1
+          if (numOfReturnedRecords >= totalRecordsToReturn) {
+            // Bounded read fully served: release the reader now instead of
+            // waiting for a further hasNext call that bounded consumers
+            // (e.g. getRange) rarely make.
+            closeCurrentReader()
+          }
           val schemaToUse = columns match {
             case Some(cols) => tableSchema.select(cols.asJava)
             case None       => tableSchema