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..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 @@ -197,7 +197,13 @@ 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 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; } 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..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 @@ -140,4 +140,47 @@ 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 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. + */ + @Test + public void testSharedSourceUnionAllFanOutSafeWithObjectReuse() { + environment.setParallelism(1); + boolean prevObjectReuse = environment.getConfig().isObjectReuseEnabled(); + 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 { + if (prevObjectReuse) { + environment.getConfig().enableObjectReuse(); + } else { + environment.getConfig().disableObjectReuse(); + } + } + } }