You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@auron.apache.org by "weiqingy (via GitHub)" <gi...@apache.org> on 2026/06/11 04:07:15 UTC
[PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
weiqingy opened a new pull request, #2327:
URL: https://github.com/apache/auron/pull/2327
# Which issue does this PR close?
Closes #1865
# Rationale for this change
A `SELECT proj … FROM kafka WHERE pred` plan runs today as two independent native operators: the native Kafka source converts its columnar output to `RowData`, then the shadowed `StreamExecCalc` operator converts it back to Arrow, runs `Project[Filter]` natively, and converts back to `RowData` — paying multiple row/column conversions where one would do. This PR fuses the source and a convertible Calc into a single native plan `Project[Filter?[KafkaScan]]` that runs inside the source, so the data stays columnar end-to-end and converts to `RowData` once, at the chain tail. This is the whole-stage native operator merging direction for the Flink integration, mirroring the plan-merge already used on the Spark side.
# What changes are included in this PR?
Strict gating in the shadowed `StreamExecCalc`: a Calc converts to native only when it can fuse into a native upstream chain — its input is the native Auron Kafka source, the source has no event-time watermark, and every expression converts. When fusible, the Calc hands its logical `Project[Filter?]` sub-plan to the source and returns the source transformation re-typed to the projected output, eliminating the Calc operator. Every other case (non-native upstream, watermarked source, any unconvertible expression) runs as a stock Flink Calc.
The source splices its `KafkaScan` into the handed-off sub-plan and prepends the partition/offset/timestamp metadata columns, so its per-record offset-commit and timestamp bookkeeping are unchanged; native column resolution is by name, so no column-index rewriting is needed.
The standalone FFI-reader native-Calc path is no longer emitted from the Calc decision. Detection and hand-off go through the source function, so no source-operator type is introduced; operator-level native identity is left to the future ExecNode-graph merge that would consume it. Event-time-watermark fusion is out of scope and tracked as a follow-up.
This PR is stacked on #2325 (no-watermark source emit fix), which it requires; that commit drops out of the diff once #2325 merges and this branch is rebased.
# Are there any user-facing changes?
Performance: a `source → Calc` chain over the native Kafka source with no event-time watermark now runs as a single native pipeline with one columnar-to-row conversion. A Calc that cannot fuse (non-native source, watermarked source, or unsupported expression) runs as Flink's codegen Calc.
# How was this patch tested?
Unit tests in `StreamExecCalcTest` assert the fusion decision (native source + no watermark + convertible → merged source transformation with no separate Calc operator; each negative case → stock Flink Calc). An end-to-end `AuronKafkaSourceMergeITCase` runs fused SQL over a no-watermark `auron-kafka` table and asserts row-set correctness; a native boolean-to-string cast renders lowercase, proving the fused path executes natively (Flink codegen renders uppercase). The full `auron-flink-planner` (130) and `auron-flink-runtime` (137) module test suites pass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "Tartarus0zm (via GitHub)" <gi...@apache.org>.
Tartarus0zm merged PR #2327:
URL: https://github.com/apache/auron/pull/2327
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3592738895
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/processor/AuronOperatorFusionProcessor.java:
##########
@@ -0,0 +1,292 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.flink.table.planner.processor;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+import org.apache.auron.flink.connector.kafka.KafkaConstants;
+import org.apache.auron.flink.runtime.operator.FlinkAuronDynamicTableSource;
+import org.apache.auron.flink.table.planner.FlinkAuronCalcNode;
+import org.apache.auron.flink.table.planner.converter.NativePlanFusionBuilder;
+import org.apache.auron.protobuf.PhysicalPlanNode;
+import org.apache.calcite.rex.RexNode;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.planner.delegation.PlannerBase;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeGraph;
+import org.apache.flink.table.planner.plan.nodes.exec.processor.ExecNodeGraphProcessor;
+import org.apache.flink.table.planner.plan.nodes.exec.processor.ProcessorContext;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecTableSourceScan;
+import org.apache.flink.table.types.logical.RowType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Graph-level fusion pass that merges a native Auron source and the {@code StreamExecCalc} that
+ * directly consumes it into a single native {@code Project[Filter?[KafkaScan]]} plan, so the chain
+ * stays columnar end-to-end and pays a single columnar→row conversion at the tail.
+ *
+ * <p>The pass runs over the whole {@link ExecNodeGraph} before Transformation translation. For each
+ * {@code source → Calc} chain it decides whether to fuse and, when it does, stages the merged
+ * Calc sub-plan on the source's {@link FlinkAuronDynamicTableSource} instance (annotate-and-skip): it
+ * does not remove or re-type any node. The shadowed {@code StreamExecCalc} later observes the staged
+ * plan and emits no standalone operator for that Calc. This decision is graph-level because
+ * sole-consumership ("does this source feed exactly one consumer?") is a whole-graph property a single
+ * node cannot see in Flink's one-directional exec graph.
+ *
+ * <p>Fusion is additive on top of the standalone native Calc path: a convertible Calc the pass does
+ * not fuse still runs as a standalone native operator; only a non-convertible Calc falls back to
+ * Flink codegen.
+ *
+ * <p>A Calc is fused into its source only when all hold:
+ *
+ * <ul>
+ * <li>the Calc's single input is a {@link StreamExecTableSourceScan} whose {@link
+ * DynamicTableSource} is a {@link FlinkAuronDynamicTableSource};
+ * <li>that scan is the sole consumer of the source (consumer count {@code == 1} —
+ * multi-consumer fusion is tracked separately);
+ * <li>the source has no event-time watermark (fusing below a watermark generator would strip
+ * per-record event-time);
+ * <li>the Calc fully converts to a native plan;
+ * <li>no projection-output or source-input column name collides with a reserved Kafka metadata
+ * column (the native engine resolves columns by name, so a collision would silently bind to the
+ * metadata column).
+ * </ul>
+ */
+public class AuronOperatorFusionProcessor implements ExecNodeGraphProcessor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AuronOperatorFusionProcessor.class);
+
+ @Override
+ public ExecNodeGraph process(ExecNodeGraph graph, ProcessorContext context) {
+ final PlannerBase planner = context.getPlanner();
+ return process(
+ graph,
+ scan -> scan.getTableSourceSpec()
+ .getScanTableSource(planner.getFlinkContext(), planner.getTypeFactory()),
+ planner.getTableConfig());
+ }
+
+ /**
+ * Runs the fusion decision over the whole graph: builds the reverse fan-out consumer count,
+ * collects every {@code scan → Calc} chain, and applies {@link #tryFuse} to each. Split from
+ * the {@link ProcessorContext} entry point so the full count→gate→stage path is drivable
+ * with the scan's source and the table config supplied directly, without a live {@link
+ * PlannerBase}.
+ *
+ * @param graph the exec-node graph to process
+ * @param sourceResolver resolves a scan node to its {@link DynamicTableSource}
+ * @param tableConfig the table config used to seed the native plan converter
+ * @return the same graph instance (annotate-and-skip: only source instances are mutated)
+ */
+ ExecNodeGraph process(
+ ExecNodeGraph graph,
+ Function<StreamExecTableSourceScan, DynamicTableSource> sourceResolver,
+ ReadableConfig tableConfig) {
+ final Map<Integer, Integer> consumerCount = buildConsumerCount(graph);
+
+ for (Map.Entry<ExecNode<?>, ExecNode<?>> chain :
+ findSourceCalcChains(graph).entrySet()) {
+ final StreamExecTableSourceScan scan = (StreamExecTableSourceScan) chain.getKey();
+ final StreamExecCalc calc = (StreamExecCalc) chain.getValue();
+ tryFuse(scan, calc, consumerCount, sourceResolver, tableConfig);
+ }
+ // Annotate-and-skip: the graph is structurally unchanged; only source instances are mutated.
+ return graph;
+ }
+
+ /**
+ * Builds a reverse fan-out map keyed by stable {@link ExecNode#getId()}, counting consumers per
+ * distinct consuming edge with visited-node dedup across all roots. A node reachable from
+ * multiple roots is counted once per distinct consumer edge, not once per traversal path, so a
+ * genuinely shared source can never be miscounted as {@code count == 1}.
+ *
+ * @param graph the exec-node graph to traverse
+ * @return a map from node id to the number of distinct consumers of that node
+ */
+ static Map<Integer, Integer> buildConsumerCount(ExecNodeGraph graph) {
+ final Map<Integer, Integer> consumerCount = new HashMap<>();
+ final Set<Integer> visited = new HashSet<>();
+ // Each visited edge contributes exactly one consumer to its source. Dedup is on the visited
+ // SET of nodes (so each node's input edges are walked once across all roots); the consumer
+ // increment happens per input edge of every visited node, which is the per-consuming-edge
+ // count the design requires.
+ final Deque<ExecNode<?>> stack = new ArrayDeque<>(graph.getRootNodes());
+ while (!stack.isEmpty()) {
+ final ExecNode<?> node = stack.pop();
+ if (!visited.add(node.getId())) {
+ continue;
+ }
+ for (ExecEdge edge : node.getInputEdges()) {
+ final ExecNode<?> src = edge.getSource();
+ consumerCount.merge(src.getId(), 1, Integer::sum);
+ stack.push(src);
+ }
+ }
+ return consumerCount;
+ }
+
+ /**
+ * Collects every {@code scan → Calc} chain where a {@link StreamExecTableSourceScan} is the
+ * single input source of a {@link StreamExecCalc}. Keyed by the scan node so each scan maps to the
+ * Calc that directly consumes it.
+ *
+ * @param graph the exec-node graph to traverse
+ * @return a map from each consuming scan to its {@link StreamExecCalc}
+ */
+ private static Map<ExecNode<?>, ExecNode<?>> findSourceCalcChains(ExecNodeGraph graph) {
+ final Map<ExecNode<?>, ExecNode<?>> chains = new HashMap<>();
+ final Set<Integer> visited = new HashSet<>();
+ final Deque<ExecNode<?>> stack = new ArrayDeque<>(graph.getRootNodes());
+ while (!stack.isEmpty()) {
+ final ExecNode<?> node = stack.pop();
+ if (!visited.add(node.getId())) {
+ continue;
+ }
+ if (node instanceof StreamExecCalc && node.getInputEdges().size() == 1) {
+ final ExecNode<?> input = node.getInputEdges().get(0).getSource();
+ if (input instanceof StreamExecTableSourceScan) {
+ chains.put(input, node);
+ }
+ }
+ for (ExecEdge edge : node.getInputEdges()) {
+ stack.push(edge.getSource());
+ }
+ }
+ return chains;
+ }
+
+ /**
+ * Applies the full fusion gate to one {@code scan → Calc} chain and, when every gate passes,
+ * stages the merged Calc plan on the source instance. Any failed gate is a silent no-fusion: the
+ * Calc then runs standalone native (if convertible) or Flink codegen.
+ *
+ * @param scan the source scan node
+ * @param calc the Calc node directly consuming the scan
+ * @param consumerCount the reverse fan-out map from {@link #buildConsumerCount}
+ * @param sourceResolver resolves the scan to its {@link DynamicTableSource}
+ * @param tableConfig the table config used to seed the native plan converter
+ */
+ void tryFuse(
+ StreamExecTableSourceScan scan,
+ StreamExecCalc calc,
+ Map<Integer, Integer> consumerCount,
+ Function<StreamExecTableSourceScan, DynamicTableSource> sourceResolver,
+ ReadableConfig tableConfig) {
+ // Sole-consumer gate: multi-consumer fusion is out of scope (tracked separately).
+ if (consumerCount.getOrDefault(scan.getId(), 0) != 1) {
+ return;
+ }
+
+ final DynamicTableSource tableSource = sourceResolver.apply(scan);
+
+ final FlinkAuronCalcNode calcNode = (FlinkAuronCalcNode) calc;
+ stagePlanIfFusible(
+ tableSource,
+ scan.getId(),
+ (RowType) calc.getInputEdges().get(0).getOutputType(),
+ (RowType) calc.getOutputType(),
+ calcNode.getProjection(),
+ calcNode.getCondition(),
+ tableConfig);
Review Comment:
`tryFuse` unconditionally casts `StreamExecCalc` to `FlinkAuronCalcNode`. If the classpath ever resolves to Flink’s stock `StreamExecCalc` (e.g., shadowing/exclusion misconfiguration), this will throw `ClassCastException` during planning and fail the job, instead of safely skipping fusion. Add an `instanceof` guard (and optionally a warning) so fusion becomes a best-effort optimization rather than a hard failure under misconfiguration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "Tartarus0zm (via GitHub)" <gi...@apache.org>.
Tartarus0zm commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3410865028
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
Review Comment:
@weiqingy This issue was originally designed for fusing adjacent Auron native operators, but it indeed only considered the single-operator (single source) scenario, without taking into account the multi-operator reuse case.
I haven't encountered the multi-operator reuse scenario very often, so I didn't consider it at the time.
I still recommend going with the graph-level approach from the start.
- First the logic is clearer; second, it covers a wider range of scenarios and is more general-purpose; and third, subsequent operator fusion work can continue to reuse it.
- If we implement this based on StreamExecCalc now, once the graph-level implementation is done in the future, how would we handle the code in StreamExecCalc?
I think this issue can be treated as covering single-operator fusion, with #2329 handling multi-operator reuse fusion. If #2329 can completely cover the single-operator fusion scenario as well, then I think this issue can be abandoned.
What do you think? @weiqingy @SteNicholas
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3400521394
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
@@ -189,17 +194,52 @@ protected Transformation<RowData> translateToPlanInternal(PlannerBase planner, E
"Auron Calc conversion failed for node " + getId() + " and fallback is disabled");
}
- final FlinkAuronCalcOperator operator =
- new FlinkAuronCalcOperator(plan.get(), inputRowType, outputRowType, "FlinkAuronCalc-" + getId());
+ // Strict gating: the Calc converts to native only by fusing into a native Auron Kafka
+ // source with no event-time watermark. A watermarked source still emits per-record
+ // timestamps the downstream Calc must not strip, so fusion is disabled there. Any other
+ // upstream has no source to fuse into. A convertible-but-unfusible Calc is not a conversion
+ // failure, so it goes to Flink's Calc regardless of the fallback flag.
+ final AuronKafkaSourceFunction source = asNativeKafkaSource(upstream);
+ if (source != null && !source.hasWatermark()) {
+ source.setMergedCalcPlan(plan.get(), outputRowType);
Review Comment:
Confirmed, and it's reachable under default config. Instrumented repro: `SELECT age FROM T5 WHERE age>20 UNION ALL SELECT age+1 FROM T5 WHERE age>10` → both Calcs get the same source instance, both fuse, result is 6 rows instead of 5.
Fixed in `03b757df`: fusion now proceeds only when the source's sole-consumership is provable — source/sub-plan reuse disabled (so each scan gets its own source), plus an already-merged guard for the stacked-Calc case. A translating Calc can't see how many operators read its source, and one physical source can't serve two projections, so turning fusion on by default needs a graph-level pass that counts each source's consumers — filed as follow-up #2329 and recorded as Rev 5 in the [design doc](https://github.com/weiqingy/auron/blob/AURON-1865-design/docs/PR-1865/AURON-1865-DESIGN.md). Added a regression test that asserts the `UNION ALL` case is correct under default reuse.
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
@@ -189,17 +194,52 @@ protected Transformation<RowData> translateToPlanInternal(PlannerBase planner, E
"Auron Calc conversion failed for node " + getId() + " and fallback is disabled");
}
- final FlinkAuronCalcOperator operator =
- new FlinkAuronCalcOperator(plan.get(), inputRowType, outputRowType, "FlinkAuronCalc-" + getId());
+ // Strict gating: the Calc converts to native only by fusing into a native Auron Kafka
+ // source with no event-time watermark. A watermarked source still emits per-record
+ // timestamps the downstream Calc must not strip, so fusion is disabled there. Any other
+ // upstream has no source to fuse into. A convertible-but-unfusible Calc is not a conversion
+ // failure, so it goes to Flink's Calc regardless of the fallback flag.
+ final AuronKafkaSourceFunction source = asNativeKafkaSource(upstream);
Review Comment:
Agreed — the `instanceof LegacySourceTransformation` + unwrap is a Kafka-specific detection that silently falls back on any future wrapper. The wrapper-agnostic version belongs in the graph-level pass that the on-by-default fusion fix also needs, so I've tracked it together in #2329 (it would detect native sources structurally instead of by this Kafka-specific check).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "Tartarus0zm (via GitHub)" <gi...@apache.org>.
Tartarus0zm commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3410865028
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
Review Comment:
@weiqingy This issue was originally designed for fusing adjacent Auron native operators, but it indeed only considered the single-operator (single source) scenario, without taking into account the multi-operator reuse case.
I haven't encountered the multi-operator reuse scenario very often, so I didn't consider it at the time.
I still recommend going with the graph-level approach from the start.
- The logic is clearer; second, it covers a wider range of scenarios and is more general-purpose; and third, subsequent operator fusion work can continue to reuse it.
- If we implement this based on StreamExecCalc now, once the graph-level implementation is done in the future, how would we handle the code in StreamExecCalc?
I think this issue can be treated as covering single-operator fusion, with #2329 handling multi-operator reuse fusion. If #2329 can completely cover the single-operator fusion scenario as well, then I think this issue can be abandoned.
What do you think? @weiqingy @SteNicholas
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3400521394
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
@@ -189,17 +194,52 @@ protected Transformation<RowData> translateToPlanInternal(PlannerBase planner, E
"Auron Calc conversion failed for node " + getId() + " and fallback is disabled");
}
- final FlinkAuronCalcOperator operator =
- new FlinkAuronCalcOperator(plan.get(), inputRowType, outputRowType, "FlinkAuronCalc-" + getId());
+ // Strict gating: the Calc converts to native only by fusing into a native Auron Kafka
+ // source with no event-time watermark. A watermarked source still emits per-record
+ // timestamps the downstream Calc must not strip, so fusion is disabled there. Any other
+ // upstream has no source to fuse into. A convertible-but-unfusible Calc is not a conversion
+ // failure, so it goes to Flink's Calc regardless of the fallback flag.
+ final AuronKafkaSourceFunction source = asNativeKafkaSource(upstream);
+ if (source != null && !source.hasWatermark()) {
+ source.setMergedCalcPlan(plan.get(), outputRowType);
Review Comment:
Confirmed, and it's reachable under default config. Instrumented repro: `SELECT age FROM T5 WHERE age>20 UNION ALL SELECT age+1 FROM T5 WHERE age>10` → both Calcs get the same source instance, both fuse, result is 6 rows instead of 5.
Fixed in `03b757df`: fusion now proceeds only when the source's sole-consumership is provable — source/sub-plan reuse disabled (so each scan gets its own source), plus an already-merged guard for the stacked-Calc case. A translating Calc can't observe its source's consumer count in Flink's model, and one physical source can't serve two projections, so fully correct on-by-default fusion needs the graph-level consumer count — deferred to the #1264 Phase 2/3 graph merge (#2329), recorded as Rev 5 in the [design doc](https://github.com/weiqingy/auron/blob/AURON-1865-design/docs/PR-1865/AURON-1865-DESIGN.md). Added a regression test that asserts the `UNION ALL` case is correct under default reuse.
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
@@ -189,17 +194,52 @@ protected Transformation<RowData> translateToPlanInternal(PlannerBase planner, E
"Auron Calc conversion failed for node " + getId() + " and fallback is disabled");
}
- final FlinkAuronCalcOperator operator =
- new FlinkAuronCalcOperator(plan.get(), inputRowType, outputRowType, "FlinkAuronCalc-" + getId());
+ // Strict gating: the Calc converts to native only by fusing into a native Auron Kafka
+ // source with no event-time watermark. A watermarked source still emits per-record
+ // timestamps the downstream Calc must not strip, so fusion is disabled there. Any other
+ // upstream has no source to fuse into. A convertible-but-unfusible Calc is not a conversion
+ // failure, so it goes to Flink's Calc regardless of the fallback flag.
+ final AuronKafkaSourceFunction source = asNativeKafkaSource(upstream);
+ if (source != null && !source.hasWatermark()) {
+ source.setMergedCalcPlan(plan.get(), outputRowType);
+ // The fused source now emits the Calc's projected rows, but the source
+ // Transformation's declared type is still its original logical RowType. Re-type it to
+ // the projected output so the downstream serializer matches. setOutputType is the same
+ // call Flink's own CommonExecTableSourceScan uses to type this Transformation.
+ // Translation-order invariant: this re-typing is only legal because no ExecNode between
+ // the source and this Calc has read getOutputType() yet — Flink's Transformation locks
+ // the type on first read. An ExecNode inserted between source and Calc that reads the
+ // type before this point would lock the original RowType and make this re-type a no-op.
+ upstream.setOutputType(InternalTypeInfo.of(outputRowType));
Review Comment:
Same root cause — agreed. The sole-consumer gate removes this: with no shared, already-typed source to re-type, the second `setOutputType` is never reached. Closed by `03b757df`.
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
@@ -260,10 +300,9 @@ private Optional<PhysicalPlanNode> tryBuildAuronPlan(RowType inputRowType, RowTy
projectExprs.add(c.get());
}
- // numPartitions = 1 because each parallel FlinkAuronCalcOperator subtask owns a
- // single Java-side exporter and one corresponding native partition; per-subtask
- // parallelism is governed by Flink's outer Transformation parallelism, not by the
- // FFI Reader's partition count.
+ // numPartitions = 1 and the resource ID is a placeholder because this is only the
+ // logical sub-plan: the FFIReader leaf is spliced out and replaced by the source's
+ // KafkaScan when the sub-plan is fused into the Auron Kafka source.
final FFIReaderExecNode ffiReader = FFIReaderExecNode.newBuilder()
.setNumPartitions(1)
.setSchema(SchemaConverters.convertToAuronSchema(inputRowType, false))
Review Comment:
Agreed — `spliceScanIntoLeaf` throws the leaf away, so the input-schema build is wasted on every fused Calc. I'll switch to a bare marker leaf, folded into the dead-code cleanup (filed as #2328) since both touch the FFIReader leaf, so the leaf shape and the splice change together.
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
@@ -189,17 +194,52 @@ protected Transformation<RowData> translateToPlanInternal(PlannerBase planner, E
"Auron Calc conversion failed for node " + getId() + " and fallback is disabled");
}
- final FlinkAuronCalcOperator operator =
- new FlinkAuronCalcOperator(plan.get(), inputRowType, outputRowType, "FlinkAuronCalc-" + getId());
+ // Strict gating: the Calc converts to native only by fusing into a native Auron Kafka
+ // source with no event-time watermark. A watermarked source still emits per-record
+ // timestamps the downstream Calc must not strip, so fusion is disabled there. Any other
+ // upstream has no source to fuse into. A convertible-but-unfusible Calc is not a conversion
+ // failure, so it goes to Flink's Calc regardless of the fallback flag.
+ final AuronKafkaSourceFunction source = asNativeKafkaSource(upstream);
Review Comment:
Agreed — the `instanceof LegacySourceTransformation` + unwrap is a Kafka-specific detection that silently falls back on any future wrapper. The wrapper-agnostic version is the graph-level merge direction (#1264 Phase 2/3 in the [design doc](https://github.com/weiqingy/auron/blob/AURON-1865-design/docs/PR-1865/AURON-1865-DESIGN.md)); filed as #2329 so the invisible-fallback risk is tracked alongside the multi-consumer detection.
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -104,6 +108,14 @@ public class AuronKafkaSourceFunction extends RichParallelSourceFunction<RowData
private String mockData;
private transient PhysicalPlanNode physicalPlanNode;
+ // Merged Calc sub-plan handed down by a downstream Calc. When set, open() fuses the
+ // KafkaScan into this logical Project[Filter?[FFIReader-placeholder]] tree and runs the
+ // fused plan; the projected output schema replaces the original outputType downstream.
+ // Set at plan time, so these must survive operator serialization to the TaskManager
+ // (non-transient, like watermarkStrategy below).
+ private PhysicalPlanNode mergedCalcPlan;
+ private RowType mergedProjectedOutputType;
Review Comment:
The field is redundant, but I'd push back on deriving it: there's no reverse Arrow→`LogicalType` converter in the repo (only the forward `convertToAuronArrowType`), so reconstructing the `RowType` from the projection's arrow types would be lossy on nullability and add fragile new code — whereas the field is handed straight from the planner with the correct type already in hand.
Counter in `93aaae62` that addresses the real concern (the duplicated ternary in `run()` and `getOutputType()`): keep the field but collapse both into one `effectiveLogicalOutputType()` accessor, so there's a single source of truth. Happy to revisit a proper reverse converter later if you'd prefer the field gone.
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -501,6 +528,37 @@ public void setWatermarkStrategy(WatermarkStrategy<RowData> watermarkStrategy) {
this.watermarkStrategy = watermarkStrategy;
}
+ /**
+ * Registers a logical Calc sub-plan to fuse into the Kafka source so the native engine runs a
+ * single {@code Project[Filter?[KafkaScan]]} plan instead of a bare {@code KafkaScan} feeding a
+ * separate downstream Calc operator.
+ *
+ * <p>The sub-plan is a {@code Project[Filter?[FFIReader-placeholder]]} tree built by the
+ * downstream Calc against the LOGICAL-only input columns; its FFIReader leaf carries the
+ * resource ID {@code "placeholder"}. At {@code open()} the placeholder leaf is replaced by the
+ * freshly-built {@code KafkaScan} and the three Kafka metadata passthrough columns are prepended
+ * to the outer projection, so the fused output is {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ *
+ * @param logicalCalcSubPlan the {@code Project[Filter?[FFIReader-placeholder]]} tree over the
+ * logical input columns
+ * @param projectedOutputType the projected logical output row type (without the metadata
+ * columns) emitted downstream when the fused plan is active
+ */
+ public void setMergedCalcPlan(PhysicalPlanNode logicalCalcSubPlan, RowType projectedOutputType) {
+ this.mergedCalcPlan = logicalCalcSubPlan;
+ this.mergedProjectedOutputType = projectedOutputType;
+ }
+
+ /**
+ * Reports whether this source has a watermark strategy configured.
+ *
+ * @return {@code true} if a {@link WatermarkStrategy} was set
+ */
+ public boolean hasWatermark() {
Review Comment:
Agreed — `copy()` reset it to null and the gate relied on Flink re-applying the watermark afterward. Fixed in `2a8db799`: `copy()` now carries `watermarkStrategy`, so the no-fuse-over-watermark gate is self-contained.
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -548,6 +606,90 @@ private void discoverNewPartitions(int subtaskIndex, int numSubtasks) {
}
}
+ /**
+ * Fuses a logical Calc sub-plan with the source's {@code KafkaScan} into a single native plan.
+ *
+ * <p>Two steps: (1) replace the {@code FFIReader} placeholder leaf of {@code logicalSubPlan}
+ * with {@code kafkaScan}; (2) prepend the three Kafka metadata passthrough columns to the outer
+ * {@link ProjectionExecNode} so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ *
+ * <p>The inner projection's logical exprs reference logical column NAMES. The native planner
+ * resolves a {@code PhysicalColumn} by name against the input schema, ignoring the proto index,
+ * so the logical exprs resolve correctly against the KafkaScan output schema {@code [meta3,
+ * ...logical]} with no ordinal shift.
+ *
+ * @param logicalSubPlan a {@code Project[Filter?[FFIReader-placeholder]]} tree
+ * @param kafkaScan the {@code KafkaScan} {@link PhysicalPlanNode} to splice in as the leaf
+ * @return the fused {@code Project[Filter?[KafkaScan]]} tree with metadata passthrough prepended
+ */
+ static PhysicalPlanNode buildMergedPlan(PhysicalPlanNode logicalSubPlan, PhysicalPlanNode kafkaScan) {
+ PhysicalPlanNode spliced = spliceScanIntoLeaf(logicalSubPlan, kafkaScan);
+ return prependMetadataPassthrough(spliced);
+ }
+
+ /**
+ * Recursively walks {@code node}, locating the {@code FFIReader} leaf and replacing the whole
+ * FFIReader-bearing {@link PhysicalPlanNode} with {@code kafkaScan}. Accepted shapes:
+ * {@code Project[FFIReader]} and {@code Project[Filter[FFIReader]]} (the Calc always wraps its
+ * output in a projection).
+ */
+ private static PhysicalPlanNode spliceScanIntoLeaf(PhysicalPlanNode node, PhysicalPlanNode kafkaScan) {
Review Comment:
Confirmed — after this PR `FlinkAuronCalcOperator` is only referenced by its own test, and `spliceScanIntoLeaf` duplicates its `injectFfiReaderLeaf`. Filed as #2328: remove the operator and its test, collapse the two walkers into one, and fold in the bare-marker-leaf cleanup. It lands as a focused PR off master right after this merges — the operator is dead only once this is in, so removing it from a clean master keeps the diff honest and this PR's diff on the fusion + correctness work.
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -548,6 +606,90 @@ private void discoverNewPartitions(int subtaskIndex, int numSubtasks) {
}
}
+ /**
+ * Fuses a logical Calc sub-plan with the source's {@code KafkaScan} into a single native plan.
+ *
+ * <p>Two steps: (1) replace the {@code FFIReader} placeholder leaf of {@code logicalSubPlan}
+ * with {@code kafkaScan}; (2) prepend the three Kafka metadata passthrough columns to the outer
+ * {@link ProjectionExecNode} so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ *
+ * <p>The inner projection's logical exprs reference logical column NAMES. The native planner
+ * resolves a {@code PhysicalColumn} by name against the input schema, ignoring the proto index,
+ * so the logical exprs resolve correctly against the KafkaScan output schema {@code [meta3,
+ * ...logical]} with no ordinal shift.
+ *
+ * @param logicalSubPlan a {@code Project[Filter?[FFIReader-placeholder]]} tree
+ * @param kafkaScan the {@code KafkaScan} {@link PhysicalPlanNode} to splice in as the leaf
+ * @return the fused {@code Project[Filter?[KafkaScan]]} tree with metadata passthrough prepended
+ */
+ static PhysicalPlanNode buildMergedPlan(PhysicalPlanNode logicalSubPlan, PhysicalPlanNode kafkaScan) {
+ PhysicalPlanNode spliced = spliceScanIntoLeaf(logicalSubPlan, kafkaScan);
+ return prependMetadataPassthrough(spliced);
+ }
+
+ /**
+ * Recursively walks {@code node}, locating the {@code FFIReader} leaf and replacing the whole
+ * FFIReader-bearing {@link PhysicalPlanNode} with {@code kafkaScan}. Accepted shapes:
+ * {@code Project[FFIReader]} and {@code Project[Filter[FFIReader]]} (the Calc always wraps its
+ * output in a projection).
+ */
+ private static PhysicalPlanNode spliceScanIntoLeaf(PhysicalPlanNode node, PhysicalPlanNode kafkaScan) {
+ if (node.hasFfiReader()) {
+ return kafkaScan;
+ }
+ if (node.hasProjection()) {
+ PhysicalPlanNode rewrittenInput =
+ spliceScanIntoLeaf(node.getProjection().getInput(), kafkaScan);
+ return node.toBuilder()
+ .setProjection(node.getProjection().toBuilder()
+ .setInput(rewrittenInput)
+ .build())
+ .build();
+ }
+ if (node.hasFilter()) {
+ PhysicalPlanNode rewrittenInput =
+ spliceScanIntoLeaf(node.getFilter().getInput(), kafkaScan);
+ return node.toBuilder()
+ .setFilter(node.getFilter().toBuilder()
+ .setInput(rewrittenInput)
+ .build())
+ .build();
+ }
+ throw new IllegalArgumentException("Merged Calc sub-plan expects Project[Filter?[FFIReader]] shape; got: "
+ + node.getPhysicalPlanTypeCase());
+ }
+
+ /**
+ * Prepends the three Kafka metadata passthrough columns (partition, offset, timestamp) to the
+ * outer {@link ProjectionExecNode}'s parallel {@code expr} / {@code expr_name} / {@code
+ * data_type} lists, so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ */
+ private static PhysicalPlanNode prependMetadataPassthrough(PhysicalPlanNode merged) {
Review Comment:
Agreed — `prependMetadataPassthrough` assumed a Projection root and would silently overwrite a non-Projection with a meta-only projection. Fixed in `93aaae62`: it now fails fast with the actual `getPhysicalPlanTypeCase()` if the root isn't a Projection.
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -548,6 +606,90 @@ private void discoverNewPartitions(int subtaskIndex, int numSubtasks) {
}
}
+ /**
+ * Fuses a logical Calc sub-plan with the source's {@code KafkaScan} into a single native plan.
+ *
+ * <p>Two steps: (1) replace the {@code FFIReader} placeholder leaf of {@code logicalSubPlan}
+ * with {@code kafkaScan}; (2) prepend the three Kafka metadata passthrough columns to the outer
+ * {@link ProjectionExecNode} so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ *
+ * <p>The inner projection's logical exprs reference logical column NAMES. The native planner
+ * resolves a {@code PhysicalColumn} by name against the input schema, ignoring the proto index,
+ * so the logical exprs resolve correctly against the KafkaScan output schema {@code [meta3,
+ * ...logical]} with no ordinal shift.
+ *
+ * @param logicalSubPlan a {@code Project[Filter?[FFIReader-placeholder]]} tree
+ * @param kafkaScan the {@code KafkaScan} {@link PhysicalPlanNode} to splice in as the leaf
+ * @return the fused {@code Project[Filter?[KafkaScan]]} tree with metadata passthrough prepended
+ */
+ static PhysicalPlanNode buildMergedPlan(PhysicalPlanNode logicalSubPlan, PhysicalPlanNode kafkaScan) {
+ PhysicalPlanNode spliced = spliceScanIntoLeaf(logicalSubPlan, kafkaScan);
+ return prependMetadataPassthrough(spliced);
+ }
+
+ /**
+ * Recursively walks {@code node}, locating the {@code FFIReader} leaf and replacing the whole
+ * FFIReader-bearing {@link PhysicalPlanNode} with {@code kafkaScan}. Accepted shapes:
+ * {@code Project[FFIReader]} and {@code Project[Filter[FFIReader]]} (the Calc always wraps its
+ * output in a projection).
+ */
+ private static PhysicalPlanNode spliceScanIntoLeaf(PhysicalPlanNode node, PhysicalPlanNode kafkaScan) {
+ if (node.hasFfiReader()) {
+ return kafkaScan;
+ }
+ if (node.hasProjection()) {
+ PhysicalPlanNode rewrittenInput =
+ spliceScanIntoLeaf(node.getProjection().getInput(), kafkaScan);
+ return node.toBuilder()
+ .setProjection(node.getProjection().toBuilder()
+ .setInput(rewrittenInput)
+ .build())
+ .build();
+ }
+ if (node.hasFilter()) {
+ PhysicalPlanNode rewrittenInput =
+ spliceScanIntoLeaf(node.getFilter().getInput(), kafkaScan);
+ return node.toBuilder()
+ .setFilter(node.getFilter().toBuilder()
+ .setInput(rewrittenInput)
+ .build())
+ .build();
+ }
+ throw new IllegalArgumentException("Merged Calc sub-plan expects Project[Filter?[FFIReader]] shape; got: "
+ + node.getPhysicalPlanTypeCase());
+ }
+
+ /**
+ * Prepends the three Kafka metadata passthrough columns (partition, offset, timestamp) to the
+ * outer {@link ProjectionExecNode}'s parallel {@code expr} / {@code expr_name} / {@code
+ * data_type} lists, so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ */
+ private static PhysicalPlanNode prependMetadataPassthrough(PhysicalPlanNode merged) {
+ ProjectionExecNode projection = merged.getProjection();
+ List<PhysicalExprNode> originalExprs = new ArrayList<>(projection.getExprList());
+ List<String> originalNames = new ArrayList<>(projection.getExprNameList());
+ List<ArrowType> originalTypes = new ArrayList<>(projection.getDataTypeList());
+
+ ProjectionExecNode.Builder rebuilt =
+ projection.toBuilder().clearExpr().clearExprName().clearDataType();
+ addMetadataColumn(rebuilt, KAFKA_AURON_META_PARTITION_ID, new IntType(false));
+ addMetadataColumn(rebuilt, KAFKA_AURON_META_OFFSET, new BigIntType(false));
+ addMetadataColumn(rebuilt, KAFKA_AURON_META_TIMESTAMP, new BigIntType(false));
+ rebuilt.addAllExpr(originalExprs).addAllExprName(originalNames).addAllDataType(originalTypes);
+
+ return merged.toBuilder().setProjection(rebuilt.build()).build();
+ }
+
+ private static void addMetadataColumn(ProjectionExecNode.Builder builder, String name, LogicalType type) {
Review Comment:
Agreed on both, fixed in `93aaae62`. The three meta columns now live in one place (`KafkaConstants.KAFKA_AURON_META_FIELDS`), and `run()`, `prependMetadataPassthrough`, and `convertToAuronSchema(.., true)` all derive from it (verified byte-identical proto), so adding or renaming a meta column happens in one spot. And a logical projected column named one of the reserved meta names now fails fast on the fused path instead of silently resolving to the metadata column.
##########
auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.flink.table.kafka;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+import org.junit.jupiter.api.Test;
+
+/**
+ * End-to-end IT cases for fusing a convertible Calc into the native Auron Kafka source. The {@code
+ * T5} table (defined in {@link AuronKafkaSourceTestBase}) uses the {@code auron-kafka} connector and
+ * declares no event-time watermark, so its Calcs satisfy the strict fusion gate and execute natively
+ * as a single {@code Project[Filter?[KafkaScan]]} pipeline with no separate Calc operator.
+ *
+ * <p>The harness asserts the final row set; for the boolean-to-string test it additionally asserts
+ * the native lowercase {@code "true"}/{@code "false"} rendering, which differs from Flink codegen's
+ * uppercase form and thereby proves the fused path ran natively rather than falling back.
+ */
+public class AuronKafkaSourceMergeITCase extends AuronKafkaSourceTestBase {
+
+ /** Filter + projection over the no-watermark source fuses into the native source as a single
+ * {@code Project[Filter[KafkaScan]]} pipeline; only the age 21 and 22 rows survive the filter. */
+ @Test
+ public void testFilterAndProjectionFuse() {
+ environment.setParallelism(1);
+ List<Row> rows = CollectionUtil.iteratorToList(tableEnvironment
+ .executeSql("SELECT `age` + 1, `name` FROM T5 WHERE `age` > 20")
+ .collect());
+ rows.sort(Comparator.comparingInt(o -> (int) o.getField(0)));
+ assertThat(rows).isEqualTo(Arrays.asList(Row.of(22, "zm2"), Row.of(23, "zm1")));
+ }
+
+ /** A projection-only Calc over the no-watermark source fuses into the native source; every row
+ * is projected through {@code age + 1}. */
+ @Test
+ public void testProjectionOnlyFuse() {
+ environment.setParallelism(1);
+ List<Row> rows = CollectionUtil.iteratorToList(
+ tableEnvironment.executeSql("SELECT `age` + 1 FROM T5").collect());
+ rows.sort(Comparator.comparingInt(o -> (int) o.getField(0)));
+ assertThat(rows).isEqualTo(Arrays.asList(Row.of(21), Row.of(22), Row.of(23)));
+ }
+
+ /** A boolean-to-string CAST over the no-watermark source fuses and runs natively. The native
+ * engine renders booleans in lowercase ({@code "true"}/{@code "false"}), unlike Flink codegen's
+ * uppercase form, so the lowercase assertion proves the fused path executed natively. */
+ @Test
+ public void testBooleanToStringCastRendersNativeLowercase() {
+ environment.setParallelism(1);
+ List<Row> rows = CollectionUtil.iteratorToList(tableEnvironment
+ .executeSql("SELECT cast(`age` > 20 as STRING) FROM T5")
+ .collect());
+ rows.sort(Comparator.comparing(o -> (String) o.getField(0)));
+ // age 20 -> false, 21 -> true, 22 -> true; native rendering is lowercase.
+ assertThat(rows).isEqualTo(Arrays.asList(Row.of("false"), Row.of("true"), Row.of("true")));
+ }
+
+ /** A Calc over the watermarked source {@code T2} cannot fuse (the watermark gate blocks fusion),
+ * so it runs as a stock Flink Calc above the native source. The row set must still be correct:
+ * only age 21 and 22 survive the filter. */
+ @Test
+ public void testCalcOverWatermarkedSourceDoesNotFuseButIsCorrect() {
Review Comment:
I'd push back here with evidence. I instrumented the T2 path: the Calc's upstream is a `LegacySourceTransformation` (native source detected) with `hasWatermark` true, and it falls back via the `hasWatermark()` guard — not the non-native-upstream branch. Because the source implements `SupportsWatermarkPushDown`, Flink folds the watermark into the source rather than inserting an assigner operator, so the Calc's immediate upstream is still the native source.
So deleting `hasWatermark()` flips T2 to fuse — the test does exercise the guard, and a directly-watermarked source would be redundant. I'll add a short comment documenting the push-down mechanism so the intent is clear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on PR #2327:
URL: https://github.com/apache/auron/pull/2327#issuecomment-4687156692
Thanks for the thorough review — the single-root-cause framing was right, and it led to a sharper finding than I'd assumed.
I reproduced the shared-source case end-to-end: `SELECT age FROM T5 WHERE age>20 UNION ALL SELECT age+1 FROM T5 WHERE age>10` over the no-watermark source, instrumented. Both Calcs receive the same memoized `AuronKafkaSourceFunction` instance and both fuse, and the result is wrong — 6 rows `[21,21,22,22,23,23]` instead of 5 `[21,21,22,22,23]`. Since `reuse-source-enabled` defaults to `true`, this is a default-config correctness bug, so thank you for catching it.
One constraint shaped the fix: a translating Calc cannot see its source's fan-out (Flink's exec-node graph is one-directional), and a single physical source can only run one plan — so a "first-Calc-claims, others-fall-back" guard isn't sufficient, since the first fuse already re-points the shared source. A correct sole-consumer check needs graph-level consumer counting, which is the `ExecNodeGraphProcessor`-based merge already scoped to #1264 Phase 2/3 in the [design doc](https://github.com/weiqingy/auron/blob/AURON-1865-design/docs/PR-1865/AURON-1865-DESIGN.md).
So for this PR I gate fusion to the cases where sole-consumership is provable (source/sub-plan reuse disabled, plus an already-merged guard for stacked Calcs), which is correct under all configs, and defer correct on-by-default multi-consumer fusion to that graph merge — filed as #2329 and recorded as Rev 5 in the design. If you'd rather pull the graph-level detection into this PR despite the added `StreamPlanner`-shadow surface, I'm happy to — flagging the tradeoff since it touches the Phase-1/Phase-2 boundary.
Fixes pushed as separate commits (`03b757df` gating, `2a8db799` watermark `copy()`, `93aaae62` source-plan hardening). Two inline points I'd push back on with evidence (the derivable field and the watermarked-source test) — details inline. Everything else adopted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on PR #2327:
URL: https://github.com/apache/auron/pull/2327#issuecomment-4748200026
I reworked this PR onto the graph-level approach from the thread — the source→Calc fusion now happens in an `ExecNodeGraphProcessor` over the whole `ExecNodeGraph`, not inside `StreamExecCalc`.
Why graph level: whether a source can fuse is a fan-out question — "does this source have exactly one consumer?" — which a single `StreamExecCalc` cannot see in Flink's one-directional exec graph. The processor counts each source's consumers from the real topology and fuses only a sole-consumer `source → Calc`. So fusion is correct under default reuse config (the `isSourceFusionSafe` session-config proxy is gone) and on compiled plans — plan restore bypasses `ExecNodeGraphProcessor`s, so a restored plan simply does not fuse, and no reuse-config change can re-permit fusion into a shared source.
The processor is installed by subclassing `StreamPlanner` (overriding `getExecNodeGraphProcessors()`) and shadowing the small `DefaultPlannerFactory` to build it — the same assembly-overlay already used for `StreamExecCalc`. `StreamExecCalc` keeps only a thin completion branch: when its source already carries a staged plan it re-types the upstream transformation and returns it, emitting no standalone operator. `ExecNodeBase.outputType` is `private final`, so the re-type must happen at translation — the processor owns the decision, the Calc only completes it.
Fusion is additive: a convertible Calc that does not fuse still runs the standalone native Calc, and only a non-convertible Calc falls back to Flink codegen. So strict mode again means native-or-throw, and there is no default-config regression.
How the earlier review items map onto the rework:
- Cross-operator reach-across / detection / graph-level direction → the processor owns all fusion decisions; detection keys on a `FlinkAuronDynamicTableSource` marker, not a concrete class.
- Strict-mode native-or-throw + javadoc → restored by the additive decision; the javadoc is reconciled.
- Compiled-plan soundness → real consumer counting plus restore-de-fuses, not a session-config proxy.
- Metadata-count single source of truth → the reader offset now derives from `KAFKA_AURON_META_FIELDS.size()`.
- Reserved-meta-name collisions → rejected at the processor gate on input and output names, so a colliding name falls back at plan time instead of crash-looping every task.
- Watermark-gate test sensitivity → the ITs assert on operator topology (the count of `Calc` operators in the plan), so deleting the watermark gate flips the test.
Scope: this PR is single-consumer fusion; multi-consumer fusion (a shared source feeding several Calcs) builds on the same processor and stays in #2329.
The reusable pieces carried over — the Rex→native converter, the metadata splice + by-name column resolution, the `FlinkAuronFunction` marker, and the no-watermark emit fix; the placement and the config-inference gate are what moved. I force-pushed the reworked sequence, so the earlier inline comments are marked outdated — the mapping above covers each.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3409066916
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
Review Comment:
Thanks — this is the right question to put on the table, and I agree with where you're pointing.
On altitude: you're right that a `StreamExec` translating its own node shouldn't reach across to mutate its upstream. The end state I want is the one you describe — a graph-level pass (an `ExecNodeGraphProcessor`, or the `StreamGraphTranslator` rewrite à la Gluten) that sees the whole chain and fuses there, so no node reaches outside itself. I've tracked that as #2329; it's also what multi-consumer / source-reuse correctness needs, so the two converge.
On detection: agreed. I've switched it to key on the `FlinkAuronFunction` marker and lifted the fusion hand-off onto that interface, so it no longer depends on the concrete Kafka source — any native Auron source function is now a fusion target. The one Kafka-specific bit still remaining is the `LegacySourceTransformation` unwrap, which the #2329 graph-level pass removes entirely.
Where I'd value your call is sequencing. This PR gates fusion to the sole-consumer-provable case (source / sub-plan reuse off), so it's correct under every config, covered end-to-end, and mirrors the plan-merge already used on the Spark side — so I see it as a correct stepping stone that delivers the single-conversion win now, with the graph-level rewrite (#2329) as the proper home the later #1264 operator phases build on. The alternative is to hold this and go straight to the `StreamGraphTranslator` approach.
My preference is to land this as the interim and do the graph-level pass in #2329, but I'm happy to pivot now if you'd rather not carry the in-`StreamExecCalc` step.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3592630637
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -552,6 +670,102 @@ private void discoverNewPartitions(int subtaskIndex, int numSubtasks) {
}
}
+ /**
+ * Fuses a logical Calc sub-plan with the source's {@code KafkaScan} into a single native plan.
+ *
+ * <p>Two steps: (1) replace the {@code FFIReader} placeholder leaf of {@code logicalSubPlan}
+ * with {@code kafkaScan}; (2) prepend the three Kafka metadata passthrough columns to the outer
+ * {@link ProjectionExecNode} so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ *
+ * <p>The inner projection's logical exprs reference logical column NAMES. The native planner
+ * resolves a {@code PhysicalColumn} by name against the input schema, ignoring the proto index,
+ * so the logical exprs resolve correctly against the KafkaScan output schema {@code [meta3,
+ * ...logical]} with no ordinal shift.
+ *
+ * @param logicalSubPlan a {@code Project[Filter?[FFIReader-placeholder]]} tree
+ * @param kafkaScan the {@code KafkaScan} {@link PhysicalPlanNode} to splice in as the leaf
+ * @return the fused {@code Project[Filter?[KafkaScan]]} tree with metadata passthrough prepended
+ */
+ static PhysicalPlanNode buildMergedPlan(PhysicalPlanNode logicalSubPlan, PhysicalPlanNode kafkaScan) {
+ PhysicalPlanNode spliced = spliceScanIntoLeaf(logicalSubPlan, kafkaScan);
+ return prependMetadataPassthrough(spliced);
+ }
+
+ /**
+ * Recursively walks {@code node}, locating the {@code FFIReader} leaf and replacing the whole
+ * FFIReader-bearing {@link PhysicalPlanNode} with {@code kafkaScan}. Accepted shapes:
+ * {@code Project[FFIReader]} and {@code Project[Filter[FFIReader]]} (the Calc always wraps its
+ * output in a projection).
+ */
+ private static PhysicalPlanNode spliceScanIntoLeaf(PhysicalPlanNode node, PhysicalPlanNode kafkaScan) {
+ return AuronPlanTreeRewriter.rewriteFfiReaderLeaf(
+ node,
+ ffiReaderLeaf -> kafkaScan,
+ "Merged Calc sub-plan expects Project[Filter?[FFIReader]] shape; got: ");
+ }
+
+ /**
+ * Prepends the three Kafka metadata passthrough columns (partition, offset, timestamp) to the
+ * outer {@link ProjectionExecNode}'s parallel {@code expr} / {@code expr_name} / {@code
+ * data_type} lists, so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ */
+ private static PhysicalPlanNode prependMetadataPassthrough(PhysicalPlanNode merged) {
+ if (!merged.hasProjection()) {
+ throw new IllegalArgumentException(
+ "Merged Calc sub-plan expects a Projection root to prepend metadata passthrough; got: "
+ + merged.getPhysicalPlanTypeCase());
+ }
+ ProjectionExecNode projection = merged.getProjection();
+ List<PhysicalExprNode> originalExprs = new ArrayList<>(projection.getExprList());
+ List<String> originalNames = new ArrayList<>(projection.getExprNameList());
+ List<ArrowType> originalTypes = new ArrayList<>(projection.getDataTypeList());
+
+ for (RowType.RowField metaField : KAFKA_AURON_META_FIELDS) {
+ if (originalNames.contains(metaField.getName())) {
+ throw new IllegalArgumentException("Logical projected column name '" + metaField.getName()
+ + "' collides with a reserved Kafka metadata column; rename the column to fuse the Calc");
+ }
+ }
+
+ ProjectionExecNode.Builder rebuilt =
+ projection.toBuilder().clearExpr().clearExprName().clearDataType();
+ for (RowType.RowField metaField : KAFKA_AURON_META_FIELDS) {
+ addMetadataColumn(rebuilt, metaField.getName(), metaField.getType());
+ }
+ rebuilt.addAllExpr(originalExprs).addAllExprName(originalNames).addAllDataType(originalTypes);
+
+ return merged.toBuilder().setProjection(rebuilt.build()).build();
+ }
+
+ /**
+ * Resolves the negative (row-end-relative) index of the metadata column named {@code fieldName}.
+ * Metadata columns occupy the leading {@code metaCount} positions of every native output row, so
+ * a field at physical position {@code p} reads back at {@code p - metaCount}.
+ *
+ * @param fieldName the metadata column name, declared in {@link KafkaConstants#KAFKA_AURON_META_FIELDS}
+ * @param metaCount the number of leading metadata columns
+ * @return the negative index to pass to the row accessors for this metadata column
+ */
+ @VisibleForTesting
+ static int metaFieldNegativeIndex(String fieldName, int metaCount) {
+ for (int p = 0; p < KAFKA_AURON_META_FIELDS.size(); p++) {
+ if (KAFKA_AURON_META_FIELDS.get(p).getName().equals(fieldName)) {
+ return p - metaCount;
+ }
+ }
+ throw new IllegalStateException("Unknown Kafka metadata field: " + fieldName);
+ }
Review Comment:
`metaFieldNegativeIndex` accepts `metaCount` but does not validate it against `KAFKA_AURON_META_FIELDS.size()`. Passing a mismatched value would still find the field name but return an incorrect negative index, which could cause subtle misreads of metadata columns.
Add a defensive precondition (or remove the parameter) so any mismatch fails fast.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3393855073
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
@@ -189,17 +194,52 @@ protected Transformation<RowData> translateToPlanInternal(PlannerBase planner, E
"Auron Calc conversion failed for node " + getId() + " and fallback is disabled");
}
- final FlinkAuronCalcOperator operator =
- new FlinkAuronCalcOperator(plan.get(), inputRowType, outputRowType, "FlinkAuronCalc-" + getId());
+ // Strict gating: the Calc converts to native only by fusing into a native Auron Kafka
+ // source with no event-time watermark. A watermarked source still emits per-record
+ // timestamps the downstream Calc must not strip, so fusion is disabled there. Any other
+ // upstream has no source to fuse into. A convertible-but-unfusible Calc is not a conversion
+ // failure, so it goes to Flink's Calc regardless of the fallback flag.
+ final AuronKafkaSourceFunction source = asNativeKafkaSource(upstream);
+ if (source != null && !source.hasWatermark()) {
+ source.setMergedCalcPlan(plan.get(), outputRowType);
Review Comment:
**Correctness (high): a source reached by two Calcs gets its merged plan overwritten.**
`setMergedCalcPlan` mutates the shared `AuronKafkaSourceFunction`, but `ExecNodeBase.translateToPlan` memoizes the source transformation, so one source instance can be handed to more than one Calc:
- **Source reuse** (default `table.optimizer.reuse-source-enabled=true`): this source declares no projection/filter push-down, so two scans of the same table collapse into one shared source node. Two fusible Calcs above it both call `setMergedCalcPlan` — last write wins, and since both return the *same* re-typed `upstream`, the branches collapse to one stream. e.g. `SELECT age FROM T5 WHERE age>20 UNION ALL SELECT age+1 FROM T5 WHERE age>10` → silently wrong results.
- **Stacked Calcs** (`source→Calc1→Calc2` left unmerged by `CalcMergeRule`): Calc1 returns the source transformation, so Calc2 re-matches the same source, overwrites Calc1's plan, and builds exprs against Calc1's projected columns the `KafkaScan` never emits → native name-resolution failure.
Suggest fusing only when the source has exactly one consumer/output edge, and refusing to re-fuse an already-merged source (guard on `mergedCalcPlan != null`).
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
@@ -189,17 +194,52 @@ protected Transformation<RowData> translateToPlanInternal(PlannerBase planner, E
"Auron Calc conversion failed for node " + getId() + " and fallback is disabled");
}
- final FlinkAuronCalcOperator operator =
- new FlinkAuronCalcOperator(plan.get(), inputRowType, outputRowType, "FlinkAuronCalc-" + getId());
+ // Strict gating: the Calc converts to native only by fusing into a native Auron Kafka
+ // source with no event-time watermark. A watermarked source still emits per-record
+ // timestamps the downstream Calc must not strip, so fusion is disabled there. Any other
+ // upstream has no source to fuse into. A convertible-but-unfusible Calc is not a conversion
+ // failure, so it goes to Flink's Calc regardless of the fallback flag.
+ final AuronKafkaSourceFunction source = asNativeKafkaSource(upstream);
+ if (source != null && !source.hasWatermark()) {
+ source.setMergedCalcPlan(plan.get(), outputRowType);
+ // The fused source now emits the Calc's projected rows, but the source
+ // Transformation's declared type is still its original logical RowType. Re-type it to
+ // the projected output so the downstream serializer matches. setOutputType is the same
+ // call Flink's own CommonExecTableSourceScan uses to type this Transformation.
+ // Translation-order invariant: this re-typing is only legal because no ExecNode between
+ // the source and this Calc has read getOutputType() yet — Flink's Transformation locks
+ // the type on first read. An ExecNode inserted between source and Calc that reads the
+ // type before this point would lock the original RowType and make this re-type a no-op.
+ upstream.setOutputType(InternalTypeInfo.of(outputRowType));
Review Comment:
**Correctness: this is a *second* `setOutputType` on the source transformation and can throw.**
`CommonExecTableSourceScan` already typed this transformation. `Transformation.setOutputType` throws `IllegalStateException` once `typeUsed` is set (any prior `getOutputType()` read). The single-consumer happy path works because nothing reads the type in between — but in the shared/stacked-source case (see the `setMergedCalcPlan` comment), a downstream of the first consumer locks the type and this re-type aborts job translation instead of falling back. A single-consumer guard also resolves this.
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -501,6 +528,37 @@ public void setWatermarkStrategy(WatermarkStrategy<RowData> watermarkStrategy) {
this.watermarkStrategy = watermarkStrategy;
}
+ /**
+ * Registers a logical Calc sub-plan to fuse into the Kafka source so the native engine runs a
+ * single {@code Project[Filter?[KafkaScan]]} plan instead of a bare {@code KafkaScan} feeding a
+ * separate downstream Calc operator.
+ *
+ * <p>The sub-plan is a {@code Project[Filter?[FFIReader-placeholder]]} tree built by the
+ * downstream Calc against the LOGICAL-only input columns; its FFIReader leaf carries the
+ * resource ID {@code "placeholder"}. At {@code open()} the placeholder leaf is replaced by the
+ * freshly-built {@code KafkaScan} and the three Kafka metadata passthrough columns are prepended
+ * to the outer projection, so the fused output is {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ *
+ * @param logicalCalcSubPlan the {@code Project[Filter?[FFIReader-placeholder]]} tree over the
+ * logical input columns
+ * @param projectedOutputType the projected logical output row type (without the metadata
+ * columns) emitted downstream when the fused plan is active
+ */
+ public void setMergedCalcPlan(PhysicalPlanNode logicalCalcSubPlan, RowType projectedOutputType) {
+ this.mergedCalcPlan = logicalCalcSubPlan;
+ this.mergedProjectedOutputType = projectedOutputType;
+ }
+
+ /**
+ * Reports whether this source has a watermark strategy configured.
+ *
+ * @return {@code true} if a {@link WatermarkStrategy} was set
+ */
+ public boolean hasWatermark() {
Review Comment:
**Correctness (latent): this gate depends on `AuronKafkaDynamicTableSource.copy()` ordering.**
`copy()` does not carry over `watermarkStrategy` (resets it to null). It's safe today because Flink's `PushWatermarkIntoTableSourceScanRuleBase` applies the watermark spec to the copy *after* `copy()`. But the gate that blocks fusing over a watermarked source (fusion would strip the per-record kafka timestamps event-time needs) now silently relies on that exact ordering — any path that copies after `applyWatermark` without re-applying drops the watermark, lets the source fuse, and produces wrong event-time results. Preserving `watermarkStrategy` in `copy()` would make the gate self-contained.
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
@@ -189,17 +194,52 @@ protected Transformation<RowData> translateToPlanInternal(PlannerBase planner, E
"Auron Calc conversion failed for node " + getId() + " and fallback is disabled");
}
- final FlinkAuronCalcOperator operator =
- new FlinkAuronCalcOperator(plan.get(), inputRowType, outputRowType, "FlinkAuronCalc-" + getId());
+ // Strict gating: the Calc converts to native only by fusing into a native Auron Kafka
+ // source with no event-time watermark. A watermarked source still emits per-record
+ // timestamps the downstream Calc must not strip, so fusion is disabled there. Any other
+ // upstream has no source to fuse into. A convertible-but-unfusible Calc is not a conversion
+ // failure, so it goes to Flink's Calc regardless of the fallback flag.
+ final AuronKafkaSourceFunction source = asNativeKafkaSource(upstream);
Review Comment:
**Altitude (follow-up):** detecting the fuse target via `instanceof LegacySourceTransformation` + unwrapping `getOperator()/getUserFunction()`, then mutating the source function from the planner, is a Kafka-specific point solution. Any future wrapper between source and Calc (Source-API migration off `LegacySourceTransformation`, a partitioner/rebalance, a second native source type) silently fails the `instanceof` and falls back with no signal — an invisible perf regression. The PR notes a general ExecNode-merge is the intended direction; worth tracking.
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
@@ -260,10 +300,9 @@ private Optional<PhysicalPlanNode> tryBuildAuronPlan(RowType inputRowType, RowTy
projectExprs.add(c.get());
}
- // numPartitions = 1 because each parallel FlinkAuronCalcOperator subtask owns a
- // single Java-side exporter and one corresponding native partition; per-subtask
- // parallelism is governed by Flink's outer Transformation parallelism, not by the
- // FFI Reader's partition count.
+ // numPartitions = 1 and the resource ID is a placeholder because this is only the
+ // logical sub-plan: the FFIReader leaf is spliced out and replaced by the source's
+ // KafkaScan when the sub-plan is fused into the Auron Kafka source.
final FFIReaderExecNode ffiReader = FFIReaderExecNode.newBuilder()
.setNumPartitions(1)
.setSchema(SchemaConverters.convertToAuronSchema(inputRowType, false))
Review Comment:
**Cleanup (efficiency):** this builds a full Arrow input schema for the FFIReader leaf, but the whole leaf (schema included) is discarded by `spliceScanIntoLeaf` and replaced with the `KafkaScan` on the fused path. The conversion is wasted on every fused Calc, and the surviving FFIReader carries a real schema the splice silently drops — inviting a future change to rely on it. A bare marker leaf would express intent and save the work.
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -548,6 +606,90 @@ private void discoverNewPartitions(int subtaskIndex, int numSubtasks) {
}
}
+ /**
+ * Fuses a logical Calc sub-plan with the source's {@code KafkaScan} into a single native plan.
+ *
+ * <p>Two steps: (1) replace the {@code FFIReader} placeholder leaf of {@code logicalSubPlan}
+ * with {@code kafkaScan}; (2) prepend the three Kafka metadata passthrough columns to the outer
+ * {@link ProjectionExecNode} so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ *
+ * <p>The inner projection's logical exprs reference logical column NAMES. The native planner
+ * resolves a {@code PhysicalColumn} by name against the input schema, ignoring the proto index,
+ * so the logical exprs resolve correctly against the KafkaScan output schema {@code [meta3,
+ * ...logical]} with no ordinal shift.
+ *
+ * @param logicalSubPlan a {@code Project[Filter?[FFIReader-placeholder]]} tree
+ * @param kafkaScan the {@code KafkaScan} {@link PhysicalPlanNode} to splice in as the leaf
+ * @return the fused {@code Project[Filter?[KafkaScan]]} tree with metadata passthrough prepended
+ */
+ static PhysicalPlanNode buildMergedPlan(PhysicalPlanNode logicalSubPlan, PhysicalPlanNode kafkaScan) {
+ PhysicalPlanNode spliced = spliceScanIntoLeaf(logicalSubPlan, kafkaScan);
+ return prependMetadataPassthrough(spliced);
+ }
+
+ /**
+ * Recursively walks {@code node}, locating the {@code FFIReader} leaf and replacing the whole
+ * FFIReader-bearing {@link PhysicalPlanNode} with {@code kafkaScan}. Accepted shapes:
+ * {@code Project[FFIReader]} and {@code Project[Filter[FFIReader]]} (the Calc always wraps its
+ * output in a projection).
+ */
+ private static PhysicalPlanNode spliceScanIntoLeaf(PhysicalPlanNode node, PhysicalPlanNode kafkaScan) {
+ if (node.hasFfiReader()) {
+ return kafkaScan;
+ }
+ if (node.hasProjection()) {
+ PhysicalPlanNode rewrittenInput =
+ spliceScanIntoLeaf(node.getProjection().getInput(), kafkaScan);
+ return node.toBuilder()
+ .setProjection(node.getProjection().toBuilder()
+ .setInput(rewrittenInput)
+ .build())
+ .build();
+ }
+ if (node.hasFilter()) {
+ PhysicalPlanNode rewrittenInput =
+ spliceScanIntoLeaf(node.getFilter().getInput(), kafkaScan);
+ return node.toBuilder()
+ .setFilter(node.getFilter().toBuilder()
+ .setInput(rewrittenInput)
+ .build())
+ .build();
+ }
+ throw new IllegalArgumentException("Merged Calc sub-plan expects Project[Filter?[FFIReader]] shape; got: "
+ + node.getPhysicalPlanTypeCase());
+ }
+
+ /**
+ * Prepends the three Kafka metadata passthrough columns (partition, offset, timestamp) to the
+ * outer {@link ProjectionExecNode}'s parallel {@code expr} / {@code expr_name} / {@code
+ * data_type} lists, so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ */
+ private static PhysicalPlanNode prependMetadataPassthrough(PhysicalPlanNode merged) {
Review Comment:
**Robustness:** `prependMetadataPassthrough` adds meta columns to the outermost projection only and blindly does `merged.getProjection()`/`setProjection()`. Correct for the single `Project[Filter?[FFIReader]]` shape `tryBuildAuronPlan` emits today, but a nested `Project[Project[…]]` would leave the inner projection without the meta columns (outer meta lookup fails by name), and a non-Projection root would be silently overwritten by a 3-column meta-only projection (dropping the Filter) rather than failing fast. An explicit shape assertion would harden this if the Calc plan shape ever broadens.
##########
auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.flink.table.kafka;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+import org.junit.jupiter.api.Test;
+
+/**
+ * End-to-end IT cases for fusing a convertible Calc into the native Auron Kafka source. The {@code
+ * T5} table (defined in {@link AuronKafkaSourceTestBase}) uses the {@code auron-kafka} connector and
+ * declares no event-time watermark, so its Calcs satisfy the strict fusion gate and execute natively
+ * as a single {@code Project[Filter?[KafkaScan]]} pipeline with no separate Calc operator.
+ *
+ * <p>The harness asserts the final row set; for the boolean-to-string test it additionally asserts
+ * the native lowercase {@code "true"}/{@code "false"} rendering, which differs from Flink codegen's
+ * uppercase form and thereby proves the fused path ran natively rather than falling back.
+ */
+public class AuronKafkaSourceMergeITCase extends AuronKafkaSourceTestBase {
+
+ /** Filter + projection over the no-watermark source fuses into the native source as a single
+ * {@code Project[Filter[KafkaScan]]} pipeline; only the age 21 and 22 rows survive the filter. */
+ @Test
+ public void testFilterAndProjectionFuse() {
+ environment.setParallelism(1);
+ List<Row> rows = CollectionUtil.iteratorToList(tableEnvironment
+ .executeSql("SELECT `age` + 1, `name` FROM T5 WHERE `age` > 20")
+ .collect());
+ rows.sort(Comparator.comparingInt(o -> (int) o.getField(0)));
+ assertThat(rows).isEqualTo(Arrays.asList(Row.of(22, "zm2"), Row.of(23, "zm1")));
+ }
+
+ /** A projection-only Calc over the no-watermark source fuses into the native source; every row
+ * is projected through {@code age + 1}. */
+ @Test
+ public void testProjectionOnlyFuse() {
+ environment.setParallelism(1);
+ List<Row> rows = CollectionUtil.iteratorToList(
+ tableEnvironment.executeSql("SELECT `age` + 1 FROM T5").collect());
+ rows.sort(Comparator.comparingInt(o -> (int) o.getField(0)));
+ assertThat(rows).isEqualTo(Arrays.asList(Row.of(21), Row.of(22), Row.of(23)));
+ }
+
+ /** A boolean-to-string CAST over the no-watermark source fuses and runs natively. The native
+ * engine renders booleans in lowercase ({@code "true"}/{@code "false"}), unlike Flink codegen's
+ * uppercase form, so the lowercase assertion proves the fused path executed natively. */
+ @Test
+ public void testBooleanToStringCastRendersNativeLowercase() {
+ environment.setParallelism(1);
+ List<Row> rows = CollectionUtil.iteratorToList(tableEnvironment
+ .executeSql("SELECT cast(`age` > 20 as STRING) FROM T5")
+ .collect());
+ rows.sort(Comparator.comparing(o -> (String) o.getField(0)));
+ // age 20 -> false, 21 -> true, 22 -> true; native rendering is lowercase.
+ assertThat(rows).isEqualTo(Arrays.asList(Row.of("false"), Row.of("true"), Row.of("true")));
+ }
+
+ /** A Calc over the watermarked source {@code T2} cannot fuse (the watermark gate blocks fusion),
+ * so it runs as a stock Flink Calc above the native source. The row set must still be correct:
+ * only age 21 and 22 survive the filter. */
+ @Test
+ public void testCalcOverWatermarkedSourceDoesNotFuseButIsCorrect() {
Review Comment:
**Test validity:** this claims to cover the `hasWatermark()` gate, but T2's watermark is on a computed column, so Flink inserts a watermark-assigner/projection between the source and the Calc — the Calc's upstream is no longer a `LegacySourceTransformation`, so `asNativeKafkaSource()` returns null and fusion is blocked by the *non-native-upstream* branch, not the watermark branch. If `hasWatermark()` were deleted this test would still pass (false confidence — and it's the only guard against the `copy()` watermark-drop risk noted on `hasWatermark`). A directly-watermarked native source (no computed column) would actually exercise the gate.
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -104,6 +108,14 @@ public class AuronKafkaSourceFunction extends RichParallelSourceFunction<RowData
private String mockData;
private transient PhysicalPlanNode physicalPlanNode;
+ // Merged Calc sub-plan handed down by a downstream Calc. When set, open() fuses the
+ // KafkaScan into this logical Project[Filter?[FFIReader-placeholder]] tree and runs the
+ // fused plan; the projected output schema replaces the original outputType downstream.
+ // Set at plan time, so these must survive operator serialization to the TaskManager
+ // (non-transient, like watermarkStrategy below).
+ private PhysicalPlanNode mergedCalcPlan;
+ private RowType mergedProjectedOutputType;
Review Comment:
**Simplification:** `mergedProjectedOutputType` is derivable from `mergedCalcPlan`'s outer `ProjectionExecNode` (field names + arrow types) and drives a duplicated `mergedProjectedOutputType != null ? … : (RowType) outputType` ternary in both `getOutputType()` (L435) and `run()` (L329). Deriving it on demand removes the field and collapses both ternaries to one source of truth.
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -548,6 +606,90 @@ private void discoverNewPartitions(int subtaskIndex, int numSubtasks) {
}
}
+ /**
+ * Fuses a logical Calc sub-plan with the source's {@code KafkaScan} into a single native plan.
+ *
+ * <p>Two steps: (1) replace the {@code FFIReader} placeholder leaf of {@code logicalSubPlan}
+ * with {@code kafkaScan}; (2) prepend the three Kafka metadata passthrough columns to the outer
+ * {@link ProjectionExecNode} so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ *
+ * <p>The inner projection's logical exprs reference logical column NAMES. The native planner
+ * resolves a {@code PhysicalColumn} by name against the input schema, ignoring the proto index,
+ * so the logical exprs resolve correctly against the KafkaScan output schema {@code [meta3,
+ * ...logical]} with no ordinal shift.
+ *
+ * @param logicalSubPlan a {@code Project[Filter?[FFIReader-placeholder]]} tree
+ * @param kafkaScan the {@code KafkaScan} {@link PhysicalPlanNode} to splice in as the leaf
+ * @return the fused {@code Project[Filter?[KafkaScan]]} tree with metadata passthrough prepended
+ */
+ static PhysicalPlanNode buildMergedPlan(PhysicalPlanNode logicalSubPlan, PhysicalPlanNode kafkaScan) {
+ PhysicalPlanNode spliced = spliceScanIntoLeaf(logicalSubPlan, kafkaScan);
+ return prependMetadataPassthrough(spliced);
+ }
+
+ /**
+ * Recursively walks {@code node}, locating the {@code FFIReader} leaf and replacing the whole
+ * FFIReader-bearing {@link PhysicalPlanNode} with {@code kafkaScan}. Accepted shapes:
+ * {@code Project[FFIReader]} and {@code Project[Filter[FFIReader]]} (the Calc always wraps its
+ * output in a projection).
+ */
+ private static PhysicalPlanNode spliceScanIntoLeaf(PhysicalPlanNode node, PhysicalPlanNode kafkaScan) {
Review Comment:
**Cleanup (reuse + dead code):** `spliceScanIntoLeaf` is a near-verbatim duplicate of `FlinkAuronCalcOperator.injectFfiReaderLeaf` (same recursive Project/Filter/FFIReader walk and error). After this PR, `FlinkAuronCalcOperator` and its (large) unit test are orphaned — nothing emits the standalone FFIReader native-Calc path anymore. Either share one parameterized tree-rewriter or remove the now-dead operator + test, so the two walkers can't drift apart.
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -548,6 +606,90 @@ private void discoverNewPartitions(int subtaskIndex, int numSubtasks) {
}
}
+ /**
+ * Fuses a logical Calc sub-plan with the source's {@code KafkaScan} into a single native plan.
+ *
+ * <p>Two steps: (1) replace the {@code FFIReader} placeholder leaf of {@code logicalSubPlan}
+ * with {@code kafkaScan}; (2) prepend the three Kafka metadata passthrough columns to the outer
+ * {@link ProjectionExecNode} so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ *
+ * <p>The inner projection's logical exprs reference logical column NAMES. The native planner
+ * resolves a {@code PhysicalColumn} by name against the input schema, ignoring the proto index,
+ * so the logical exprs resolve correctly against the KafkaScan output schema {@code [meta3,
+ * ...logical]} with no ordinal shift.
+ *
+ * @param logicalSubPlan a {@code Project[Filter?[FFIReader-placeholder]]} tree
+ * @param kafkaScan the {@code KafkaScan} {@link PhysicalPlanNode} to splice in as the leaf
+ * @return the fused {@code Project[Filter?[KafkaScan]]} tree with metadata passthrough prepended
+ */
+ static PhysicalPlanNode buildMergedPlan(PhysicalPlanNode logicalSubPlan, PhysicalPlanNode kafkaScan) {
+ PhysicalPlanNode spliced = spliceScanIntoLeaf(logicalSubPlan, kafkaScan);
+ return prependMetadataPassthrough(spliced);
+ }
+
+ /**
+ * Recursively walks {@code node}, locating the {@code FFIReader} leaf and replacing the whole
+ * FFIReader-bearing {@link PhysicalPlanNode} with {@code kafkaScan}. Accepted shapes:
+ * {@code Project[FFIReader]} and {@code Project[Filter[FFIReader]]} (the Calc always wraps its
+ * output in a projection).
+ */
+ private static PhysicalPlanNode spliceScanIntoLeaf(PhysicalPlanNode node, PhysicalPlanNode kafkaScan) {
+ if (node.hasFfiReader()) {
+ return kafkaScan;
+ }
+ if (node.hasProjection()) {
+ PhysicalPlanNode rewrittenInput =
+ spliceScanIntoLeaf(node.getProjection().getInput(), kafkaScan);
+ return node.toBuilder()
+ .setProjection(node.getProjection().toBuilder()
+ .setInput(rewrittenInput)
+ .build())
+ .build();
+ }
+ if (node.hasFilter()) {
+ PhysicalPlanNode rewrittenInput =
+ spliceScanIntoLeaf(node.getFilter().getInput(), kafkaScan);
+ return node.toBuilder()
+ .setFilter(node.getFilter().toBuilder()
+ .setInput(rewrittenInput)
+ .build())
+ .build();
+ }
+ throw new IllegalArgumentException("Merged Calc sub-plan expects Project[Filter?[FFIReader]] shape; got: "
+ + node.getPhysicalPlanTypeCase());
+ }
+
+ /**
+ * Prepends the three Kafka metadata passthrough columns (partition, offset, timestamp) to the
+ * outer {@link ProjectionExecNode}'s parallel {@code expr} / {@code expr_name} / {@code
+ * data_type} lists, so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ */
+ private static PhysicalPlanNode prependMetadataPassthrough(PhysicalPlanNode merged) {
+ ProjectionExecNode projection = merged.getProjection();
+ List<PhysicalExprNode> originalExprs = new ArrayList<>(projection.getExprList());
+ List<String> originalNames = new ArrayList<>(projection.getExprNameList());
+ List<ArrowType> originalTypes = new ArrayList<>(projection.getDataTypeList());
+
+ ProjectionExecNode.Builder rebuilt =
+ projection.toBuilder().clearExpr().clearExprName().clearDataType();
+ addMetadataColumn(rebuilt, KAFKA_AURON_META_PARTITION_ID, new IntType(false));
+ addMetadataColumn(rebuilt, KAFKA_AURON_META_OFFSET, new BigIntType(false));
+ addMetadataColumn(rebuilt, KAFKA_AURON_META_TIMESTAMP, new BigIntType(false));
+ rebuilt.addAllExpr(originalExprs).addAllExprName(originalNames).addAllDataType(originalTypes);
+
+ return merged.toBuilder().setProjection(rebuilt.build()).build();
+ }
+
+ private static void addMetadataColumn(ProjectionExecNode.Builder builder, String name, LogicalType type) {
Review Comment:
**Cleanup + minor correctness:**
- (reuse) This hand-encodes the 3 Kafka meta columns a *third* time — `SchemaConverters.convertToAuronSchema(.., true)` and `run()` (L304-306) already define the same name/order/type/nullability set. The contract now lives in three spots; adding/renaming a meta column needs all three edited in lockstep.
- (correctness, low-likelihood) Native resolves these passthroughs by name (`planner.rs:876`). A user logical column literally named `serialized_kafka_records_partition/offset/timestamp` would collide and resolve to the meta column. Unguarded.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3409066916
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
Review Comment:
Thanks — this is the right question to put on the table, and I agree with where you're pointing.
On altitude: you're right that a `StreamExec` translating its own node shouldn't reach across to mutate its upstream. The end state I want is the one you describe — a graph-level pass (an `ExecNodeGraphProcessor`, or rewriting `StreamGraphTranslator` as Gluten does) that sees the whole chain and fuses there, so no node reaches outside itself. I've tracked that as #2329; it's also what multi-consumer / source-reuse correctness needs, so the two converge.
On detection: agreed. I've switched it to key on the `FlinkAuronFunction` marker and lifted the fusion hand-off onto that interface, so it no longer depends on the concrete Kafka source — any native Auron source function is now a fusion target. The one Kafka-specific bit still remaining is the `LegacySourceTransformation` unwrap, which the #2329 graph-level pass removes entirely.
Where I'd value your call is sequencing. This PR gates fusion to the sole-consumer-provable case (source / sub-plan reuse off), so it's correct under every config, covered end-to-end, and mirrors the plan-merge already used on the Spark side — so I see it as a correct stepping stone that delivers the single-conversion win now, with the graph-level rewrite (#2329) as the proper home the later #1264 operator phases build on. The alternative is to hold this and go straight to the `StreamGraphTranslator` approach.
My preference is to land this as the interim and do the graph-level pass in #2329, but I'm happy to pivot now if you'd rather not carry the in-`StreamExecCalc` step.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3412624049
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -300,11 +321,10 @@ public void add(String name, long value) {
LOG.debug("Metric Auron Source: {} = {}", name, value);
}
};
- List<RowType.RowField> fieldList = new LinkedList<>();
- fieldList.add(new RowType.RowField(KAFKA_AURON_META_PARTITION_ID, new IntType(false)));
- fieldList.add(new RowType.RowField(KAFKA_AURON_META_OFFSET, new BigIntType(false)));
- fieldList.add(new RowType.RowField(KAFKA_AURON_META_TIMESTAMP, new BigIntType(false)));
- fieldList.addAll(((RowType) outputType).getFields());
+ // The native output carries [meta3, logical], where logical is the projected output when a
+ // merged Calc plan is active and the original output otherwise.
+ List<RowType.RowField> fieldList = new LinkedList<>(KAFKA_AURON_META_FIELDS);
+ fieldList.addAll(effectiveLogicalOutputType().getFields());
RowType auronOutputRowType = new RowType(fieldList);
Review Comment:
**Cleanup (correctness footgun): the metadata count is still hardcoded `3` despite `KAFKA_AURON_META_FIELDS` being declared the single source of truth.** This line correctly derives the row type from `KAFKA_AURON_META_FIELDS`, but both `run()` loops below still use literal `FlinkArrowReader.create(batch, auronOutputRowType, 3)` and `getInt(-3)` / `getLong(-2)` / `getLong(-1)`. `KafkaConstants`' javadoc promises "adding or renaming a metadata column happens in one place," yet adding a 4th meta column would grow `auronOutputRowType` to 4 meta cols while the reader still assumes 3 → every row reads the wrong Arrow vector. Derive the offset from `KAFKA_AURON_META_FIELDS.size()` so the contract actually holds.
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
@@ -189,17 +195,81 @@ protected Transformation<RowData> translateToPlanInternal(PlannerBase planner, E
"Auron Calc conversion failed for node " + getId() + " and fallback is disabled");
}
- final FlinkAuronCalcOperator operator =
- new FlinkAuronCalcOperator(plan.get(), inputRowType, outputRowType, "FlinkAuronCalc-" + getId());
+ // Strict gating: the Calc converts to native only by fusing into a native Auron source
+ // with no event-time watermark. A watermarked source still emits per-record
+ // timestamps the downstream Calc must not strip, so fusion is disabled there. Any other
+ // upstream has no source to fuse into. A convertible-but-unfusible Calc is not a conversion
+ // failure, so it goes to Flink's Calc regardless of the fallback flag.
+ //
+ // Fusion mutates the shared source operator, so it is only correct when this Calc is the
+ // source's SOLE consumer. A Calc cannot observe its source's fan-out (Flink's exec graph is
+ // one-directional), so sole-consumership is inferred two ways: the source must not already
+ // carry a fused plan (a second Calc would otherwise overwrite the first), and source/sub-plan
+ // reuse must be disabled (with reuse off, Flink builds a distinct source per scan, so each
+ // source has exactly one consumer). When reuse is on, one source can feed multiple Calcs and
+ // last-write-wins would silently corrupt results, so fusion is skipped.
+ final FlinkAuronFunction source = asNativeFusibleSource(upstream);
+ if (source != null && !source.hasWatermark() && !source.isMergedCalcPlanSet() && isSourceFusionSafe(config)) {
+ source.setMergedCalcPlan(plan.get(), outputRowType);
+ // The fused source now emits the Calc's projected rows, but the source
+ // Transformation's declared type is still its original logical RowType. Re-type it to
+ // the projected output so the downstream serializer matches. setOutputType is the same
+ // call Flink's own CommonExecTableSourceScan uses to type this Transformation.
+ // Translation-order invariant: this re-typing is only legal because no ExecNode between
+ // the source and this Calc has read getOutputType() yet — Flink's Transformation locks
+ // the type on first read. An ExecNode inserted between source and Calc that reads the
+ // type before this point would lock the original RowType and make this re-type a no-op.
+ upstream.setOutputType(InternalTypeInfo.of(outputRowType));
+ return upstream;
+ }
+
+ LOG.debug("Calc node {} converts but cannot fuse into a native source; using Flink Calc.", getId());
+ return translateToFlinkCalc(planner, config);
+ }
- return ExecNodeUtil.createOneInputTransformation(
- upstream,
- createTransformationMeta(CALC_TRANSFORMATION, config),
- SimpleOperatorFactory.of(operator),
- InternalTypeInfo.of(outputRowType),
- upstream.getParallelism(),
- 0L,
- false);
+ /**
+ * Returns the {@link FlinkAuronFunction} backing {@code upstream} if it is a direct legacy source
+ * transformation over a native Auron source function, otherwise {@code null}.
+ *
+ * <p>{@code execEnv.addSource(sourceFunction)} produces a {@link LegacySourceTransformation} whose
+ * operator is a {@link StreamSource} (an {@link AbstractUdfStreamOperator}); its user function is
+ * the source function. Detection is by the {@link FlinkAuronFunction} marker, so any native Auron
+ * source function is a fusion target, not only the Kafka source.
+ *
+ * @param upstream the translated upstream {@link Transformation}
+ * @return the backing native source function, or {@code null} if {@code upstream} is not a direct
+ * legacy transformation over a {@link FlinkAuronFunction}
+ */
+ private static FlinkAuronFunction asNativeFusibleSource(Transformation<?> upstream) {
+ if (!(upstream instanceof LegacySourceTransformation)) {
+ return null;
+ }
+ final StreamSource<?, ?> streamSource = ((LegacySourceTransformation<?>) upstream).getOperator();
+ final Object userFunction = ((AbstractUdfStreamOperator<?, ?>) streamSource).getUserFunction();
+ if (userFunction instanceof FlinkAuronFunction) {
+ return (FlinkAuronFunction) userFunction;
+ }
+ return null;
+ }
+
+ /**
+ * Reports whether fusing into the upstream source is provably safe under the active optimizer
+ * configuration. Fusion is safe only when no source can be shared between Calcs: Flink shares a
+ * source across consumers only when sub-plan or source reuse is enabled, so fusion is gated to
+ * the case where both are disabled. With both off, every scan produces its own source operator,
+ * guaranteeing the fusing Calc is that source's sole consumer.
+ *
+ * <p>This is conservative: a Calc cannot inspect its source's fan-out in Flink's one-directional
+ * exec graph, so it cannot distinguish a shared source from a private one at translate time.
+ * Disabled-reuse is the only configuration that makes sole-consumership provable without
+ * graph-level consumer counting.
+ *
+ * @param config the exec-node config, which exposes the Flink table optimizer options
+ * @return {@code true} only when both source reuse and sub-plan reuse are disabled
+ */
+ private static boolean isSourceFusionSafe(ReadableConfig config) {
+ return !config.get(OptimizerConfigOptions.TABLE_OPTIMIZER_REUSE_SOURCE_ENABLED)
Review Comment:
**Correctness (latent): inferring sole-consumership from the session reuse config breaks for compiled plans.** `isSourceFusionSafe` reads `reuse-source-enabled`/`reuse-sub-plan-enabled` at translate time as a proxy for "this source has exactly one consumer." That holds for a freshly-planned SQL job, but `StreamExecCalc` is plan-serializable (`@JsonCreator` + `@ExecNodeMetadata`) and a compiled plan freezes source sharing at compile time. Compile a `UNION ALL` over one table with reuse enabled (one shared source → two Calcs), then `EXECUTE PLAN` in a session with reuse disabled: this returns `true`, the first Calc fuses and re-types the shared upstream, and the second consumer reads corrupted rows — exactly the last-write-wins case the gate is meant to block. Session config no longer reflects the compiled topology; real consumer-counting needs the graph-level pass (#2329).
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
@@ -189,17 +195,81 @@ protected Transformation<RowData> translateToPlanInternal(PlannerBase planner, E
"Auron Calc conversion failed for node " + getId() + " and fallback is disabled");
}
- final FlinkAuronCalcOperator operator =
- new FlinkAuronCalcOperator(plan.get(), inputRowType, outputRowType, "FlinkAuronCalc-" + getId());
+ // Strict gating: the Calc converts to native only by fusing into a native Auron source
+ // with no event-time watermark. A watermarked source still emits per-record
+ // timestamps the downstream Calc must not strip, so fusion is disabled there. Any other
+ // upstream has no source to fuse into. A convertible-but-unfusible Calc is not a conversion
+ // failure, so it goes to Flink's Calc regardless of the fallback flag.
+ //
+ // Fusion mutates the shared source operator, so it is only correct when this Calc is the
+ // source's SOLE consumer. A Calc cannot observe its source's fan-out (Flink's exec graph is
+ // one-directional), so sole-consumership is inferred two ways: the source must not already
+ // carry a fused plan (a second Calc would otherwise overwrite the first), and source/sub-plan
+ // reuse must be disabled (with reuse off, Flink builds a distinct source per scan, so each
+ // source has exactly one consumer). When reuse is on, one source can feed multiple Calcs and
+ // last-write-wins would silently corrupt results, so fusion is skipped.
+ final FlinkAuronFunction source = asNativeFusibleSource(upstream);
+ if (source != null && !source.hasWatermark() && !source.isMergedCalcPlanSet() && isSourceFusionSafe(config)) {
+ source.setMergedCalcPlan(plan.get(), outputRowType);
+ // The fused source now emits the Calc's projected rows, but the source
+ // Transformation's declared type is still its original logical RowType. Re-type it to
+ // the projected output so the downstream serializer matches. setOutputType is the same
+ // call Flink's own CommonExecTableSourceScan uses to type this Transformation.
+ // Translation-order invariant: this re-typing is only legal because no ExecNode between
+ // the source and this Calc has read getOutputType() yet — Flink's Transformation locks
+ // the type on first read. An ExecNode inserted between source and Calc that reads the
+ // type before this point would lock the original RowType and make this re-type a no-op.
+ upstream.setOutputType(InternalTypeInfo.of(outputRowType));
+ return upstream;
+ }
+
+ LOG.debug("Calc node {} converts but cannot fuse into a native source; using Flink Calc.", getId());
+ return translateToFlinkCalc(planner, config);
Review Comment:
**Correctness (behavior change + doc mismatch): strict mode no longer guarantees native-or-throw for a Calc.** When `plan.isPresent()` but the Calc can't fuse, this always returns `translateToFlinkCalc(...)` regardless of `FAIL_BACK_FLINK_ENGINE_ENABLED`. Two issues:
1. The class javadoc claims the opposite — "except when `FAIL_BACK_FLINK_ENGINE_ENABLED` is `false` and the Calc is fully convertible yet cannot fuse, where it throws `IllegalStateException`." The code never throws on that branch; the only throw is in the `!plan.isPresent()` branch above. Doc and code disagree.
2. Pre-PR a convertible Calc unconditionally emitted `FlinkAuronCalcOperator`, so strict mode meant "convert natively or throw." Now a convertible-but-unfusible Calc (which under default reuse is *every* Calc) silently degrades to Flink codegen at `debug` level — strict mode no longer implies native execution. Either re-check the fallback flag and throw here, or fix the javadoc to match.
##########
auron-flink-extension/auron-flink-planner/src/test/java/org/apache/auron/flink/table/kafka/AuronKafkaSourceMergeITCase.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.flink.table.kafka;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * End-to-end IT cases for fusing a convertible Calc into the native Auron Kafka source. The {@code
+ * T5} table (defined in {@link AuronKafkaSourceTestBase}) uses the {@code auron-kafka} connector and
+ * declares no event-time watermark, so its Calcs satisfy the strict fusion gate and execute natively
+ * as a single {@code Project[Filter?[KafkaScan]]} pipeline with no separate Calc operator.
+ *
+ * <p>The harness asserts the final row set; for the boolean-to-string test it additionally asserts
+ * the native lowercase {@code "true"}/{@code "false"} rendering, which differs from Flink codegen's
+ * uppercase form and thereby proves the fused path ran natively rather than falling back.
+ *
+ * <p>Fusion mutates the shared source operator, so it only runs when the fusing Calc is provably the
+ * source's sole consumer — which holds only when source and sub-plan reuse are both disabled. The
+ * fusion tests below therefore disable reuse so they exercise the fused native path; {@code
+ * testSharedSourceUnionAllIsCorrectUnderDefaultReuse} keeps reuse at its default (enabled) to verify
+ * the gate blocks fusion when a source can be shared.
+ */
+public class AuronKafkaSourceMergeITCase extends AuronKafkaSourceTestBase {
+
+ /**
+ * Disables source and sub-plan reuse by default so the fusion tests run with the gate satisfied.
+ * The shared-source regression test re-enables both before running.
+ */
+ @BeforeEach
+ public void disableSourceReuse() {
+ tableEnvironment.getConfig().set(OptimizerConfigOptions.TABLE_OPTIMIZER_REUSE_SOURCE_ENABLED, false);
+ tableEnvironment.getConfig().set(OptimizerConfigOptions.TABLE_OPTIMIZER_REUSE_SUB_PLAN_ENABLED, false);
+ }
+
+ /** Filter + projection over the no-watermark source fuses into the native source as a single
+ * {@code Project[Filter[KafkaScan]]} pipeline; only the age 21 and 22 rows survive the filter.
+ * Runs with source reuse disabled so the fusion gate is satisfied. */
+ @Test
+ public void testFilterAndProjectionFuse() {
+ environment.setParallelism(1);
+ List<Row> rows = CollectionUtil.iteratorToList(tableEnvironment
+ .executeSql("SELECT `age` + 1, `name` FROM T5 WHERE `age` > 20")
+ .collect());
+ rows.sort(Comparator.comparingInt(o -> (int) o.getField(0)));
+ assertThat(rows).isEqualTo(Arrays.asList(Row.of(22, "zm2"), Row.of(23, "zm1")));
+ }
+
+ /** A projection-only Calc over the no-watermark source fuses into the native source; every row
+ * is projected through {@code age + 1}. Runs with source reuse disabled so the fusion gate is
+ * satisfied. */
+ @Test
+ public void testProjectionOnlyFuse() {
+ environment.setParallelism(1);
+ List<Row> rows = CollectionUtil.iteratorToList(
+ tableEnvironment.executeSql("SELECT `age` + 1 FROM T5").collect());
+ rows.sort(Comparator.comparingInt(o -> (int) o.getField(0)));
+ assertThat(rows).isEqualTo(Arrays.asList(Row.of(21), Row.of(22), Row.of(23)));
+ }
+
+ /** A boolean-to-string CAST over the no-watermark source fuses and runs natively. The native
+ * engine renders booleans in lowercase ({@code "true"}/{@code "false"}), unlike Flink codegen's
+ * uppercase form, so the lowercase assertion proves the fused path executed natively. Runs with
+ * source reuse disabled so the fusion gate is satisfied. */
+ @Test
+ public void testBooleanToStringCastRendersNativeLowercase() {
+ environment.setParallelism(1);
+ List<Row> rows = CollectionUtil.iteratorToList(tableEnvironment
+ .executeSql("SELECT cast(`age` > 20 as STRING) FROM T5")
+ .collect());
+ rows.sort(Comparator.comparing(o -> (String) o.getField(0)));
+ // age 20 -> false, 21 -> true, 22 -> true; native rendering is lowercase.
+ assertThat(rows).isEqualTo(Arrays.asList(Row.of("false"), Row.of("true"), Row.of("true")));
+ }
+
+ /** When a single source feeds two Calcs (here via UNION ALL over the same table) under the
+ * default optimizer config, source reuse shares one source operator between both Calcs. The
+ * fusion gate must block fusion in that case — fusing both Calcs into the shared source would let
+ * the second overwrite the first and silently drop rows. With the gate blocking fusion, both
+ * Calcs run as stock Flink Calcs and the row set is correct: {@code WHERE age > 20} yields ages
+ * {@code [21, 22]}, and {@code WHERE age > 10} yields {@code [20, 21, 22]} projected through
+ * {@code age + 1} to {@code [21, 22, 23]}, for a 5-row union {@code [21, 21, 22, 22, 23]}. */
+ @Test
+ public void testSharedSourceUnionAllIsCorrectUnderDefaultReuse() {
+ tableEnvironment.getConfig().set(OptimizerConfigOptions.TABLE_OPTIMIZER_REUSE_SOURCE_ENABLED, true);
+ tableEnvironment.getConfig().set(OptimizerConfigOptions.TABLE_OPTIMIZER_REUSE_SUB_PLAN_ENABLED, true);
+ environment.setParallelism(1);
+ List<Row> rows = CollectionUtil.iteratorToList(tableEnvironment
+ .executeSql("SELECT `age` FROM T5 WHERE `age` > 20 "
+ + "UNION ALL SELECT `age` + 1 FROM T5 WHERE `age` > 10")
+ .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)));
+ }
+
+ /** A Calc over the watermarked source {@code T2} cannot fuse (the watermark gate blocks fusion),
+ * so it runs as a stock Flink Calc above the native source. The row set must still be correct:
+ * only age 21 and 22 survive the filter. */
+ @Test
+ public void testCalcOverWatermarkedSourceDoesNotFuseButIsCorrect() {
+ environment.setParallelism(1);
+ List<Row> rows = CollectionUtil.iteratorToList(tableEnvironment
+ .executeSql("SELECT `age` + 1 FROM T2 WHERE `age` > 20")
+ .collect());
+ rows.sort(Comparator.comparingInt(o -> (int) o.getField(0)));
+ assertThat(rows).isEqualTo(Arrays.asList(Row.of(22), Row.of(23)));
Review Comment:
**Test validity: this asserts only the row set `[22, 23]`, which is identical whether or not the Calc fuses, so it can't catch a regression in the `hasWatermark()` gate it's meant to cover.** To close the loop on our thread: you're right that `SupportsWatermarkPushDown` folds the watermark into the source, so the Calc's upstream stays a `LegacySourceTransformation` and `hasWatermark()` (not the non-native branch) is what fires — my earlier guess was wrong. But the assertion here is insensitive to that: delete `hasWatermark()` and T2 would wrongly fuse into a watermarked source (stripping per-record event-time), yet this test still passes because filter+projection yields `[22, 23]` either way. A watermark-dependent assertion (a windowed aggregation over `ts`) or an operator-topology assertion would actually lock the gate. Same insensitivity applies to the positive fusion cases — only the boolean-cast lowercase/uppercase case truly distinguishes native from Flink codegen.
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -548,6 +621,102 @@ private void discoverNewPartitions(int subtaskIndex, int numSubtasks) {
}
}
+ /**
+ * Fuses a logical Calc sub-plan with the source's {@code KafkaScan} into a single native plan.
+ *
+ * <p>Two steps: (1) replace the {@code FFIReader} placeholder leaf of {@code logicalSubPlan}
+ * with {@code kafkaScan}; (2) prepend the three Kafka metadata passthrough columns to the outer
+ * {@link ProjectionExecNode} so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ *
+ * <p>The inner projection's logical exprs reference logical column NAMES. The native planner
+ * resolves a {@code PhysicalColumn} by name against the input schema, ignoring the proto index,
+ * so the logical exprs resolve correctly against the KafkaScan output schema {@code [meta3,
+ * ...logical]} with no ordinal shift.
+ *
+ * @param logicalSubPlan a {@code Project[Filter?[FFIReader-placeholder]]} tree
+ * @param kafkaScan the {@code KafkaScan} {@link PhysicalPlanNode} to splice in as the leaf
+ * @return the fused {@code Project[Filter?[KafkaScan]]} tree with metadata passthrough prepended
+ */
+ static PhysicalPlanNode buildMergedPlan(PhysicalPlanNode logicalSubPlan, PhysicalPlanNode kafkaScan) {
+ PhysicalPlanNode spliced = spliceScanIntoLeaf(logicalSubPlan, kafkaScan);
+ return prependMetadataPassthrough(spliced);
+ }
+
+ /**
+ * Recursively walks {@code node}, locating the {@code FFIReader} leaf and replacing the whole
+ * FFIReader-bearing {@link PhysicalPlanNode} with {@code kafkaScan}. Accepted shapes:
+ * {@code Project[FFIReader]} and {@code Project[Filter[FFIReader]]} (the Calc always wraps its
+ * output in a projection).
+ */
+ private static PhysicalPlanNode spliceScanIntoLeaf(PhysicalPlanNode node, PhysicalPlanNode kafkaScan) {
+ if (node.hasFfiReader()) {
+ return kafkaScan;
+ }
+ if (node.hasProjection()) {
+ PhysicalPlanNode rewrittenInput =
+ spliceScanIntoLeaf(node.getProjection().getInput(), kafkaScan);
+ return node.toBuilder()
+ .setProjection(node.getProjection().toBuilder()
+ .setInput(rewrittenInput)
+ .build())
+ .build();
+ }
+ if (node.hasFilter()) {
+ PhysicalPlanNode rewrittenInput =
+ spliceScanIntoLeaf(node.getFilter().getInput(), kafkaScan);
+ return node.toBuilder()
+ .setFilter(node.getFilter().toBuilder()
+ .setInput(rewrittenInput)
+ .build())
+ .build();
+ }
+ throw new IllegalArgumentException("Merged Calc sub-plan expects Project[Filter?[FFIReader]] shape; got: "
+ + node.getPhysicalPlanTypeCase());
+ }
+
+ /**
+ * Prepends the three Kafka metadata passthrough columns (partition, offset, timestamp) to the
+ * outer {@link ProjectionExecNode}'s parallel {@code expr} / {@code expr_name} / {@code
+ * data_type} lists, so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ */
+ private static PhysicalPlanNode prependMetadataPassthrough(PhysicalPlanNode merged) {
+ if (!merged.hasProjection()) {
+ throw new IllegalArgumentException(
+ "Merged Calc sub-plan expects a Projection root to prepend metadata passthrough; got: "
+ + merged.getPhysicalPlanTypeCase());
+ }
+ ProjectionExecNode projection = merged.getProjection();
+ List<PhysicalExprNode> originalExprs = new ArrayList<>(projection.getExprList());
+ List<String> originalNames = new ArrayList<>(projection.getExprNameList());
+ List<ArrowType> originalTypes = new ArrayList<>(projection.getDataTypeList());
+
+ for (RowType.RowField metaField : KAFKA_AURON_META_FIELDS) {
+ if (originalNames.contains(metaField.getName())) {
Review Comment:
**Robustness: reserved-meta-name collisions are caught late and incompletely.**
1. *(output names)* This check runs at TaskManager `open()` and throws. But the planner fusion gate in `StreamExecCalc` has no equivalent check, so `SELECT age AS serialized_kafka_records_offset FROM T5` passes planning, fuses, then crash-loops every task instead of degrading to a stock Flink Calc at plan time. Checking the projected output names at the gate would let an unfusable name fall back rather than fail the job.
2. *(input names)* Only the projection's *output* names are checked. A logical *input* column literally named e.g. `serialized_kafka_records_timestamp` is unguarded: the spliced `KafkaScan` schema becomes `[meta3, ...logical]` with a duplicate name, and since native resolves columns by name (`planner.rs:876`) the inner expr binds to the meta column (index 2), silently returning Kafka timestamps in place of user data. Low-likelihood but silent — worth rejecting on input names too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3592703575
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -505,6 +572,54 @@ public void setWatermarkStrategy(WatermarkStrategy<RowData> watermarkStrategy) {
this.watermarkStrategy = watermarkStrategy;
}
+ /**
+ * Registers a logical Calc sub-plan to fuse into the Kafka source so the native engine runs a
+ * single {@code Project[Filter?[KafkaScan]]} plan instead of a bare {@code KafkaScan} feeding a
+ * separate downstream Calc operator.
+ *
+ * <p>The sub-plan is a {@code Project[Filter?[FFIReader-placeholder]]} tree built by the
+ * downstream Calc against the LOGICAL-only input columns; its FFIReader leaf carries the
+ * resource ID {@code "placeholder"}. At {@code open()} the placeholder leaf is replaced by the
+ * freshly-built {@code KafkaScan} and the three Kafka metadata passthrough columns are prepended
+ * to the outer projection, so the fused output is {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ *
+ * @param logicalCalcSubPlan the {@code Project[Filter?[FFIReader-placeholder]]} tree over the
+ * logical input columns
+ * @param projectedOutputType the projected logical output row type (without the metadata
+ * columns) emitted downstream when the fused plan is active
+ */
+ @Override
+ public void setMergedCalcPlan(PhysicalPlanNode logicalCalcSubPlan, RowType projectedOutputType) {
+ this.mergedCalcPlan = logicalCalcSubPlan;
+ this.mergedProjectedOutputType = projectedOutputType;
+ }
Review Comment:
Added fail-fast guards in `3a9cd6db` — `setMergedCalcPlan` now rejects null arguments and a second registration (`checkState(!isMergedCalcPlanSet())`), here and in the `AuronKafkaDynamicTableSource` mirror, with the stage-once/non-null contract documented on the `FlinkAuronFunction` interface.
For the record on current behavior: the readers already null-guard (`effectiveLogicalOutputType()` falls back to `outputType`, and the `open()`-time apply early-returns when the plan is null), and the fusion processor only stages after checking `isMergedCalcPlanSet()` on a sole-consumer source — so there was no reachable double-set or NPE today. The guards make that invariant self-enforcing at the interface boundary, which matters as multi-consumer fusion (#2329) adds new callers.
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaDynamicTableSource.java:
##########
@@ -140,4 +169,20 @@ public String asSummaryString() {
public void applyWatermark(WatermarkStrategy<RowData> watermarkStrategy) {
this.watermarkStrategy = watermarkStrategy;
}
+
+ @Override
+ public void setMergedCalcPlan(PhysicalPlanNode logicalCalcSubPlan, RowType projectedOutputType) {
+ this.mergedCalcPlan = logicalCalcSubPlan;
+ this.mergedProjectedOutputType = projectedOutputType;
+ }
Review Comment:
Same fix as the source-function setter, in `3a9cd6db`: this implementation now rejects null arguments and a second registration (`checkState(!isMergedCalcPlanSet())`), and the `FlinkAuronDynamicTableSource` interface documents the non-null, stage-once contract.
The overwrite was already prevented upstream — the fusion processor stages only after checking `isMergedCalcPlanSet()` on a sole-consumer source, and `copy()` carries the staged fields directly rather than through the setter — so this is boundary hardening for future callers (e.g. multi-consumer fusion in #2329) rather than a live bug.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on PR #2327:
URL: https://github.com/apache/auron/pull/2327#issuecomment-4677154054
Hi @Tartarus0zm, could you please help review this PR when you get a chance? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3440073921
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
Review Comment:
Went with the graph-level approach. Fusion now lives in an `ExecNodeGraphProcessor` that counts each source's consumers from the whole `ExecNodeGraph` and fuses only a sole-consumer `source → Calc`; `StreamExecCalc` no longer reaches across to decide. The per-finding mapping is in the summary comment above. This PR is the single-consumer case; multi-consumer stays in #2329.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3592804473
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/processor/AuronOperatorFusionProcessor.java:
##########
@@ -0,0 +1,292 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.flink.table.planner.processor;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+import org.apache.auron.flink.connector.kafka.KafkaConstants;
+import org.apache.auron.flink.runtime.operator.FlinkAuronDynamicTableSource;
+import org.apache.auron.flink.table.planner.FlinkAuronCalcNode;
+import org.apache.auron.flink.table.planner.converter.NativePlanFusionBuilder;
+import org.apache.auron.protobuf.PhysicalPlanNode;
+import org.apache.calcite.rex.RexNode;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.planner.delegation.PlannerBase;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeGraph;
+import org.apache.flink.table.planner.plan.nodes.exec.processor.ExecNodeGraphProcessor;
+import org.apache.flink.table.planner.plan.nodes.exec.processor.ProcessorContext;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecTableSourceScan;
+import org.apache.flink.table.types.logical.RowType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Graph-level fusion pass that merges a native Auron source and the {@code StreamExecCalc} that
+ * directly consumes it into a single native {@code Project[Filter?[KafkaScan]]} plan, so the chain
+ * stays columnar end-to-end and pays a single columnar→row conversion at the tail.
+ *
+ * <p>The pass runs over the whole {@link ExecNodeGraph} before Transformation translation. For each
+ * {@code source → Calc} chain it decides whether to fuse and, when it does, stages the merged
+ * Calc sub-plan on the source's {@link FlinkAuronDynamicTableSource} instance (annotate-and-skip): it
+ * does not remove or re-type any node. The shadowed {@code StreamExecCalc} later observes the staged
+ * plan and emits no standalone operator for that Calc. This decision is graph-level because
+ * sole-consumership ("does this source feed exactly one consumer?") is a whole-graph property a single
+ * node cannot see in Flink's one-directional exec graph.
+ *
+ * <p>Fusion is additive on top of the standalone native Calc path: a convertible Calc the pass does
+ * not fuse still runs as a standalone native operator; only a non-convertible Calc falls back to
+ * Flink codegen.
+ *
+ * <p>A Calc is fused into its source only when all hold:
+ *
+ * <ul>
+ * <li>the Calc's single input is a {@link StreamExecTableSourceScan} whose {@link
+ * DynamicTableSource} is a {@link FlinkAuronDynamicTableSource};
+ * <li>that scan is the sole consumer of the source (consumer count {@code == 1} —
+ * multi-consumer fusion is tracked separately);
+ * <li>the source has no event-time watermark (fusing below a watermark generator would strip
+ * per-record event-time);
+ * <li>the Calc fully converts to a native plan;
+ * <li>no projection-output or source-input column name collides with a reserved Kafka metadata
+ * column (the native engine resolves columns by name, so a collision would silently bind to the
+ * metadata column).
+ * </ul>
+ */
+public class AuronOperatorFusionProcessor implements ExecNodeGraphProcessor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AuronOperatorFusionProcessor.class);
+
+ @Override
+ public ExecNodeGraph process(ExecNodeGraph graph, ProcessorContext context) {
+ final PlannerBase planner = context.getPlanner();
+ return process(
+ graph,
+ scan -> scan.getTableSourceSpec()
+ .getScanTableSource(planner.getFlinkContext(), planner.getTypeFactory()),
+ planner.getTableConfig());
+ }
+
+ /**
+ * Runs the fusion decision over the whole graph: builds the reverse fan-out consumer count,
+ * collects every {@code scan → Calc} chain, and applies {@link #tryFuse} to each. Split from
+ * the {@link ProcessorContext} entry point so the full count→gate→stage path is drivable
+ * with the scan's source and the table config supplied directly, without a live {@link
+ * PlannerBase}.
+ *
+ * @param graph the exec-node graph to process
+ * @param sourceResolver resolves a scan node to its {@link DynamicTableSource}
+ * @param tableConfig the table config used to seed the native plan converter
+ * @return the same graph instance (annotate-and-skip: only source instances are mutated)
+ */
+ ExecNodeGraph process(
+ ExecNodeGraph graph,
+ Function<StreamExecTableSourceScan, DynamicTableSource> sourceResolver,
+ ReadableConfig tableConfig) {
+ final Map<Integer, Integer> consumerCount = buildConsumerCount(graph);
+
+ for (Map.Entry<ExecNode<?>, ExecNode<?>> chain :
+ findSourceCalcChains(graph).entrySet()) {
+ final StreamExecTableSourceScan scan = (StreamExecTableSourceScan) chain.getKey();
+ final StreamExecCalc calc = (StreamExecCalc) chain.getValue();
+ tryFuse(scan, calc, consumerCount, sourceResolver, tableConfig);
+ }
+ // Annotate-and-skip: the graph is structurally unchanged; only source instances are mutated.
+ return graph;
+ }
+
+ /**
+ * Builds a reverse fan-out map keyed by stable {@link ExecNode#getId()}, counting consumers per
+ * distinct consuming edge with visited-node dedup across all roots. A node reachable from
+ * multiple roots is counted once per distinct consumer edge, not once per traversal path, so a
+ * genuinely shared source can never be miscounted as {@code count == 1}.
+ *
+ * @param graph the exec-node graph to traverse
+ * @return a map from node id to the number of distinct consumers of that node
+ */
+ static Map<Integer, Integer> buildConsumerCount(ExecNodeGraph graph) {
+ final Map<Integer, Integer> consumerCount = new HashMap<>();
+ final Set<Integer> visited = new HashSet<>();
+ // Each visited edge contributes exactly one consumer to its source. Dedup is on the visited
+ // SET of nodes (so each node's input edges are walked once across all roots); the consumer
+ // increment happens per input edge of every visited node, which is the per-consuming-edge
+ // count the design requires.
+ final Deque<ExecNode<?>> stack = new ArrayDeque<>(graph.getRootNodes());
+ while (!stack.isEmpty()) {
+ final ExecNode<?> node = stack.pop();
+ if (!visited.add(node.getId())) {
+ continue;
+ }
+ for (ExecEdge edge : node.getInputEdges()) {
+ final ExecNode<?> src = edge.getSource();
+ consumerCount.merge(src.getId(), 1, Integer::sum);
+ stack.push(src);
+ }
+ }
+ return consumerCount;
+ }
+
+ /**
+ * Collects every {@code scan → Calc} chain where a {@link StreamExecTableSourceScan} is the
+ * single input source of a {@link StreamExecCalc}. Keyed by the scan node so each scan maps to the
+ * Calc that directly consumes it.
+ *
+ * @param graph the exec-node graph to traverse
+ * @return a map from each consuming scan to its {@link StreamExecCalc}
+ */
+ private static Map<ExecNode<?>, ExecNode<?>> findSourceCalcChains(ExecNodeGraph graph) {
+ final Map<ExecNode<?>, ExecNode<?>> chains = new HashMap<>();
+ final Set<Integer> visited = new HashSet<>();
+ final Deque<ExecNode<?>> stack = new ArrayDeque<>(graph.getRootNodes());
+ while (!stack.isEmpty()) {
+ final ExecNode<?> node = stack.pop();
+ if (!visited.add(node.getId())) {
+ continue;
+ }
+ if (node instanceof StreamExecCalc && node.getInputEdges().size() == 1) {
+ final ExecNode<?> input = node.getInputEdges().get(0).getSource();
+ if (input instanceof StreamExecTableSourceScan) {
+ chains.put(input, node);
+ }
+ }
+ for (ExecEdge edge : node.getInputEdges()) {
+ stack.push(edge.getSource());
+ }
+ }
+ return chains;
+ }
+
+ /**
+ * Applies the full fusion gate to one {@code scan → Calc} chain and, when every gate passes,
+ * stages the merged Calc plan on the source instance. Any failed gate is a silent no-fusion: the
+ * Calc then runs standalone native (if convertible) or Flink codegen.
+ *
+ * @param scan the source scan node
+ * @param calc the Calc node directly consuming the scan
+ * @param consumerCount the reverse fan-out map from {@link #buildConsumerCount}
+ * @param sourceResolver resolves the scan to its {@link DynamicTableSource}
+ * @param tableConfig the table config used to seed the native plan converter
+ */
+ void tryFuse(
+ StreamExecTableSourceScan scan,
+ StreamExecCalc calc,
+ Map<Integer, Integer> consumerCount,
+ Function<StreamExecTableSourceScan, DynamicTableSource> sourceResolver,
+ ReadableConfig tableConfig) {
+ // Sole-consumer gate: multi-consumer fusion is out of scope (tracked separately).
+ if (consumerCount.getOrDefault(scan.getId(), 0) != 1) {
+ return;
+ }
+
+ final DynamicTableSource tableSource = sourceResolver.apply(scan);
+
+ final FlinkAuronCalcNode calcNode = (FlinkAuronCalcNode) calc;
+ stagePlanIfFusible(
+ tableSource,
+ scan.getId(),
+ (RowType) calc.getInputEdges().get(0).getOutputType(),
+ (RowType) calc.getOutputType(),
+ calcNode.getProjection(),
+ calcNode.getCondition(),
+ tableConfig);
Review Comment:
Good point. Added an `instanceof FlinkAuronCalcNode` guard in `tryFuse` (`1ac7fe57`): if the classpath ever resolves to Flink's stock `StreamExecCalc` instead of Auron's shadowed one, fusion now logs a warning and skips rather than throwing. That also makes the cast honor `tryFuse`'s documented "silent no-fusion" contract.
For context: the shadowed `StreamExecCalc` always implements `FlinkAuronCalcNode` (declared that way, and shipped in the same assembly overlay that excludes the stock class), so in a correctly built jar the cast can't fail; the guard covers the misconfigured-classpath edge you flagged. I didn't add a unit test since the in-module test classpath only has the shadowed `StreamExecCalc`, so a stock instance can't be constructed to exercise the branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3592728432
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -552,6 +670,102 @@ private void discoverNewPartitions(int subtaskIndex, int numSubtasks) {
}
}
+ /**
+ * Fuses a logical Calc sub-plan with the source's {@code KafkaScan} into a single native plan.
+ *
+ * <p>Two steps: (1) replace the {@code FFIReader} placeholder leaf of {@code logicalSubPlan}
+ * with {@code kafkaScan}; (2) prepend the three Kafka metadata passthrough columns to the outer
+ * {@link ProjectionExecNode} so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ *
+ * <p>The inner projection's logical exprs reference logical column NAMES. The native planner
+ * resolves a {@code PhysicalColumn} by name against the input schema, ignoring the proto index,
+ * so the logical exprs resolve correctly against the KafkaScan output schema {@code [meta3,
+ * ...logical]} with no ordinal shift.
+ *
+ * @param logicalSubPlan a {@code Project[Filter?[FFIReader-placeholder]]} tree
+ * @param kafkaScan the {@code KafkaScan} {@link PhysicalPlanNode} to splice in as the leaf
+ * @return the fused {@code Project[Filter?[KafkaScan]]} tree with metadata passthrough prepended
+ */
+ static PhysicalPlanNode buildMergedPlan(PhysicalPlanNode logicalSubPlan, PhysicalPlanNode kafkaScan) {
+ PhysicalPlanNode spliced = spliceScanIntoLeaf(logicalSubPlan, kafkaScan);
+ return prependMetadataPassthrough(spliced);
+ }
+
+ /**
+ * Recursively walks {@code node}, locating the {@code FFIReader} leaf and replacing the whole
+ * FFIReader-bearing {@link PhysicalPlanNode} with {@code kafkaScan}. Accepted shapes:
+ * {@code Project[FFIReader]} and {@code Project[Filter[FFIReader]]} (the Calc always wraps its
+ * output in a projection).
+ */
+ private static PhysicalPlanNode spliceScanIntoLeaf(PhysicalPlanNode node, PhysicalPlanNode kafkaScan) {
+ return AuronPlanTreeRewriter.rewriteFfiReaderLeaf(
+ node,
+ ffiReaderLeaf -> kafkaScan,
+ "Merged Calc sub-plan expects Project[Filter?[FFIReader]] shape; got: ");
+ }
+
+ /**
+ * Prepends the three Kafka metadata passthrough columns (partition, offset, timestamp) to the
+ * outer {@link ProjectionExecNode}'s parallel {@code expr} / {@code expr_name} / {@code
+ * data_type} lists, so the fused output stays {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ */
+ private static PhysicalPlanNode prependMetadataPassthrough(PhysicalPlanNode merged) {
+ if (!merged.hasProjection()) {
+ throw new IllegalArgumentException(
+ "Merged Calc sub-plan expects a Projection root to prepend metadata passthrough; got: "
+ + merged.getPhysicalPlanTypeCase());
+ }
+ ProjectionExecNode projection = merged.getProjection();
+ List<PhysicalExprNode> originalExprs = new ArrayList<>(projection.getExprList());
+ List<String> originalNames = new ArrayList<>(projection.getExprNameList());
+ List<ArrowType> originalTypes = new ArrayList<>(projection.getDataTypeList());
+
+ for (RowType.RowField metaField : KAFKA_AURON_META_FIELDS) {
+ if (originalNames.contains(metaField.getName())) {
+ throw new IllegalArgumentException("Logical projected column name '" + metaField.getName()
+ + "' collides with a reserved Kafka metadata column; rename the column to fuse the Calc");
+ }
+ }
+
+ ProjectionExecNode.Builder rebuilt =
+ projection.toBuilder().clearExpr().clearExprName().clearDataType();
+ for (RowType.RowField metaField : KAFKA_AURON_META_FIELDS) {
+ addMetadataColumn(rebuilt, metaField.getName(), metaField.getType());
+ }
+ rebuilt.addAllExpr(originalExprs).addAllExprName(originalNames).addAllDataType(originalTypes);
+
+ return merged.toBuilder().setProjection(rebuilt.build()).build();
+ }
+
+ /**
+ * Resolves the negative (row-end-relative) index of the metadata column named {@code fieldName}.
+ * Metadata columns occupy the leading {@code metaCount} positions of every native output row, so
+ * a field at physical position {@code p} reads back at {@code p - metaCount}.
+ *
+ * @param fieldName the metadata column name, declared in {@link KafkaConstants#KAFKA_AURON_META_FIELDS}
+ * @param metaCount the number of leading metadata columns
+ * @return the negative index to pass to the row accessors for this metadata column
+ */
+ @VisibleForTesting
+ static int metaFieldNegativeIndex(String fieldName, int metaCount) {
+ for (int p = 0; p < KAFKA_AURON_META_FIELDS.size(); p++) {
+ if (KAFKA_AURON_META_FIELDS.get(p).getName().equals(fieldName)) {
+ return p - metaCount;
+ }
+ }
+ throw new IllegalStateException("Unknown Kafka metadata field: " + fieldName);
+ }
Review Comment:
Good catch. Removed the parameter in `aaddea35` rather than adding a precondition: `metaFieldNegativeIndex` now derives `metaCount` internally from `KAFKA_AURON_META_FIELDS.size()`. Its only caller was already passing exactly that value, and the method loops over the same list, so the parameter could only ever hold the field-list size. Dropping it removes the mismatch failure mode entirely instead of guarding against it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3412600017
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
Review Comment:
@Tartarus0zm @weiqingy I re-reviewed the current revision with this question in mind, and I come down on the graph-level approach too. Three things from the re-review tipped me:
**1. The interim delivers almost nothing under default config.** Fusion is gated to `isSourceFusionSafe` — both `table.optimizer.reuse-source-enabled` and `reuse-sub-plan-enabled` off — and both default to `true`. So out of the box nothing fuses; the IT cases have to disable reuse in `@BeforeEach` to exercise the path at all. And since this PR also removes the standalone `FlinkAuronCalcOperator` emission, under default config every convertible Calc now goes to Flink codegen rather than native. The "single-conversion win now" only materializes when a user opts into a non-default reuse config (which itself can regress other queries by un-sharing sources), so the default-config value is close to nil — which is most of the stepping-stone argument.
**2. The correctness gate infers a whole-graph property from session config, which is exactly what a node can't see.** `isSourceFusionSafe` uses "reuse disabled ⇒ one consumer per source" as a proxy for sole-consumership. That's sound for a freshly-planned SQL job but it isn't topology — a compiled plan (`@JsonCreator`/`COMPILE PLAN` is wired here) freezes source sharing at compile time, and restoring it under a different reuse config can re-permit fusion into a genuinely shared source, reopening the last-write-wins corruption the gate closed. Counting a source's real consumers needs the graph, so it belongs in #2329.
**3. Your maintenance question is the deciding one.** Once #2329 lands graph-level fusion, the in-`StreamExecCalc` reach-across (the `LegacySourceTransformation` unwrap, the reuse-config gate, the upstream re-typing) is all throwaway — and it lives in the shared Calc shadow every query flows through, carrying a behavior change (strict mode `FAIL_BACK=false` now silently falls a convertible-but-unfusible Calc back to Flink codegen instead of throwing). I'd rather not carry that.
So +1 to building #2329 graph-level from the start and treating #1865 as the single-operator case it subsumes; if #2329 covers single-operator fusion, folding #1865 into it as you suggest makes sense.
@weiqingy — to be clear, none of the implementation is wasted: the Rex→native converter wiring, the metadata-splice + by-name column resolution, and the no-watermark emit fix all carry straight over to the graph-level pass; it's really the *placement* and the config-inference gate that move. And both your pushbacks hold on re-check — there's genuinely no reverse Arrow→`LogicalType` converter so keeping `mergedProjectedOutputType` is right, and T2 does exercise the `hasWatermark()` gate (watermark pushed into the source, Calc's upstream stays a `LegacySourceTransformation`). One small follow-up either way: that test still passes with `hasWatermark()` deleted since `[22,23]` is the result fused or not, so a watermark-dependent assertion would actually lock the gate.
If the team would still rather land this as an interim, I'm not blocking — I'd just want it to (a) state explicitly that fusion is off under default reuse, (b) guard the compiled-plan restore case, and (c) preserve strict mode's native-or-throw contract.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "Tartarus0zm (via GitHub)" <gi...@apache.org>.
Tartarus0zm commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3402495170
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java:
##########
Review Comment:
From my gut feeling, doing the merge here isn't a good choice:
- Each StreamExec should only be responsible for translating its own operator — it shouldn't reach across operators.
- The way we're checking whether to merge operators is too hacky. I'd suggest getting the upstream operator and checking operator instanceof FlinkAuronOperator or function instanceof FlinkAuronFunction instead — that's much more reasonable.
Also, the ideal approach for multi-operator fusion would be to use a Rule/optimizer, but that's significantly more complex, and we'd also need to figure out how to inject that optimization rule into Flink.
So a relatively pragmatic middle ground is to handle this during JobGraph generation, similar to how Gluten does it — we'd need to rewrite Flink's StreamGraphTranslator.
What do you think? @weiqingy @SteNicholas
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on PR #2327:
URL: https://github.com/apache/auron/pull/2327#issuecomment-4988275939
@Tartarus0zm Thanks for the review! Sounds good, updated the names in `d891c9a9`:
`AuronSourceCalcFusionProcessor` → `AuronOperatorFusionProcessor`
`NativeCalcPlanBuilder` → `NativePlanFusionBuilder`
On your question about whether other operators would need a separate set of components: the processor's skeleton (walk the `ExecNodeGraph`, count consumers, gate, then stage the merged plan on the source) isn't tied to Calc, so a future operator can reuse it instead of cloning a parallel processor. I left the method names like `buildNativeCalcPlan` as they are, since those build the Calc's `Project[Filter]` shape; a new operator would just add a sibling build method there rather than a whole new class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3592577427
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java:
##########
@@ -505,6 +572,54 @@ public void setWatermarkStrategy(WatermarkStrategy<RowData> watermarkStrategy) {
this.watermarkStrategy = watermarkStrategy;
}
+ /**
+ * Registers a logical Calc sub-plan to fuse into the Kafka source so the native engine runs a
+ * single {@code Project[Filter?[KafkaScan]]} plan instead of a bare {@code KafkaScan} feeding a
+ * separate downstream Calc operator.
+ *
+ * <p>The sub-plan is a {@code Project[Filter?[FFIReader-placeholder]]} tree built by the
+ * downstream Calc against the LOGICAL-only input columns; its FFIReader leaf carries the
+ * resource ID {@code "placeholder"}. At {@code open()} the placeholder leaf is replaced by the
+ * freshly-built {@code KafkaScan} and the three Kafka metadata passthrough columns are prepended
+ * to the outer projection, so the fused output is {@code [partition, offset, timestamp,
+ * ...projected-logical]}.
+ *
+ * @param logicalCalcSubPlan the {@code Project[Filter?[FFIReader-placeholder]]} tree over the
+ * logical input columns
+ * @param projectedOutputType the projected logical output row type (without the metadata
+ * columns) emitted downstream when the fused plan is active
+ */
+ @Override
+ public void setMergedCalcPlan(PhysicalPlanNode logicalCalcSubPlan, RowType projectedOutputType) {
+ this.mergedCalcPlan = logicalCalcSubPlan;
+ this.mergedProjectedOutputType = projectedOutputType;
+ }
Review Comment:
setMergedCalcPlan() will silently overwrite an already-registered merged plan and also allows null inputs, but later code assumes the staged plan/output type are stable and non-null (e.g., effectiveLogicalOutputType()). This can lead to hard-to-debug plan corruption or NPEs if called twice or with nulls.
##########
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaDynamicTableSource.java:
##########
@@ -140,4 +169,20 @@ public String asSummaryString() {
public void applyWatermark(WatermarkStrategy<RowData> watermarkStrategy) {
this.watermarkStrategy = watermarkStrategy;
}
+
+ @Override
+ public void setMergedCalcPlan(PhysicalPlanNode logicalCalcSubPlan, RowType projectedOutputType) {
+ this.mergedCalcPlan = logicalCalcSubPlan;
+ this.mergedProjectedOutputType = projectedOutputType;
+ }
Review Comment:
setMergedCalcPlan() allows nulls and silently overwrites any previously staged merged plan, even though the fusion processor relies on staged plans being immutable once set (to prevent one Calc from discarding another's staged plan). Add defensive checks to fail fast if this is called twice or with null inputs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1865] Merge native operators: fuse Flink source and Calc into one native plan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #2327:
URL: https://github.com/apache/auron/pull/2327#discussion_r3592804473
##########
auron-flink-extension/auron-flink-planner/src/main/java/org/apache/auron/flink/table/planner/processor/AuronOperatorFusionProcessor.java:
##########
@@ -0,0 +1,292 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.auron.flink.table.planner.processor;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+import org.apache.auron.flink.connector.kafka.KafkaConstants;
+import org.apache.auron.flink.runtime.operator.FlinkAuronDynamicTableSource;
+import org.apache.auron.flink.table.planner.FlinkAuronCalcNode;
+import org.apache.auron.flink.table.planner.converter.NativePlanFusionBuilder;
+import org.apache.auron.protobuf.PhysicalPlanNode;
+import org.apache.calcite.rex.RexNode;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.planner.delegation.PlannerBase;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeGraph;
+import org.apache.flink.table.planner.plan.nodes.exec.processor.ExecNodeGraphProcessor;
+import org.apache.flink.table.planner.plan.nodes.exec.processor.ProcessorContext;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecTableSourceScan;
+import org.apache.flink.table.types.logical.RowType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Graph-level fusion pass that merges a native Auron source and the {@code StreamExecCalc} that
+ * directly consumes it into a single native {@code Project[Filter?[KafkaScan]]} plan, so the chain
+ * stays columnar end-to-end and pays a single columnar→row conversion at the tail.
+ *
+ * <p>The pass runs over the whole {@link ExecNodeGraph} before Transformation translation. For each
+ * {@code source → Calc} chain it decides whether to fuse and, when it does, stages the merged
+ * Calc sub-plan on the source's {@link FlinkAuronDynamicTableSource} instance (annotate-and-skip): it
+ * does not remove or re-type any node. The shadowed {@code StreamExecCalc} later observes the staged
+ * plan and emits no standalone operator for that Calc. This decision is graph-level because
+ * sole-consumership ("does this source feed exactly one consumer?") is a whole-graph property a single
+ * node cannot see in Flink's one-directional exec graph.
+ *
+ * <p>Fusion is additive on top of the standalone native Calc path: a convertible Calc the pass does
+ * not fuse still runs as a standalone native operator; only a non-convertible Calc falls back to
+ * Flink codegen.
+ *
+ * <p>A Calc is fused into its source only when all hold:
+ *
+ * <ul>
+ * <li>the Calc's single input is a {@link StreamExecTableSourceScan} whose {@link
+ * DynamicTableSource} is a {@link FlinkAuronDynamicTableSource};
+ * <li>that scan is the sole consumer of the source (consumer count {@code == 1} —
+ * multi-consumer fusion is tracked separately);
+ * <li>the source has no event-time watermark (fusing below a watermark generator would strip
+ * per-record event-time);
+ * <li>the Calc fully converts to a native plan;
+ * <li>no projection-output or source-input column name collides with a reserved Kafka metadata
+ * column (the native engine resolves columns by name, so a collision would silently bind to the
+ * metadata column).
+ * </ul>
+ */
+public class AuronOperatorFusionProcessor implements ExecNodeGraphProcessor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AuronOperatorFusionProcessor.class);
+
+ @Override
+ public ExecNodeGraph process(ExecNodeGraph graph, ProcessorContext context) {
+ final PlannerBase planner = context.getPlanner();
+ return process(
+ graph,
+ scan -> scan.getTableSourceSpec()
+ .getScanTableSource(planner.getFlinkContext(), planner.getTypeFactory()),
+ planner.getTableConfig());
+ }
+
+ /**
+ * Runs the fusion decision over the whole graph: builds the reverse fan-out consumer count,
+ * collects every {@code scan → Calc} chain, and applies {@link #tryFuse} to each. Split from
+ * the {@link ProcessorContext} entry point so the full count→gate→stage path is drivable
+ * with the scan's source and the table config supplied directly, without a live {@link
+ * PlannerBase}.
+ *
+ * @param graph the exec-node graph to process
+ * @param sourceResolver resolves a scan node to its {@link DynamicTableSource}
+ * @param tableConfig the table config used to seed the native plan converter
+ * @return the same graph instance (annotate-and-skip: only source instances are mutated)
+ */
+ ExecNodeGraph process(
+ ExecNodeGraph graph,
+ Function<StreamExecTableSourceScan, DynamicTableSource> sourceResolver,
+ ReadableConfig tableConfig) {
+ final Map<Integer, Integer> consumerCount = buildConsumerCount(graph);
+
+ for (Map.Entry<ExecNode<?>, ExecNode<?>> chain :
+ findSourceCalcChains(graph).entrySet()) {
+ final StreamExecTableSourceScan scan = (StreamExecTableSourceScan) chain.getKey();
+ final StreamExecCalc calc = (StreamExecCalc) chain.getValue();
+ tryFuse(scan, calc, consumerCount, sourceResolver, tableConfig);
+ }
+ // Annotate-and-skip: the graph is structurally unchanged; only source instances are mutated.
+ return graph;
+ }
+
+ /**
+ * Builds a reverse fan-out map keyed by stable {@link ExecNode#getId()}, counting consumers per
+ * distinct consuming edge with visited-node dedup across all roots. A node reachable from
+ * multiple roots is counted once per distinct consumer edge, not once per traversal path, so a
+ * genuinely shared source can never be miscounted as {@code count == 1}.
+ *
+ * @param graph the exec-node graph to traverse
+ * @return a map from node id to the number of distinct consumers of that node
+ */
+ static Map<Integer, Integer> buildConsumerCount(ExecNodeGraph graph) {
+ final Map<Integer, Integer> consumerCount = new HashMap<>();
+ final Set<Integer> visited = new HashSet<>();
+ // Each visited edge contributes exactly one consumer to its source. Dedup is on the visited
+ // SET of nodes (so each node's input edges are walked once across all roots); the consumer
+ // increment happens per input edge of every visited node, which is the per-consuming-edge
+ // count the design requires.
+ final Deque<ExecNode<?>> stack = new ArrayDeque<>(graph.getRootNodes());
+ while (!stack.isEmpty()) {
+ final ExecNode<?> node = stack.pop();
+ if (!visited.add(node.getId())) {
+ continue;
+ }
+ for (ExecEdge edge : node.getInputEdges()) {
+ final ExecNode<?> src = edge.getSource();
+ consumerCount.merge(src.getId(), 1, Integer::sum);
+ stack.push(src);
+ }
+ }
+ return consumerCount;
+ }
+
+ /**
+ * Collects every {@code scan → Calc} chain where a {@link StreamExecTableSourceScan} is the
+ * single input source of a {@link StreamExecCalc}. Keyed by the scan node so each scan maps to the
+ * Calc that directly consumes it.
+ *
+ * @param graph the exec-node graph to traverse
+ * @return a map from each consuming scan to its {@link StreamExecCalc}
+ */
+ private static Map<ExecNode<?>, ExecNode<?>> findSourceCalcChains(ExecNodeGraph graph) {
+ final Map<ExecNode<?>, ExecNode<?>> chains = new HashMap<>();
+ final Set<Integer> visited = new HashSet<>();
+ final Deque<ExecNode<?>> stack = new ArrayDeque<>(graph.getRootNodes());
+ while (!stack.isEmpty()) {
+ final ExecNode<?> node = stack.pop();
+ if (!visited.add(node.getId())) {
+ continue;
+ }
+ if (node instanceof StreamExecCalc && node.getInputEdges().size() == 1) {
+ final ExecNode<?> input = node.getInputEdges().get(0).getSource();
+ if (input instanceof StreamExecTableSourceScan) {
+ chains.put(input, node);
+ }
+ }
+ for (ExecEdge edge : node.getInputEdges()) {
+ stack.push(edge.getSource());
+ }
+ }
+ return chains;
+ }
+
+ /**
+ * Applies the full fusion gate to one {@code scan → Calc} chain and, when every gate passes,
+ * stages the merged Calc plan on the source instance. Any failed gate is a silent no-fusion: the
+ * Calc then runs standalone native (if convertible) or Flink codegen.
+ *
+ * @param scan the source scan node
+ * @param calc the Calc node directly consuming the scan
+ * @param consumerCount the reverse fan-out map from {@link #buildConsumerCount}
+ * @param sourceResolver resolves the scan to its {@link DynamicTableSource}
+ * @param tableConfig the table config used to seed the native plan converter
+ */
+ void tryFuse(
+ StreamExecTableSourceScan scan,
+ StreamExecCalc calc,
+ Map<Integer, Integer> consumerCount,
+ Function<StreamExecTableSourceScan, DynamicTableSource> sourceResolver,
+ ReadableConfig tableConfig) {
+ // Sole-consumer gate: multi-consumer fusion is out of scope (tracked separately).
+ if (consumerCount.getOrDefault(scan.getId(), 0) != 1) {
+ return;
+ }
+
+ final DynamicTableSource tableSource = sourceResolver.apply(scan);
+
+ final FlinkAuronCalcNode calcNode = (FlinkAuronCalcNode) calc;
+ stagePlanIfFusible(
+ tableSource,
+ scan.getId(),
+ (RowType) calc.getInputEdges().get(0).getOutputType(),
+ (RowType) calc.getOutputType(),
+ calcNode.getProjection(),
+ calcNode.getCondition(),
+ tableConfig);
Review Comment:
Added an `instanceof FlinkAuronCalcNode` guard in `tryFuse` (`1ac7fe57`): if the classpath ever resolves to Flink's stock `StreamExecCalc` instead of Auron's shadowed one, fusion now logs a warning and skips rather than throwing. That also makes the cast honor `tryFuse`'s documented "silent no-fusion" contract.
For context: the shadowed `StreamExecCalc` always implements `FlinkAuronCalcNode` (declared that way, and shipped in the same assembly overlay that excludes the stock class), so in a correctly built jar the cast can't fail; the guard covers the misconfigured-classpath edge you flagged. I didn't add a unit test since the in-module test classpath only has the shadowed `StreamExecCalc`, so a stock instance can't be constructed to exercise the branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org