refactor(amber): read the loop input table from its materialization instead of shipping it in state - #6971
Open
aglinxinyuan wants to merge 6 commits into
Open
refactor(amber): read the loop input table from its materialization instead of shipping it in state#6971aglinxinyuan wants to merge 6 commits into
aglinxinyuan wants to merge 6 commits into
Conversation
…nstead of shipping it in state The loop's input table used to ride INSIDE the State content: LoopStart encoded its buffered input as Arrow IPC bytes, base64'd inside the JSON content, and that payload was re-written and re-read at every loop-body hop, every iteration (~33% base64 bloat, JSON-column size limits, and a second, divergent Arrow codec via pa.Table.from_pandas inference). That data already exists: in the fully-materialized mode loops require, the Loop Start's input-port materialization holds exactly the loop's input table for the whole loop (Loop Start re-reads it every iteration; the back-edge truncates only the state doc at the same base URI, never the result doc). So ship the port's BASE URI in the setup config and derive both addresses from it: - proto: InitializeExecutorRequest.loopStartStateUris -> loopStartPortUris (field 4 unchanged); the value is now the base URI. WorkflowExecutionManager ships storagePairs.head._1 directly. - back-edge write: main_loop derives state_uri(base) (unchanged behavior, one derivation later). - table read: on the matching consume, MainLoop reads result_uri(base) and injects the table into the LoopEnd via a new runtime-only attach_loop_table hook; a read failure is reported like a UDF error (report_exception) instead of silently dropping the consume. - LoopStartOperator.produce_state_on_finish emits only the user loop variables (the reserved-`table` collision raise stays: a user var named `table` would now be silently shadowed by the injected table); run_update uses the attached table; the "consumed" marker (_loop_table) is still set only by a successful update, so condition()'s short-circuit semantics are unchanged. - table_to_ipc_bytes / table_from_ipc_bytes and the now-unused TableOperator._buffered_table accessor are deleted. Nested loops keep working by construction: the inner Loop Start's map entry points at the outer Loop Start's output port, whose result doc is recreated per OUTER iteration but persists across inner iterations (the jump rewinds to the inner level only). Tests updated across test_loop_operators (attach-based flow, plus new pins: attach alone does not mark the loop consumed; run_update fails loud when no table was attached; produced state carries no table), test_main_loop (base-URI derivation for the back-edge write, consume injects the read table), test_initialize_executor_handler (renamed field), and the Scala ctor sites. The four LoopIntegrationSpec e2e cases exercise the full path in CI.
Contributor
Automated Reviewer SuggestionsBased on the
|
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #6971 +/- ##
=========================================
Coverage 79.37% 79.37%
- Complexity 3788 3789 +1
=========================================
Files 1158 1159 +1
Lines 46131 46114 -17
Branches 5124 5124
=========================================
- Hits 36618 36605 -13
+ Misses 7889 7887 -2
+ Partials 1624 1622 -2
*This pull request uses carry forward flags. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Contributor
|
| config | throughput | MB/s | latency | max Δ latest / 7d | |
|---|---|---|---|---|---|
| 🔴 | bs=10 sw=10 sl=64 | 545 | 0.332 | 18,222/27,524/27,524 us | 🔴 +14.0% / 🔴 +74.5% |
| 🔴 | bs=100 sw=10 sl=64 | 1,160 | 0.708 | 84,652/109,903/109,903 us | 🔴 +15.0% / 🟢 +16.0% |
| 🟢 | bs=1000 sw=10 sl=64 | 1,356 | 0.828 | 737,712/764,288/764,288 us | 🟢 -10.7% / 🟢 +31.5% |
Baseline details
Latest main 7d5276d from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 545 tuples/sec | 589 tuples/sec | 786.12 tuples/sec | -7.5% | -30.7% |
| bs=10 sw=10 sl=64 | MB/s | 0.332 MB/s | 0.36 MB/s | 0.48 MB/s | -7.8% | -30.8% |
| bs=10 sw=10 sl=64 | p50 | 18,222 us | 15,984 us | 12,305 us | +14.0% | +48.1% |
| bs=10 sw=10 sl=64 | p95 | 27,524 us | 25,624 us | 15,774 us | +7.4% | +74.5% |
| bs=10 sw=10 sl=64 | p99 | 27,524 us | 25,624 us | 18,978 us | +7.4% | +45.0% |
| bs=100 sw=10 sl=64 | throughput | 1,160 tuples/sec | 1,198 tuples/sec | 999.71 tuples/sec | -3.2% | +16.0% |
| bs=100 sw=10 sl=64 | MB/s | 0.708 MB/s | 0.731 MB/s | 0.61 MB/s | -3.1% | +16.0% |
| bs=100 sw=10 sl=64 | p50 | 84,652 us | 81,432 us | 100,616 us | +4.0% | -15.9% |
| bs=100 sw=10 sl=64 | p95 | 109,903 us | 95,530 us | 107,356 us | +15.0% | +2.4% |
| bs=100 sw=10 sl=64 | p99 | 109,903 us | 95,530 us | 113,255 us | +15.0% | -3.0% |
| bs=1000 sw=10 sl=64 | throughput | 1,356 tuples/sec | 1,349 tuples/sec | 1,031 tuples/sec | +0.5% | +31.5% |
| bs=1000 sw=10 sl=64 | MB/s | 0.828 MB/s | 0.823 MB/s | 0.63 MB/s | +0.6% | +31.5% |
| bs=1000 sw=10 sl=64 | p50 | 737,712 us | 734,536 us | 980,328 us | +0.4% | -24.7% |
| bs=1000 sw=10 sl=64 | p95 | 764,288 us | 855,425 us | 1,027,528 us | -10.7% | -25.6% |
| bs=1000 sw=10 sl=64 | p99 | 764,288 us | 855,425 us | 1,054,298 us | -10.7% | -27.5% |
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,367.27,200,128000,545,0.332,18222.41,27523.75,27523.75
1,100,10,64,20,1724.81,2000,1280000,1160,0.708,84652.00,109903.00,109903.00
2,1000,10,64,20,14747.63,20000,12800000,1356,0.828,737711.78,764287.87,764287.87
mengw15
reviewed
Jul 29, 2026
| # Reading it here at consume time means the table never has to ride | ||
| # inside the State content through the loop body. | ||
| result_uri = VFSURIFactory.result_uri(self._loop_start_base_uri()) | ||
| document, _ = DocumentFactory.open_document(result_uri) |
Contributor
There was a problem hiding this comment.
amber-integration is red on both ubuntu and macos here: LoopIntegrationSpec never completes (job cancelled at 20m), with Access Denied errors from the storage read path. The same suite is green on #6884 and on #6974 ~1.5h earlier with none of those, so it looks like it may be coming from this PR rather than environment noise.
Contributor
Author
There was a problem hiding this comment.
Fixed — both integration jobs are green now (~9 min each, was a 20 min cancel).
Cause was the timing of the new read, not the read itself: at consume time this worker's own materialization reader is still streaming, and issuing a second iceberg/S3 read from the main loop thread in that window made the reader fail with Access Denied. The read now happens at EndChannel, after the reader has finished — the matching consume emits nothing downstream, so nothing else moves.
(The macos run also hit the unrelated worker still has unprocessed messages race that #6960 fixes; it passed on re-run.)
aglinxinyuan
marked this pull request as draft
July 29, 2026 05:05
…lap the reader CI experiment for the amber-integration failure reported in review: the Access Denied errors are NOT in the new read path (0 of 56 tracebacks contain _read_loop_input_table) and never on the doc it opens. They are in this worker's own materialization READER thread, on loop-INTERNAL docs (LoopStart / Limit outputs) that region re-execution drops and recreates while IcebergDocument.get() iterates a lazily-pinned snapshot -- MinIO answers a deleted key with 403 Access Denied. The read itself is fine (the Loop Start's input-port materialization is created once upstream of the loop and is stable for the whole run), but issuing it at consume time puts a second iceberg/S3 read on the main loop thread WHILE that reader is still streaming. Move it out of that window: stash the matching state at consume and run the operator's update (with the table attached) at EndChannel, by which point the reader has finished. If this turns amber-integration green, the overlap was the trigger; if the Access Denied errors persist, the recreate-vs-active-reader race is an engine bug independent of this PR. The matching consume emits no state downstream, so deferring it changes nothing observable outside the operator.
aglinxinyuan
marked this pull request as ready for review
July 29, 2026 05:59
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this PR?
The loop's input table used to ride inside the State content: LoopStart encoded its buffered input as Arrow IPC bytes, base64'd into the JSON
contentcolumn, and that payload was re-written and re-read at every loop-body hop, every iteration.That data already exists. In the fully-materialized mode loops require, the Loop Start's input-port materialization holds exactly the loop's input table for the whole loop — Loop Start re-reads it every iteration, and the back-edge truncates only the state doc at the same base URI, never the result doc. So this PR ships the port's base URI in the setup config and derives both addresses from it:
loopStartStateUris= state URIloopStartPortUris= base URI (renamed so the semantic change is loud)result_uri(base)at EndChannel, injected via a new runtime-onlyattach_loop_tablehooktable_to_ipc_bytes/table_from_ipc_bytesfrom_pandasinference)When the read happens matters. The read is issued at EndChannel (the matching state is stashed at consume and the operator's update runs in
complete()), not at consume time. At consume time this worker's own materialization reader is still streaming, and issuing a second iceberg/S3 read from the main loop thread in that window made the reader fail with S3Access Denied—LoopIntegrationSpechung to the CI job timeout on both OSes. Deferring past the reader removes the overlap: integration went from a 20-minute cancel to green in ~9 minutes. The matching consume emits no state downstream, so moving it is unobservable outside the operator.Semantics deliberately preserved:
tablecollision raise stays (a user var namedtablewould now be silently shadowed by the injected table — worse than before);_loop_table) is still set only by a successfulrun_update, socondition()'s short-circuit for pass-through-only Loop Ends is unchanged;Wins: no ~33% base64 bloat, no JSON-column size ceiling on the table (large-table loops become viable), strictly less I/O for any non-empty loop body (one read per iteration replaces N state-doc writes+reads per hop), and one Arrow codec instead of two.
Note: this deepens the read-side use of
storagePairs.head._1— the same shared upstream URI as the known back-edge fan-out design discussion; if that ever moves to a per-loop private doc, this read moves with it.Any related issues, documentation, discussions?
Builds on #5900 (State columns) and #6661 (envelope through JVM hops). Related design context: #6660.
How was this PR tested?
test_loop_operators.pyrewritten for the attach-based flow plus new pins: the produced state carries notable; attaching alone does not mark the loop consumed;run_updatefails loud when no table was attached.test_main_loop.pypins the base-URI derivation for the back-edge write (state_uri(base)), the missing-config fail-loud, and that the matching consume stashes the state without touching storage, with the read + update happening once at EndChannel.test_initialize_executor_handler.pycovers the renamed proto field. 237 tests green locally (the only failures in a full sweep are pre-existing environment ones, identical on unmodified main).scalafmtCheckAll,scalafixAll --check, and the worker/descriptor spec suites (WorkerSpec,WorkflowWorkerSpec,SerializationManagerSpec,WorkflowExecutionManagerSpec,LoopStartOpDescSpec,LoopEndOpDescSpec) all pass on Java 17.LoopIntegrationSpeccases (single, nested 3×3, JVM chain, nested JVM chain) exercise the full read path in theamber-integrationCI job (both jobs green in ~9 min); the nested cases specifically cover the inner-loop read against the outer Loop Start's per-outer-iteration output doc.Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Fable 5)