IcebergDocument iterators leak S3/Parquet input streams on probe and bounded reads
- Dominant language
- Scala
- Stars
- 314
- Forks
- 187
- Avg merge
- 1d 21h
- Merged PRs (30d)
- 214
Description
### 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.(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` (429be110a7), 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).
Contributor guide
Assessment
This issue has not been assessed yet.