From 787f4102893e294359466bf3dc39ebf44d4d9eff Mon Sep 17 00:00:00 2001 From: weiqingy Date: Tue, 21 Jul 2026 21:02:16 -0700 Subject: [PATCH 1/4] [AURON #2329] Test and document multi-consumer source-Calc fan-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. --- .../AuronOperatorFusionProcessor.java | 7 +++- .../kafka/AuronKafkaSourceMergeITCase.java | 37 +++++++++++++++++++ 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/processor/AuronOperatorFusionProcessor.java b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/processor/AuronOperatorFusionProcessor.java index fcd7d9720..9fde84857 100644 --- a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/processor/AuronOperatorFusionProcessor.java +++ b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/processor/AuronOperatorFusionProcessor.java @@ -197,7 +197,12 @@ void tryFuse( Map consumerCount, Function sourceResolver, ReadableConfig tableConfig) { - // Sole-consumer gate: multi-consumer fusion is out of scope (tracked separately). + // Sole-consumer gate: a source feeding more than one consumer is deliberately not fused. + // The native source runtime is single-plan / single-output — one source runs exactly one + // PhysicalPlanNode and emits one stream, so N distinct per-consumer Calc plans cannot share + // 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. if (consumerCount.getOrDefault(scan.getId(), 0) != 1) { return; } diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java index 89114a9ff..739c38c76 100644 --- a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java +++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java @@ -140,4 +140,41 @@ public void testSharedSourceUnionAllDoesNotFuseUnderDefaultReuse() { 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))); } + + /** + * The same shared-source {@code UNION ALL} as {@link + * #testSharedSourceUnionAllDoesNotFuseUnderDefaultReuse}, but with object reuse enabled. Under + * object reuse Flink hands the same {@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. + * + *

The gate assertion still holds under object reuse: a multi-consumer source is never fused, + * so both Calcs remain standalone operators and {@code calcOperatorCount} is 2. + */ + @Test + public void testSharedSourceUnionAllFanOutSafeWithObjectReuse() { + 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 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(); + } + } } From f525cc00775486f228f56111acbdc6728c83dc68 Mon Sep 17 00:00:00 2001 From: weiqingy Date: Tue, 21 Jul 2026 22:03:57 -0700 Subject: [PATCH 2/4] [AURON #2329] Reword sole-consumer-gate comment for the codegen-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. --- .../planner/processor/AuronOperatorFusionProcessor.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/processor/AuronOperatorFusionProcessor.java b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/processor/AuronOperatorFusionProcessor.java index 9fde84857..1f9beadbc 100644 --- a/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/processor/AuronOperatorFusionProcessor.java +++ b/auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/processor/AuronOperatorFusionProcessor.java @@ -200,9 +200,10 @@ void tryFuse( // Sole-consumer gate: a source feeding more than one consumer is deliberately not fused. // The native source runtime is single-plan / single-output — one source runs exactly one // PhysicalPlanNode and emits one stream, so N distinct per-consumer Calc plans cannot share - // 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. + // a single source. Declining loses nothing structurally: each consumer instead translates + // on its own over the source's shared row stream (a standalone native Calc when the Calc + // converts, otherwise Flink's codegen Calc), so no per-consumer plan is ever staged onto the + // shared source and there is no last-write-wins between consumers. if (consumerCount.getOrDefault(scan.getId(), 0) != 1) { return; } From c12805bb76b97e3aa44f0fa60d01008f926fdca5 Mon Sep 17 00:00:00 2001 From: weiqingy Date: Tue, 21 Jul 2026 22:17:37 -0700 Subject: [PATCH 3/4] [AURON #2329] Frame fan-out safety comment as the native Calc path 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. --- .../table/kafka/AuronKafkaSourceMergeITCase.java | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java index 739c38c76..be1a54b66 100644 --- a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java +++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java @@ -146,12 +146,13 @@ public void testSharedSourceUnionAllDoesNotFuseUnderDefaultReuse() { * #testSharedSourceUnionAllDoesNotFuseUnderDefaultReuse}, but with object reuse enabled. Under * object reuse Flink hands the same {@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. + * 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, + * the row set would collapse or corrupt (all rows folding onto the last row of a batch) or fault + * on freed off-heap buffers, so the row-set assertion below is what pins that native path safe. * *

The gate assertion still holds under object reuse: a multi-consumer source is never fused, * so both Calcs remain standalone operators and {@code calcOperatorCount} is 2. From fdd685b1e656cff9b032f1d86bd932a8cd3be4cc Mon Sep 17 00:00:00 2001 From: weiqingy Date: Tue, 21 Jul 2026 22:29:50 -0700 Subject: [PATCH 4/4] [AURON #2329] Save and restore object reuse in the fan-out test 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. --- .../flink/table/kafka/AuronKafkaSourceMergeITCase.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java index be1a54b66..36724665b 100644 --- a/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java +++ b/auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java @@ -160,6 +160,7 @@ public void testSharedSourceUnionAllDoesNotFuseUnderDefaultReuse() { @Test public void testSharedSourceUnionAllFanOutSafeWithObjectReuse() { environment.setParallelism(1); + boolean prevObjectReuse = environment.getConfig().isObjectReuseEnabled(); environment.getConfig().enableObjectReuse(); try { String sql = @@ -175,7 +176,11 @@ public void testSharedSourceUnionAllFanOutSafeWithObjectReuse() { 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(); + if (prevObjectReuse) { + environment.getConfig().enableObjectReuse(); + } else { + environment.getConfig().disableObjectReuse(); + } } } }