Repository navigation
[Flink] Graph-level Calc-source fusion: multi-consumer-safe and wrapper-agnostic #2329
Description
#1865 fuses a convertible Calc into the native Kafka source using detection inside the shadowed StreamExecCalc.translateToPlanInternal.
Two limitations follow from doing detection at the Calc node:
-
Multi-consumer safety. A translating
ExecNodecannot see how many operators read its source (Flink's exec-node graph only points from consumer to producer), and Flink source reuse is on by default. When one source feeds more than one Calc, fusing mutates the shared source and produces wrong results. Support merge native operators #1865 therefore conservatively fuses only when source/sub-plan reuse is disabled, which makes fusion effectively opt-in. -
Wrapper-agnostic detection. The fuse target is found via
instanceof LegacySourceTransformationplus unwrappinggetOperator().getUserFunction(). Any future wrapper between source and Calc (a Source-API migration, a partitioner/rebalance, a second native source type) silently fails the check and falls back with no signal.
Both are addressed by moving fusion to a graph-level pass (an ExecNodeGraphProcessor) that has the full ExecNodeGraph: it can count each source's consumers and fuse only sole-consumer sources (so fusion can be on by default regardless of the reuse config), and detect native sources structurally instead of by a Kafka-specific instanceof.
This is the graph-level whole-stage-merge groundwork for the Flink integration; the later operator phases in #1264 (agg, join) build on the same mechanism. Refs #1865, #1264.
Activity
weiqingy commentedon Jun 14, 2026
Design input from the #2327 review (discussion on
StreamExecCalc:177) worth carrying into this work:Layering principle. The motivation for moving fusion out of
StreamExecCalc: aStreamExecshould translate only its own node, not reach across to mutate its upstream. The graph-level pass is what removes that cross-operator reach entirely.Mechanism candidates (rough order of cleanliness vs. cost):
ExecNodeGraph(ExecNodeGraphProcessor, as above) — fuses before translation.StreamGraphTranslatorand merging during JobGraph generation, as Gluten does — operates lower, on the Transformation DAG.Whichever we pick needs the full graph, so it can count each source's consumers (multi-consumer safety) and detect native sources structurally.
Scope update from #2327. Detection has since moved off the concrete Kafka source onto the
FlinkAuronFunctionmarker (commit99403ef2), so item 2's remaining Kafka-specific piece is narrower than written above: only theLegacySourceTransformationunwrap is still wrapper-specific. The structural detection here would replace that unwrap.The open question deferred in #2327 is sequencing — land the interim (#1865's conservative fusion) first, or go straight to the graph pass.
weiqingy commentedon Jun 19, 2026
The reworked #1865 adds the graph-level fusion mechanism (an
ExecNodeGraphProcessorover theExecNodeGraph) together with the single-consumersource → Calccase, correct under default reuse config.This issue builds on that mechanism for the multi-consumer case: a shared source feeding several Calcs (e.g.
UNION ALLover one table under source reuse), where all consumers should fuse without the last-write-wins corruption a node-level merge would hit. The processor already counts consumers and declines to fuse when the count is greater than one, so the remaining work is the safe shared-source rewrite.weiqingy commentedon Jul 22, 2026
Scoping update after digging into the native execution model.
The graph-level move and the consumer-count gate landed in #1865, which also made the multi-consumer case correct: a shared source (consumer count > 1) declines fusion at the sole-consumer gate and each Calc runs as a standalone native operator over the shared source stream.
The "all consumers should fuse" idea from the earlier comment turns out not to be reachable as written. The native source runtime is single-plan and single-output (one source runs exactly one PhysicalPlanNode and emits one stream), so N distinct per-consumer Calc plans cannot share one source. Duplicating the source per consumer is worse (each copy opens its own consumer, splitting partitions or re-reading the topic). So declining fusion and running standalone native Calcs is the correct and safe shape for the multi-consumer case, not a fallback.
Given that, the remaining gaps were narrow: the multi-consumer path was correct but the object-reuse fan-out was untested (the existing test only covered reuse-off, where Flink deep-copies rows to heap), and the reason for declining was undocumented at the gate. The PR below closes both.
There is one real inefficiency left that is out of scope here: when a shared source fans out to N standalone Calcs, each Calc re-materializes every row into its own Arrow batch, so the row round-trip is paid N times. Fixing that needs a batch-oriented operator exchange rather than a per-row RowData stream, which is a separate change. I will file a follow-up issue for it.
[AURON apache#2329] Reword sole-consumer-gate comment for the codegen…
[AURON apache#2329] Frame fan-out safety comment as the native Calc path
[AURON apache#2329] Save and restore object reuse in the fan-out test
Tartarus0zm commentedon Jul 27, 2026
@weiqingy Do you think this issue still needs to stay open? The latest PR doesn't seem to be closely related to the issue's main topic.
weiqingy commentedon Jul 27, 2026
Agreed, closing is right. Thanks for merging. The title is where the issue started, and the scope moved under it without the title being updated.
Both original items landed in #1865 (merged as #2327): fusion moved to the graph-level
ExecNodeGraphProcessorthat counts consumers and fuses only sole-consumer sources, and detection moved off theLegacySourceTransformationunwrap onto theFlinkAuronDynamicTableSourcemarker, so it is no longer tied to one connector.That left only the narrow gaps this PR closed: the object-reuse fan-out was untested, and the reason a multi-consumer source declines fusion was a one-line comment. Hence test plus doc rather than the headline in the title.
The remaining piece is the per-consumer row round-trip, tracked in #2416.
[AURON apache#2329] Test and document multi-consumer source-Calc fan-…