carloea2 opened a new pull request, #7388:
URL: https://github.com/apache/texera/pull/7388
<!--
Pull-request description draft ??? follows .github/PULL_REQUEST_TEMPLATE.
Suggested PR title (Conventional Commits):
feat(engine): add TryCatch/Finally control blocks with in-band error
handling
Attach trycatch_unified_diagram.svg where the diagram is referenced.
-->
### What changes were proposed in this PR?
This PR adds block-level **try/catch/finally** control blocks: a `TryCatch`
operator that guards a subgraph and replays its input through a fallback
subgraph on failure, and a `Finally` operator that reconverges the two
branches, 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 any
frame keep the existing report-and-pause behavior exactly.
The design principle is *the `If` pattern, all the way down* ??? no new
coordinator state machines, no new State/message types, no scheduler changes:
- **Failure becomes a dataflow event.** A failing worker broadcasts an
ordinary `State` with a reserved `__error__` key (the `If`-condition
convention) 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.
- **Opt-in via a compile-time `guarded` flag.** `TryCatchFramePass` marks
every operator inside a frame's cones (plus the frame apparatus);
`InitializeExecutorRequest.guarded` delivers it per worker. Guarded
failure ??? error State + drain; unguarded failure ??? the existing
console +
pause path (current input retriable). Plans without frames are untouched.
- **TryCatch expands to splitter + catch gate** (an `If` generalized to N
conditions). The pass synthesizes signal edges from every try-cone tail to
the gate; the new `SignalPartitioning` drops tuples at the sender so those
edges 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.
- **Finally stages both sides and flushes the winner** at finish time via
port-targeted emission; `From Catch` depends on `From Try`, so the release
decision 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".
- **Compile-time validation with clear messages**: disjoint try/catch cones,
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):
- `ExpansionGreedyScheduleGenerator` fabricated dependency pairs for ports
with more than one dependee (`sliding(2,1)`); it now walks real dependency
edges (`PhysicalOp.getInputPortDependencyEdges`).
- `IfOpExec` crashed on States that do not carry its condition key; it now
ignores unknown States (loop envelopes, error States).
- The Python worker's failed-cycle handshake could leave the DataProcessor
thread one context switch out of sync with MainLoop; the cycle is now
finished before the final switch, matching the normal path's ordering.
- A failing Python worker's error State is also written to its output
port's state storage (mirroring the Scala worker's `emitState`): a
try-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?
- Feature request: #7387 (fill in after opening the issue ???
`TRY_CATCH_ISSUE.md`).
- User-facing documentation is included in this PR:
`docs/reference/operators/control-block/try-catch.md` and
`docs/reference/operators/control-block/finally.md`.
### How was this PR tested?
New unit suites (all green):
- `TryCatchFramePassSpec` (14): frame pairing, cone computation, signal-edge
synthesis (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,
`SignalPartition` registry/JSON round-trip.
- Python worker (`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, materialized
results):
1. success ??? try results out `Try Result`, catch branch stays empty;
2. failure ??? catch results out `Catch Result`, never a mix;
3. both result ports wired downstream ??? loser subgraph completes empty;
4. frame with unconnected Catch completes on success;
5. guarded failure with no catch wired ??? drains and terminates (no hang);
6. nested: inner frame recovers, outer frame undisturbed;
7. nested: double inner failure escalates to the outer catch;
8. `catch (SpecificError)`: Error Info ??? classifier UDF ??? State ??? If
routes
the replay to the matching handler;
9. a failing Python UDF falls back to the catch branch (exercises the
Python worker's drain/error-State path end to end, including the state
write to materialized port storage);
10. two sibling Finally-less inner frames recover independently inside an
outer frame (terminal branches; recovery invisible to the outer frame);
11. an inner frame inside the CATCH branch ???
`try1 {} catch1 { try2 {} catch2 {} } finally1` ??? with both attempts
failing, the inner recovery becomes the outer construct's value.
All eleven cases pass.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: CoAuthored by Codex 5.6 Sol Ultra, Fable 5 UltraCode and Me
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
