[AURON #2329] Test and document multi-consumer source-Calc fan-out safety#2415
[AURON #2329] Test and document multi-consumer source-Calc fan-out safety#2415weiqingy wants to merge 4 commits into
Conversation
…out safety A source feeding multiple Calcs (UNION ALL over one table under source reuse) declines fusion at the graph-level sole-consumer gate and runs each Calc as a standalone native operator over the shared source stream. Add an end-to-end test that keeps this correct with Flink object reuse enabled, where the source hands one AuronColumnarRowData reference to both consumers with no defensive copy. Document the invariant at the sole-consumer gate: the native source runtime is single-plan and single-output, so multiple per-consumer plans cannot share one source; declining fusion is safe because each consumer copies out of the shared columnar view.
There was a problem hiding this comment.
Pull request overview
Adds coverage and documentation for the “shared source → multiple Calc consumers” case, specifically validating that graph-level fusion correctly declines multi-consumer sources and that the fan-out remains correct under Flink object reuse.
Changes:
- Adds a new end-to-end IT case that runs the shared-source
UNION ALLscenario with object reuse enabled and asserts both correctness and “no fusion” operator topology. - Expands the sole-consumer gate comment in
AuronOperatorFusionProcessorto explain why multi-consumer sources are not fused and why the unfused path remains safe.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java | Adds an object-reuse-enabled shared-source UNION ALL fan-out safety IT case and validates calc operator count + row set. |
| auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/processor/AuronOperatorFusionProcessor.java | Improves documentation at the sole-consumer gate to clarify multi-consumer fusion constraints and safety rationale. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| // a single source. Declining loses nothing structurally: each consumer instead runs as a | ||
| // standalone native Calc over the source's shared row stream, copying every column out of | ||
| // the shared columnar RowData view, so the fan-out is safe. |
There was a problem hiding this comment.
Good catch. The unfused Calc can indeed run as Flink codegen. StreamExecCalc.translateToPlanInternal falls back to super when the Calc does not convert to native, so asserting native-copy semantics unconditionally was wrong. Reworded to make the structural point the gate actually protects: no per-consumer plan is staged onto the shared source, so there is no last-write-wins between consumers, whether each consumer ends up native or codegen. Fixed in f525cc0.
…-fallback case An unfused multi-consumer Calc runs as a standalone native Calc only when it converts; otherwise StreamExecCalc falls back to Flink's codegen Calc. Drop the unconditional native-copy claim and state the structural point the gate protects: no per-consumer plan is staged onto the shared source, so there is no last-write-wins between consumers.
| /** | ||
| * The same shared-source {@code UNION ALL} as {@link | ||
| * #testSharedSourceUnionAllDoesNotFuseUnderDefaultReuse}, but with object reuse enabled. Under | ||
| * object reuse Flink hands the <em>same</em> {@code AuronColumnarRowData} reference to both | ||
| * standalone Calc consumers with no defensive copy in between (the sibling test runs with reuse | ||
| * off, where Flink deep-copies the row to heap per consumer). The fan-out still produces the | ||
| * correct row set because each Calc eagerly copies every column out of the shared columnar view | ||
| * into its own Arrow batch before returning, so neither consumer retains a reference to the | ||
| * reused row. If the consumers instead aliased the shared row id or read the batch after it was | ||
| * recycled, the row set would collapse or corrupt (all rows folding onto the last row of a | ||
| * batch) or fault on freed off-heap buffers. |
There was a problem hiding this comment.
Good catch. You're right that the assertions pin non-fusion (count == 2) and the row set, not that the standalone Calcs run through the native path.
Reworded in c12805b: the Arrow-batch copy is now framed as the native Auron Calc behavior the projections and filters convert to, and the comment names the row-set assertion as the empirical guarantee for that path instead of stating the copy as an unconditional fact.
The object-reuse test comment described the Arrow-batch copy as unconditional, but the assertions pin non-fusion and the row set, not that the standalone Calcs run natively. Tie the copy mechanism to the native Auron Calc path the projections/filters convert to, and name the row-set assertion as the empirical guarantee.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.
Comments suppressed due to low confidence (1)
auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java:153
- The Javadoc explains the safety guarantee in terms of the Calcs converting to a standalone native Auron Calc and copying columns into an Arrow batch, but the test assertions only pin (1) non-fusion (2 standalone Calcs) and (2) the produced row set. To avoid making the comment incorrect if the Calc translation ever falls back to Flink codegen, reword this section to describe the aliasing risk/row-set guarantee independent of whether the Calc runs native or codegen, and make the native-copy detail explicitly conditional.
* off, where Flink deep-copies the row to heap per consumer). The row set is the empirical
* guarantee here: the projections and filters convert, so each consumer runs as a standalone
* native Auron Calc that eagerly copies every column out of the shared columnar view into its
* own Arrow batch before returning, and neither consumer retains a reference to the reused row.
* If a native Calc instead aliased the shared row id or read the batch after it was recycled,
| environment.setParallelism(1); | ||
| environment.getConfig().enableObjectReuse(); | ||
| try { | ||
| String sql = | ||
| "SELECT `age` FROM T5 WHERE `age` > 20 " + "UNION ALL SELECT `age` + 1 FROM T5 WHERE `age` > 10"; | ||
|
|
||
| assertThat(calcOperatorCount(sql)) | ||
| .as( | ||
| "a shared (multi-consumer) source must not fuse under object reuse; both Calcs must remain operators") | ||
| .isEqualTo(2); | ||
|
|
||
| List<Row> rows = CollectionUtil.iteratorToList( | ||
| tableEnvironment.executeSql(sql).collect()); | ||
| rows.sort(Comparator.comparingInt(o -> (int) o.getField(0))); | ||
| assertThat(rows).isEqualTo(Arrays.asList(Row.of(21), Row.of(21), Row.of(22), Row.of(22), Row.of(23))); | ||
| } finally { | ||
| environment.getConfig().disableObjectReuse(); | ||
| } |
There was a problem hiding this comment.
Good point. The environment is shared across the class (PER_CLASS, created once in @BeforeAll), so an unconditional disable would clobber the setting if it were ever enabled globally.
Fixed in fdd685b: capture isObjectReuseEnabled() before enabling and restore it in the finally block, so the test stays order-independent.
The environment is shared across the class (PER_CLASS, created once in @BeforeAll), so forcing object reuse off in the finally block would clobber the prior setting if it were ever enabled globally. Capture the setting before enabling and restore it, keeping the test order-independent.
|
Hi @Tartarus0zm, could you please help review this PR when you get a chance? Thanks! |
Which issue does this PR close?
Closes #2329
Rationale for this change
#1865 moved Calc-source fusion to a graph-level pass (
AuronOperatorFusionProcessor) that counts each source's consumers and fuses only sole-consumer sources. A source feeding more than one Calc (for exampleUNION ALLover one table under Flink's default source reuse) is declined at the sole-consumer gate, and each Calc runs as a standalone native operator over the shared source stream.That multi-consumer path is correct but was under-tested and under-documented. The existing shared-source test only covered the default config, where object reuse is effectively off so Flink deep-copies each row to heap per consumer. The interesting case is object reuse ON: Flink then hands the same
AuronColumnarRowDatareference to both Calc consumers with no defensive copy in between, which is exactly where a fan-out aliasing or use-after-free bug would surface. Nothing pinned that. And the reason a multi-consumer source declines fusion was only a one-line comment.What changes are included in this PR?
A new end-to-end test
testSharedSourceUnionAllFanOutSafeWithObjectReuseinAuronKafkaSourceMergeITCase: the same shared-sourceUNION ALLas the existing test, but with object reuse enabled. It asserts the row set stays correct and both Calcs remain standalone operators (the multi-consumer source is not fused).An expanded doc-comment at the sole-consumer gate in
AuronOperatorFusionProcessorexplaining why a multi-consumer source is not fused (the native source runtime is single-plan and single-output, so N per-consumer plans cannot share one source) and why declining is safe (each consumer runs a standalone native Calc that copies every column out of the shared columnar view).No production logic changes.
Are there any user-facing changes?
No.
How was this patch tested?
./build/mvn test -Pspark-3.5 -Pscala-2.12 -Pflink-1.18 -pl auron-flink-extension/auron-flink-planner -am -Dtest=AuronKafkaSourceMergeITCasewith the native library built. All 4 tests pass.Was this patch authored or co-authored using generative AI tooling?