IcebergDocument iterators leak S3/Parquet input streams on probe and bounded reads

[aglinxinyuan]

What happened?

Amber CI logs (and any deployment reading results from S3-backed Iceberg storage) fill with finalizer warnings like:

[WARN] [org.apache.iceberg.aws.s3.S3InputStream] [Finalizer] - Unclosed input stream created by:
	org.apache.iceberg.aws.s3.S3InputStream.<init>(S3InputStream.java:102)
	...
	org.apache.texera.amber.util.IcebergUtil$.readDataFileAsIterator(IcebergUtil.scala:464)
	org.apache.texera.amber.core.storage.result.iceberg.IcebergDocument$$anon$1.hasNext(IcebergDocument.scala:295)
	...
	org.apache.texera.amber.core.storage.model.VirtualDocumentSpec.$anonfun$$init$$6(VirtualDocumentSpec.scala:105)

plus the sibling Unclosed S3FileIO instance created by: warnings after IcebergRestCatalogIntegrationSpec.

These are real resource leaks, not just noise: each one is a Parquet reader + S3 input stream + AWS HTTP-pool slot held until GC finalization.

Root cause (reader streams): IcebergDocument.getUsingFileSequenceOrder's iterator opened the Parquet reader inside hasNext. The existing closes cover file-advance, exhaustion, and (lazily) the until limit — but two consumer shapes never hit any of them:

Consumer shape	Why it leaks
Probes — isEmpty / nonEmpty / lone hasNext	hasNext opens a stream to answer, caller abandons the iterator; no close point ever runs
Bounded reads — getRange(from, until) consumed to exactly the limit	The limit close lives in a subsequent hasNext call that bounded consumers never make

Root cause (S3FileIO): IcebergRestCatalogIntegrationSpec never closes its RESTCatalog; Iceberg 1.9.2 tracks per-table FileIO instances (FileIOTracker) and closes them with the catalog, so the missing close leaks one S3FileIO per createTable/loadTable.

Fix: make hasNext resource-free — it claims the next data file from FileScanTask.recordCount metadata alone (the same field the existing whole-file skip already trusts; planFiles() never splits files in Iceberg 1.9.2 and amber's write path is strictly append-only, so the count is exact). The Parquet reader opens lazily in next(), which also closes it deterministically the moment a bounded read has served its last record. Plus afterAll catalog close in the REST catalog spec.

Known residual leak paths (pre-existing, not regressed — follow-up candidates):

Call site	Shape
SyncExecutionResource.collectOperatorResult visualization branch	unbounded get() + next() + early return — leaks deterministically on every sync fetch of a visualization result; a bounded getRange(0, totalCount) would now close deterministically
SyncExecutionResource oversized-first-tuple return / swallowed-exception catch	abandons mid-file
ResultExportService CSV/Arrow streaming	leaks only if the HTTP client disconnects mid-download
InputPortMaterializationReaderThread	leaks only on exception/interrupt mid-replay

How to reproduce?

1. Run the amber storage/integration specs against S3-backed Iceberg storage (e.g. IcebergDocumentSpec, which inherits VirtualDocumentSpec — its "clear the document" test does document.get().nonEmpty), or call document.get().nonEmpty / getRange(a, b).toList against any populated S3-backed IcebergDocument.
2. Wait for a GC cycle (CI runs trigger plenty).
3. Observe [S3InputStream] Unclosed input stream created by … finalizer warnings pointing at IcebergUtil.readDataFileAsIterator via IcebergDocument$$anon$1.hasNext.

Version/Branch

main (429be11), Iceberg 1.9.2.

Relevant log output

See the stack traces under "What happened?" — observed in the amber-integration and amber CI jobs (e.g. runs on PR #6797).