This is an automated email from the ASF dual-hosted git repository.
Tartarus0zm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/auron.git
The following commit(s) were added to refs/heads/master by this push:
new c1e59433 [AURON #2329] Test and document multi-consumer source-Calc
fan-out safety (#2415)
c1e59433 is described below
commit c1e5943327482fe7190febfbfba78cf9d23bce28
Author: Weiqing Yang <[email protected]>
AuthorDate: Mon Jul 27 05:48:46 2026 -0700
[AURON #2329] Test and document multi-consumer source-Calc fan-out safety
(#2415)
# 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 example `UNION ALL` over 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 `AuronColumnarRowData`
reference 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
`testSharedSourceUnionAllFanOutSafeWithObjectReuse` in
`AuronKafkaSourceMergeITCase`: the same shared-source `UNION ALL` as 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
`AuronOperatorFusionProcessor` explaining 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=AuronKafkaSourceMergeITCase` with the native library built. All 4
tests pass.
# Was this patch authored or co-authored using generative AI tooling?
- [x] Yes （Claude Code (Opus 4.8)）
- [ ] No
---
.../processor/AuronOperatorFusionProcessor.java | 8 +++-
.../table/kafka/AuronKafkaSourceMergeITCase.java | 43 ++++++++++++++++++++++
2 files changed, 50 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 fcd7d972..1f9beadb 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 @@ public class AuronOperatorFusionProcessor implements
ExecNodeGraphProcessor {
Map<Integer, Integer> consumerCount,
Function<StreamExecTableSourceScan, DynamicTableSource>
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 89114a9f..36724665 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 class AuronKafkaSourceMergeITCase extends
AuronKafkaSourceTestBase {
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 <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 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.
+ *
+ * <p>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<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 {
+ if (prevObjectReuse) {
+ environment.getConfig().enableObjectReuse();
+ } else {
+ environment.getConfig().disableObjectReuse();
+ }
+ }
+ }
}
