feat(engine): add TryCatch/Finally control blocks with in-band error handling - #7388
Draft
carloea2 wants to merge 9 commits into
Draft
feat(engine): add TryCatch/Finally control blocks with in-band error handling#7388carloea2 wants to merge 9 commits into
carloea2 wants to merge 9 commits into
Conversation
…handling Introduce block-level try/catch/finally semantics for workflows: - TryCatch logical operator (Try / Catch / Error Info ports) expanding to a splitter + catch gate; Finally logical operator (From Try / From Catch in, Try Result / Catch Result out) backed by a staging merger that releases exactly one branch's complete output through the port named for the winner. - Operator failures inside a frame become ordinary States with a reserved __error__ key, broadcast in-band; per-port drain contagion in both workers discards post-failure data without invoking executors and suppresses finish hooks while ports still complete, so the stream terminates. - Failures outside any frame keep the existing behavior (console error + pause, current input retriable): TryCatchFramePass bakes a per-operator guarded flag delivered via InitializeExecutorRequest. - TryCatchFramePass pairs frames, computes try/catch cones, synthesizes signal edges (SignalPartitioning drops tuples at the sender) with a dependee snapshot port for structural resolution timing, and validates the wiring rules (disjoint cones, Finally provenance, no Merger bypass, Error Info cannot feed its own try cone). - Fix multi-dependee port scheduling (sliding(2,1) fabricated dependency pairs) and guard IfOpExec against unknown States. - pyamber parity: error State emission, drain guards, signal partitioner, and a failed-cycle handshake fix in DataProcessor so a failing Python UDF terminates cleanly. - Docs pages for both operators; unit + integration test coverage.
Rows never cross the Merger's result ports (try rows leave through Try Result, catch rows through Catch Result), so requiring the two branches to produce the same schema was unnecessarily restrictive — it also made every downstream operator report "schema is not available" whenever the branches differed. Each result port now adopts its own branch's schema; wiring both ports into one downstream input remains a Union, which enforces compatibility itself.
A failing Python worker emitted its error State only through the network partitioners — but a try-cone tail's outgoing edges (a Finally's dependee From Try, a catch gate's dependee signal ports) are materialized and have no live partitioners, so the failure signal reached nobody: the gate never released the snapshot and the catch branch ran empty. Mirror the Scala worker's emitState, which always pairs the network emit with a write to the output port's state storage that the materialization readers replay.
Two fixes for nested frames without their own Finally: - A nested frame's terminal catch leaf qualifies as both an enclosing-frame-owned tail and an escalation tap; wiring both signal edges materialized the same source port twice (signal ports are dependees), racing to create one storage table. The gate signal list is now deduplicated. e2e added: two sibling Finally-less inner frames recover independently inside an outer frame. - A frame without a Finally is terminal: if its subgraph flows into the Finally of a frame not nested inside it (Try1 -> Try2 -> Finally-of-1 with Try2 unclosed), the inner frame never closes -- its unbounded cone swallows the outer Merger and its gate goes blind to failures. Now rejected at compile time: Finallys close inside-out.
Cover the PL shape `try1 {} catch1 { try2 {} catch2 {} } finally1`: the
outer recovery is itself guarded, the inner construct closes with its own
Finally, and both result ports union into the outer From Catch. With both
attempts failing, the inner recovery becomes the outer construct's value.
github-actions
Bot
added
engine
pyamber
frontend
Contributor
Automated Reviewer SuggestionsBased on the
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #7388 +/- ##
============================================
- Coverage 84.81% 84.69% -0.12%
- Complexity 4149 4211 +62
============================================
Files 1169 1179 +10
Lines 46740 47142 +402
Branches 5202 5228 +26
============================================
+ Hits 39643 39929 +286
- Misses 5384 5489 +105
- Partials 1713 1724 +11
*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 | 378 | 0.231 | 26,085/34,170/34,170 us | 🔴 +20.0% / 🔴 +111.4% |
| 🔴 | bs=100 sw=10 sl=64 | 792 | 0.483 | 125,884/143,313/143,313 us | 🔴 +9.0% / 🔴 +29.3% |
| ⚪ | bs=1000 sw=10 sl=64 | 906 | 0.553 | 1,096,180/1,187,564/1,187,564 us | ⚪ within ±5% / 🔴 +12.1% |
Baseline details
Latest main 2c57707 from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 378 tuples/sec | 418 tuples/sec | 755.31 tuples/sec | -9.6% | -50.0% |
| bs=10 sw=10 sl=64 | MB/s | 0.231 MB/s | 0.255 MB/s | 0.461 MB/s | -9.4% | -49.9% |
| bs=10 sw=10 sl=64 | p50 | 26,085 us | 21,732 us | 12,952 us | +20.0% | +101.4% |
| bs=10 sw=10 sl=64 | p95 | 34,170 us | 41,781 us | 16,161 us | -18.2% | +111.4% |
| bs=10 sw=10 sl=64 | p99 | 34,170 us | 41,781 us | 19,292 us | -18.2% | +77.1% |
| bs=100 sw=10 sl=64 | throughput | 792 tuples/sec | 834 tuples/sec | 957.58 tuples/sec | -5.0% | -17.3% |
| bs=100 sw=10 sl=64 | MB/s | 0.483 MB/s | 0.509 MB/s | 0.584 MB/s | -5.1% | -17.4% |
| bs=100 sw=10 sl=64 | p50 | 125,884 us | 120,735 us | 104,473 us | +4.3% | +20.5% |
| bs=100 sw=10 sl=64 | p95 | 143,313 us | 131,447 us | 110,867 us | +9.0% | +29.3% |
| bs=100 sw=10 sl=64 | p99 | 143,313 us | 131,447 us | 120,336 us | +9.0% | +19.1% |
| bs=1000 sw=10 sl=64 | throughput | 906 tuples/sec | 919 tuples/sec | 986.93 tuples/sec | -1.4% | -8.2% |
| bs=1000 sw=10 sl=64 | MB/s | 0.553 MB/s | 0.561 MB/s | 0.602 MB/s | -1.4% | -8.2% |
| bs=1000 sw=10 sl=64 | p50 | 1,096,180 us | 1,084,784 us | 1,021,008 us | +1.1% | +7.4% |
| bs=1000 sw=10 sl=64 | p95 | 1,187,564 us | 1,182,929 us | 1,059,187 us | +0.4% | +12.1% |
| bs=1000 sw=10 sl=64 | p99 | 1,187,564 us | 1,182,929 us | 1,093,309 us | +0.4% | +8.6% |
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,529.14,200,128000,378,0.231,26084.98,34170.13,34170.13
1,100,10,64,20,2525.59,2000,1280000,792,0.483,125883.62,143312.55,143312.55
2,1000,10,64,20,22064.07,20000,12800000,906,0.553,1096179.86,1187564.29,1187564.29The guard added for try/catch frames skipped routing for ANY state missing the configured conditionName, which silently dropped the existing contract that a missing key throws NoSuchElementException -- the safeguard against a typo'd conditionName quietly misrouting every tuple to the default port (pinned by IfOpExecSpec, and caught by CI). Frames only need error States to pass through, so exempt exactly those: a failure traveling downstream as a dataflow event must not become a second failure inside an If, while any other state missing conditionName still fails loudly. Adds a test pinning that an error State is forwarded without disturbing the branch already chosen.
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
2 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?
This PR adds block-level try/catch/finally control blocks: a
TryCatchoperator that guards a subgraph and replays its input through a fallback
subgraph on failure, and a
Finallyoperator that reconverges the twobranches, releasing exactly one branch's complete output through the port
named for the winner (
Try Result/Catch Result).Before: any operator failure (e.g. a Python UDF raising) ends the run —
the error is reported and the workflow pauses, with no recovery path.
After: a failure inside a TryCatch frame aborts that attempt cleanly (no
user code runs on post-failure data), the catch branch runs on the same
input, and downstream sees exactly one branch's results — plus an
Error Info table (one deduplicated row per caught failure) usable for
auditing or
catch (SpecificError)-style routing. Failures outside anyframe keep the existing report-and-pause behavior exactly.
The design principle is the
Ifpattern, all the way down — no newcoordinator state machines, no new State/message types, no scheduler changes:
ordinary
Statewith a reserved__error__key (theIf-conditionconvention) and drains; ports still complete so the stream terminates.
Per-port drain contagion in both the Scala and Python workers delivers the
error State to the executor first (default pass-through = escalation to the
enclosing frame), then poisons the port: later data is discarded without
invoking the executor and finish hooks are suppressed.
guardedflag.TryCatchFramePassmarksevery operator inside a frame's cones (plus the frame apparatus);
InitializeExecutorRequest.guardeddelivers it per worker. Guardedfailure → error State + drain; unguarded failure → the existing console +
pause path (current input retriable). Plans without frames are untouched.
Ifgeneralized to Nconditions). The pass synthesizes signal edges from every try-cone tail to
the gate; the new
SignalPartitioningdrops tuples at the sender so thoseedges carry only States and end-of-stream. The gate's snapshot port depends
on all signal ports, making resolution timing structural (two-phase region
execution) rather than timing-dependent.
port-targeted emission;
From Catchdepends onFrom Try, so the releasedecision is deterministic. Each result port carries its own branch's schema
(rows never cross ports, so the branches need not agree); when they do
agree, unioning the two ports recovers "the winner, whichever it was".
Finally input provenance, catch-port connectivity, no reaching the
post-frame region around the Finally (Merger bypass), Error Info never
feeding its own frame's try cone, and Finallys close inside-out — a
frame without a Finally is terminal (branches end in their own sinks); if
its subgraph flows into an enclosing frame's Finally, compilation rejects
it with a message telling the user to close the inner frame first. Nesting
forms a tree; the innermost frame owns a failure; double failures escalate
to the enclosing frame; a nested frame's terminal catch leaf is signaled
exactly once (deduplicated between its owned-tail and escalation-tap
roles, since a doubled dependee edge would materialize the same port
twice).
Also fixed in passing (pre-existing engine issues surfaced by the feature):
ExpansionGreedyScheduleGeneratorfabricated dependency pairs for portswith more than one dependee (
sliding(2,1)); it now walks real dependencyedges (
PhysicalOp.getInputPortDependencyEdges).IfOpExeccrashed on States that do not carry its condition key; it nowignores unknown States (loop envelopes, error States).
thread one context switch out of sync with MainLoop; the cycle is now
finished before the final switch, matching the normal path's ordering.
port's state storage (mirroring the Scala worker's
emitState): atry-cone tail's outgoing edges are materialized and have no live
partitioners, so storage is the only path the failure signal can travel.
(1–8 = execution order. Green = allowed external wiring; red ✕ = rejected at
compile time; dotted = synthesized signal edges; dashed = materialized
snapshot; ↔ = Error Info and catch cone may interconnect.)
Any related issues, documentation, discussions?
docs/reference/operators/control-block/try-catch.mdanddocs/reference/operators/control-block/finally.md.How was this PR tested?
New unit suites (all green):
TryCatchFramePassSpec(14): frame pairing, cone computation, signal-edgesynthesis (single tail / forked cone / nested-catch-leaf deduplication),
per-link partitioning (the user's data link into Finally is not
signal-partitioned), and every wiring rule — cross-cone rejection, Finally
provenance, Merger-bypass rejection, inside-out-Finally rejection,
Error-Info-into-try-cone rejection, Error-Info-to-catch/downstream
acceptance, external upstreams joining cones, guarded-flag marking.
TryCatchOpDescSpec/CatchGateOpExecSpec/FinallyMergerOpExecSpec(23): port declarations, schema propagation, gate release/drop/attribution
and Error Info dedup, merger winner routing by port and field-level output
integrity.
DataProcessorSpec(10): guarded failure → error State + per-port drain +finish-hook suppression (state, tuple, and output-iterator paths);
unguarded failure → pause, no error State (existing behavior pinned);
received-error poisoning is per-port.
PhysicalOpSpec/PartitionInfoSpec: multi-dependee edges,SignalPartitionregistry/JSON round-trip.pytest, 368 in runnables/architecture, full suite 1038):drain guards, error-State emission ordering (console RPC before error
State), guarded/unguarded failed-cycle handshake, LoopEnd failure guards,
end-channel completion after failure.
New end-to-end suite
TryCatchIntegrationSpec(real engine, materializedresults):
Try Result, catch branch stays empty;Catch Result, never a mix;catch (SpecificError): Error Info → classifier UDF → State → If routesthe replay to the matching handler;
Python worker's drain/error-State path end to end, including the state
write to materialized port storage);
outer frame (terminal branches; recovery invisible to the outer frame);
try1 {} catch1 { try2 {} catch2 {} } finally1— with both attemptsfailing, the inner recovery becomes the outer construct's value.
All eleven cases pass.
Generated-by: CoAuthored by Codex 5.6 Sol Ultra, Fable 5 UltraCode and Me