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

[aglinxinyuan]

What changes were proposed in this PR?

Root cause. IcebergDocument.getUsingFileSequenceOrder's iterator opened the Parquet reader (and the S3InputStream beneath it) inside hasNext. Two consumer shapes then leak the stream until the GC finalizer reclaims it — producing the [S3InputStream] Unclosed input stream created by … warnings all over the amber CI jobs:

Consumer shape	Why it leaked
Probes — isEmpty / nonEmpty / lone hasNext (e.g. VirtualDocumentSpec's "clear the document" test)	hasNext opened a stream to answer, the caller abandoned the iterator; no close point ever ran
Bounded reads — getRange(from, until) consumed to exactly the limit	The limit close ran only on a subsequent hasNext call that bounded consumers never make

Fix — hasNext no longer acquires resources:

• hasNext claims the next data file from FileScanTask.recordCount metadata alone. This adds no new trust: the whole-file skip in seekToUsableFile already relies on the same field; planFiles() never splits files in Iceberg 1.9.2 (splitting is planTasks()-only) and amber's write paths are strictly append-only (no delete files anywhere in-repo), so the count is exact.
• The Parquet reader opens lazily in next() (openPendingFile()), the only resource-acquisition point.
• next() closes the reader deterministically the moment a bounded read has served its last record — before → after: getRange(a, b).toList used to keep the last file's stream open until GC; it now closes inside the final next().
• All closes funnel through an idempotent closeCurrentReader() that also resets the record iterator, so a closed reader is never polled (the old code did poll one on the exhaustion path — benign with today's Iceberg iterator internals, but the hazard is gone).

Record sequences, hasNext semantics, NoSuchElementException behavior, and the incremental-snapshot refresh (lastSnapshotId bookkeeping) are unchanged — only where streams open and close moved.

Also: close the RESTCatalog in IcebergRestCatalogIntegrationSpec.afterAll. Iceberg 1.9.2's RESTSessionCatalog tracks per-table FileIO instances (FileIOTracker) and closes them with the catalog, which removes the sibling Unclosed S3FileIO instance finalizer warnings.

Residual (pre-existing, untouched) abandonment paths — notably SyncExecutionResource.collectOperatorResult's visualization early-return — are inventoried in #6881 as follow-ups rather than expanded here.

Any related issues, documentation, discussions?

Closes #6881

How was this PR tested?

The changed paths are pinned by the existing VirtualDocumentSpec contract suite (probe, range, getAfter, incremental second-batch arrival, concurrent writes), which IcebergDocumentSpec runs against real Iceberg storage in the amber-integration CI job — those specs exercise every branch of the restructured iterator. Verified locally: WorkflowCore/Test/compile, WorkflowExecutionService/Test/compile, and scalafmtCheck on both source sets all pass.

Additionally verified by an exhaustive old-vs-new state-machine review covering: probe-only use, repeated hasNext, multi-file drains, bounded reads with partial mid-file skip, from beyond EOF, empty ranges, empty tables, snapshot arrival mid-read, zero-record files, loop termination, and closed-reader polling — record sequences and hasNext booleans are identical in every scenario; the only differences are the intended open/close points. The partial-skip invariant (from - numOfSkippedRecords non-zero only for the first claimed file) follows from seekToUsableFile's dropWhile guarantee and is documented in the code.

No new unit test is added because the leak itself is only observable through GC-finalizer instrumentation; the observable regression signal is the disappearance of the Unclosed input stream warnings from the amber CI logs.

Was this PR authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Opus 4.8 [1M context])

[github-actions]

Automated Reviewer Suggestions

Based on the git blame history of the changed files, we recommend the following reviewers:

• Contributors with relevant context: @Ma77Ball, @mengw15
  You can notify them by mentioning @Ma77Ball, @mengw15 in a comment.

[Copilot]

Pull request overview

This PR fixes real resource leaks in Amber’s Iceberg result iteration by ensuring Parquet/S3 readers are opened only when records are actually consumed and are closed deterministically for bounded reads. This change lives in the Iceberg-backed result storage layer (common/workflow-core) and adds a missing catalog close in an Amber integration spec to eliminate Iceberg finalizer warnings.

Changes:

• Refactors IcebergDocument iteration so hasNext no longer opens Parquet/S3 resources; readers are opened in next() and closed via a centralized idempotent close path.
• Ensures bounded reads (e.g., getRange(...).toList) close the active reader immediately after serving the final record rather than waiting for a subsequent hasNext.
• Closes RESTCatalog in IcebergRestCatalogIntegrationSpec.afterAll to release HTTP client + tracked S3FileIO instances.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.

File	Description
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala	Moves resource acquisition out of hasNext, adds pending-file claiming + deterministic reader close to prevent Parquet/S3 stream leaks.
amber/src/test/integration/org/apache/texera/amber/storage/iceberg/IcebergRestCatalogIntegrationSpec.scala	Closes RESTCatalog in afterAll to prevent S3FileIO finalizer leak warnings in integration tests.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 89.18919% with 4 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.97%. Comparing base (429be11) to head (be111a9).
⚠️ Report is 2 commits behind head on main.

Files with missing lines	Patch %	Lines
.../core/storage/result/iceberg/IcebergDocument.scala	89.18%	1 Missing and 3 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #6882      +/-   ##
============================================
+ Coverage     77.31%   82.97%   +5.66%     
+ Complexity     3524     2356    -1168     
============================================
  Files          1161      892     -269     
  Lines         45920    37603    -8317     
  Branches       5099     3963    -1136     
============================================
- Hits          35501    31200    -4301     
+ Misses         8839     5269    -3570     
+ Partials       1580     1134     -446     

Flag	Coverage Δ		*Carryforward flag
access-control-service	70.00% <ø> (ø)
agent-service	76.76% <ø> (ø)		Carriedforward from 429be11
amber	85.93% <89.18%> (+16.82%)	⬆️
computing-unit-managing-service	20.49% <ø> (ø)
config-service	66.66% <ø> (ø)
file-service	67.21% <ø> (ø)
frontend	82.58% <ø> (ø)		Carriedforward from 429be11
notebook-migration-service	78.94% <ø> (ø)
pyamber	92.15% <ø> (ø)		Carriedforward from 429be11
workflow-compiling-service	55.14% <ø> (ø)

*This pull request uses carry forward flags. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

⚠️ Benchmark changes need a look

🟢 0 better · 🔴 6 worse · ⚪ 9 noise (<±5%) · 0 without baseline

Compared against main 429be11 benchmarked on this same runner, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.

Dashboard · Run

	config	throughput	MB/s	latency	max Δ latest / 7d
🔴	bs=10 sw=10 sl=64	381	0.233	26,836/37,319/37,319 us	🔴 +7.3% / 🔴 +124.9%
🔴	bs=100 sw=10 sl=64	798	0.487	121,066/164,243/164,243 us	🔴 +16.8% / 🔴 +50.5%
⚪	bs=1000 sw=10 sl=64	922	0.563	1,086,229/1,108,232/1,108,232 us	⚪ within ±5% / 🔴 +8.4%

Baseline details

Latest main 429be11 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	381 tuples/sec	395 tuples/sec	754.55 tuples/sec	-3.5%	-49.5%
bs=10 sw=10 sl=64	MB/s	0.233 MB/s	0.241 MB/s	0.461 MB/s	-3.3%	-49.4%
bs=10 sw=10 sl=64	p50	26,836 us	25,015 us	12,816 us	+7.3%	+109.4%
bs=10 sw=10 sl=64	p95	37,319 us	35,227 us	16,594 us	+5.9%	+124.9%
bs=10 sw=10 sl=64	p99	37,319 us	35,227 us	19,806 us	+5.9%	+88.4%
bs=100 sw=10 sl=64	throughput	798 tuples/sec	840 tuples/sec	969.38 tuples/sec	-5.0%	-17.7%
bs=100 sw=10 sl=64	MB/s	0.487 MB/s	0.513 MB/s	0.592 MB/s	-5.1%	-17.7%
bs=100 sw=10 sl=64	p50	121,066 us	119,223 us	103,584 us	+1.5%	+16.9%
bs=100 sw=10 sl=64	p95	164,243 us	140,620 us	109,097 us	+16.8%	+50.5%
bs=100 sw=10 sl=64	p99	164,243 us	140,620 us	117,304 us	+16.8%	+40.0%
bs=1000 sw=10 sl=64	throughput	922 tuples/sec	900 tuples/sec	1,004 tuples/sec	+2.4%	-8.1%
bs=1000 sw=10 sl=64	MB/s	0.563 MB/s	0.549 MB/s	0.613 MB/s	+2.6%	-8.1%
bs=1000 sw=10 sl=64	p50	1,086,229 us	1,108,891 us	1,002,357 us	-2.0%	+8.4%
bs=1000 sw=10 sl=64	p95	1,108,232 us	1,145,213 us	1,046,463 us	-3.2%	+5.9%
bs=1000 sw=10 sl=64	p99	1,108,232 us	1,145,213 us	1,073,661 us	-3.2%	+3.2%

Raw CSV

config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,524.81,200,128000,381,0.233,26836.05,37319.02,37319.02
1,100,10,64,20,2504.70,2000,1280000,798,0.487,121066.11,164242.86,164242.86
2,1000,10,64,20,21700.57,20000,12800000,922,0.563,1086229.06,1108232.49,1108232.49