This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 8e9e69cb0ed [FLINK-39986][table-planner] Support Python UDF
deduplication in projection and condition (#28638)
8e9e69cb0ed is described below
commit 8e9e69cb0ed430fbce873ea0e6ab6afdbad85421
Author: RaoraoXiong <[email protected]>
AuthorDate: Fri Aug 28 11:14:24 2026 +0800
[FLINK-39986][table-planner] Support Python UDF deduplication in projection
and condition (#28638)
The duplication originates in the planner: RexProgram already shares
structurally identical expressions through RexLocalRef, but the Python calc
translation calls RexProgram#expandLocalRef, which expands the shared
reference back into independent expression trees.
This is now addressed by two logical optimizer rules, so that neither
CommonExecPythonCalc nor the projection codegen needs to be aware of it:
- RemoteCalcConditionProjectionCseRule rewrites a Calc projection to
reference UDF calls already computed by the Calc below it, which removes
duplicates shared between a WHERE condition and a SELECT projection.
- RemoteCalcProjectionCseRule splits a Calc containing duplicated
deterministic calls into a bottom Calc computing each distinct call once
plus a top Calc projecting the shared results back into their original
positions.
Both rules share their reusability predicate through RemoteCalcCseUtil so
they agree on which calls may safely share a single evaluation.
Non-deterministic calls are always evaluated independently.
Generated-by: Claude-4.6-Opus
---
.../planner/plan/rules/FlinkBatchRuleSets.scala | 2 +
.../planner/plan/rules/FlinkStreamRuleSets.scala | 4 +
.../plan/rules/logical/PythonCalcSplitRule.scala | 9 +-
.../RemoteCalcConditionProjectionCseRule.java | 252 +++++++++++++++++++++
.../plan/rules/logical/RemoteCalcCseUtil.java | 125 ++++++++++
.../rules/logical/RemoteCalcProjectionCseRule.java | 237 +++++++++++++++++++
.../planner/plan/stream/sql/PythonCalcCseTest.java | 126 +++++++++++
.../utils/JavaUserDefinedScalarFunctions.java | 34 +++
.../planner/plan/stream/sql/PythonCalcCseTest.xml | 208 +++++++++++++++++
9 files changed, 995 insertions(+), 2 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
index 26553f24ddc..780041f862c 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
@@ -391,11 +391,13 @@ object FlinkBatchRuleSets {
PythonCalcSplitRule.SPLIT_CONDITION_REX_FIELD,
PythonCalcSplitRule.SPLIT_PROJECTION_REX_FIELD,
PythonCalcSplitRule.SPLIT_CONDITION,
+ PythonCalcSplitRule.CONDITION_PROJECTION_CSE,
PythonCalcSplitRule.SPLIT_PROJECT,
PythonCalcSplitRule.SPLIT_PANDAS_IN_PROJECT,
PythonCalcSplitRule.EXPAND_PROJECT,
PythonCalcSplitRule.PUSH_CONDITION,
PythonCalcSplitRule.REWRITE_PROJECT,
+ PythonCalcSplitRule.PROJECTION_CSE,
PythonMapRenameRule.INSTANCE,
PythonMapMergeRule.INSTANCE,
AsyncCalcSplitRule.SPLIT_CONDITION_REX_FIELD,
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
index de637d07a24..ed5327e49fa 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
@@ -415,6 +415,8 @@ object FlinkStreamRuleSets {
PythonCalcSplitRule.SPLIT_PROJECTION_REX_FIELD,
// Avoids dealing with a python call in the condition.
PythonCalcSplitRule.SPLIT_CONDITION,
+ // Deduplicates Python UDF calls shared between condition and projection.
+ PythonCalcSplitRule.CONDITION_PROJECTION_CSE,
// Avoids dealing with Java calls in the same Calc as python calls.
PythonCalcSplitRule.SPLIT_PROJECT,
// Splits calcs which contain both general Python functions and pandas
Python functions
@@ -425,6 +427,8 @@ object FlinkStreamRuleSets {
PythonCalcSplitRule.PUSH_CONDITION,
// Orders the projections so that input references are first, followed by
python calls.
PythonCalcSplitRule.REWRITE_PROJECT,
+ // Deduplicates identical deterministic python calls within the projection.
+ PythonCalcSplitRule.PROJECTION_CSE,
// Renames the field names of the Flatten calc which is right after a calc
representing a
// Python Map operation to the output names of the map function
PythonMapRenameRule.INSTANCE,
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala
index 7e54ffbca03..642ca5f5691 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala
@@ -96,11 +96,14 @@ class PythonRemoteCallFinder extends RemoteCallFinder {
object PythonCalcSplitRule {
/**
- * These rules should be applied sequentially in the order of
SPLIT_CONDITION, SPLIT_PROJECT,
- * SPLIT_PANDAS_IN_PROJECT, EXPAND_PROJECT, PUSH_CONDITION and
REWRITE_PROJECT.
+ * These rules should be applied sequentially in the order of
SPLIT_CONDITION,
+ * CONDITION_PROJECTION_CSE, SPLIT_PROJECT, SPLIT_PANDAS_IN_PROJECT,
EXPAND_PROJECT,
+ * PUSH_CONDITION, REWRITE_PROJECT and PROJECTION_CSE.
*/
private val callFinder = new PythonRemoteCallFinder()
val SPLIT_CONDITION: RelOptRule = new
RemoteCalcSplitConditionRule(callFinder)
+ val CONDITION_PROJECTION_CSE: RelOptRule =
+
RemoteCalcConditionProjectionCseRule.Config.DEFAULT.withRemoteCallFinder(callFinder).toRule()
val SPLIT_PROJECT: RelOptRule = new RemoteCalcSplitProjectionRule(callFinder)
val SPLIT_PANDAS_IN_PROJECT: RelOptRule = new
PythonCalcSplitPandasInProjectionRule(callFinder)
val SPLIT_PROJECTION_REX_FIELD: RelOptRule = new
RemoteCalcSplitProjectionRexFieldRule(callFinder)
@@ -108,4 +111,6 @@ object PythonCalcSplitRule {
val EXPAND_PROJECT: RelOptRule = new RemoteCalcExpandProjectRule(callFinder)
val PUSH_CONDITION: RelOptRule = new RemoteCalcPushConditionRule(callFinder)
val REWRITE_PROJECT: RelOptRule = new
RemoteCalcRewriteProjectionRule(callFinder)
+ val PROJECTION_CSE: RelOptRule =
+
RemoteCalcProjectionCseRule.Config.DEFAULT.withRemoteCallFinder(callFinder).toRule()
}
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcConditionProjectionCseRule.java
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcConditionProjectionCseRule.java
new file mode 100644
index 00000000000..a1f9b328e48
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcConditionProjectionCseRule.java
@@ -0,0 +1,252 @@
+/*
+ * 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.flink.table.planner.plan.rules.logical;
+
+import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalCalc;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexShuttle;
+import org.immutables.value.Value;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Rule that eliminates common remote (e.g. Python or async) UDF
sub-expressions between the
+ * condition and projection of a Calc node.
+ *
+ * <p>After {@link RemoteCalcSplitConditionRule} splits a Calc with remote
UDFs in its condition, the
+ * result is a two-level Calc structure:
+ *
+ * <pre>
+ * TopCalc(projection=[remoteFunc(a, b) + 1, remoteFunc(a, b) + 2],
condition=[$2 > 0])
+ * BottomCalc(projection=[a, b, remoteFunc(a, b) AS f0])
+ * </pre>
+ *
+ * <p>The TopCalc's projection still contains {@code remoteFunc(a, b)} which
is structurally
+ * identical to the already-computed {@code f0} in the BottomCalc. This rule
detects such duplicates
+ * and rewrites the TopCalc's projection to reference the BottomCalc's output
directly:
+ *
+ * <pre>
+ * TopCalc(projection=[$2 + 1, $2 + 2], condition=[$2 > 0])
+ * BottomCalc(projection=[a, b, remoteFunc(a, b) AS f0])
+ * </pre>
+ *
+ * <p>Note that the two Calcs do not share a coordinate system: the
BottomCalc's expressions are
+ * written against its own input, while the TopCalc's are written against the
BottomCalc's output.
+ * The BottomCalc's calls are therefore translated into the TopCalc's frame of
reference before
+ * being compared, so that calls which merely look alike are not treated as
equal.
+ */
[email protected]
+public class RemoteCalcConditionProjectionCseRule
+ extends RelRule<RemoteCalcConditionProjectionCseRule.Config> {
+
+ protected RemoteCalcConditionProjectionCseRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ FlinkLogicalCalc topCalc = call.rel(0);
+ FlinkLogicalCalc bottomCalc = call.rel(1);
+ RemoteCallFinder callFinder = config.remoteCallFinder();
+
+ // Only applies when the top calc has a condition.
+ if (topCalc.getProgram().getCondition() == null) {
+ return false;
+ }
+
+ Map<RexNode, Integer> bottomRemoteCalls =
buildBottomRemoteCallMap(bottomCalc, callFinder);
+ if (bottomRemoteCalls.isEmpty()) {
+ return false;
+ }
+
+ List<RexNode> topProjects = RemoteCalcCseUtil.expandProjects(topCalc);
+ return topProjects.stream()
+ .anyMatch(node -> containsCallMatchingBottom(node,
bottomRemoteCalls, callFinder));
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ FlinkLogicalCalc topCalc = call.rel(0);
+ FlinkLogicalCalc bottomCalc = call.rel(1);
+ RemoteCallFinder callFinder = config.remoteCallFinder();
+ RexBuilder rexBuilder = call.builder().getRexBuilder();
+
+ List<RexNode> topProjects = RemoteCalcCseUtil.expandProjects(topCalc);
+ RexNode topCondition =
+
topCalc.getProgram().expandLocalRef(topCalc.getProgram().getCondition());
+
+ Map<RexNode, Integer> bottomRemoteCalls =
buildBottomRemoteCallMap(bottomCalc, callFinder);
+ RelDataType bottomRowType = bottomCalc.getRowType();
+
+ // Rewrite top projections: replace matching calls with RexInputRef.
+ CseRewriteShuttle rewriter = new CseRewriteShuttle(bottomRemoteCalls,
bottomRowType);
+ List<RexNode> newTopProjects =
+ topProjects.stream().map(p ->
p.accept(rewriter)).collect(Collectors.toList());
+
+ if (!rewriter.hasRewritten()) {
+ return;
+ }
+
+ // Build the new top calc with rewritten projections.
+ call.transformTo(
+ topCalc.copy(
+ topCalc.getTraitSet(),
+ bottomCalc,
+ RexProgram.create(
+ bottomRowType,
+ newTopProjects,
+ topCondition,
+ topCalc.getRowType(),
+ rexBuilder)));
+ }
+
+ /**
+ * Builds a map from the bottom calc's reusable remote calls to their
output index, with each
+ * call rewritten into the top calc's frame of reference.
+ *
+ * <p>The bottom calc's expressions are written against its own input,
whereas the top calc
+ * addresses the bottom calc's output, so the two must be brought into a
common frame before
+ * being compared. See {@link RemoteCalcCseUtil#translateToOutputFrame}.
+ */
+ private static Map<RexNode, Integer> buildBottomRemoteCallMap(
+ FlinkLogicalCalc bottomCalc, RemoteCallFinder callFinder) {
+ List<RexNode> bottomProjects =
RemoteCalcCseUtil.expandProjects(bottomCalc);
+ Map<Integer, Integer> forwardedFieldPositions =
+ RemoteCalcCseUtil.forwardedFieldPositions(bottomProjects);
+
+ Map<RexNode, Integer> result = new HashMap<>();
+ for (int i = 0; i < bottomProjects.size(); i++) {
+ RexNode project = bottomProjects.get(i);
+ if (!RemoteCalcCseUtil.containsReusableRemoteCall(project,
callFinder)) {
+ continue;
+ }
+ RexNode translated =
+ RemoteCalcCseUtil.translateToOutputFrame(project,
forwardedFieldPositions);
+ if (translated != null) {
+ result.put(translated, i);
+ }
+ }
+ return result;
+ }
+
+ private static boolean containsCallMatchingBottom(
+ RexNode node, Map<RexNode, Integer> bottomRemoteCalls,
RemoteCallFinder callFinder) {
+ if (node instanceof RexCall) {
+ RexCall rexCall = (RexCall) node;
+ if (callFinder.isRemoteCall(rexCall) &&
bottomRemoteCalls.containsKey(rexCall)) {
+ return true;
+ }
+ return rexCall.getOperands().stream()
+ .anyMatch(op -> containsCallMatchingBottom(op,
bottomRemoteCalls, callFinder));
+ }
+ return false;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof RemoteCalcConditionProjectionCseRule)) {
+ return false;
+ }
+ RemoteCalcConditionProjectionCseRule other =
(RemoteCalcConditionProjectionCseRule) obj;
+ return super.equals(other)
+ && config.remoteCallFinder()
+ .getClass()
+ .equals(other.config.remoteCallFinder().getClass());
+ }
+
+ @Override
+ public int hashCode() {
+ return super.hashCode() * 31 +
config.remoteCallFinder().getClass().hashCode();
+ }
+
+ //
-------------------------------------------------------------------------
+
+ /**
+ * Replaces remote UDF calls in the top calc's projection with a
RexInputRef pointing at the
+ * bottom calc's output position where the same call was already computed.
+ */
+ private static class CseRewriteShuttle extends RexShuttle {
+ private final Map<RexNode, Integer> bottomRemoteCalls;
+ private final RelDataType bottomRowType;
+ private boolean rewritten = false;
+
+ CseRewriteShuttle(Map<RexNode, Integer> bottomRemoteCalls, RelDataType
bottomRowType) {
+ this.bottomRemoteCalls = bottomRemoteCalls;
+ this.bottomRowType = bottomRowType;
+ }
+
+ boolean hasRewritten() {
+ return rewritten;
+ }
+
+ @Override
+ public RexNode visitCall(RexCall call) {
+ Integer idx = bottomRemoteCalls.get(call);
+ if (idx != null) {
+ rewritten = true;
+ return new RexInputRef(idx,
bottomRowType.getFieldList().get(idx).getType());
+ }
+ return super.visitCall(call);
+ }
+ }
+
+ //
-------------------------------------------------------------------------
+
+ /** Rule configuration. */
+ @Value.Immutable(singleton = false)
+ public interface Config extends RelRule.Config {
+ Config DEFAULT =
+ ImmutableRemoteCalcConditionProjectionCseRule.Config.builder()
+ .operandSupplier(
+ b0 ->
+ b0.operand(FlinkLogicalCalc.class)
+ .oneInput(
+ b1 ->
+
b1.operand(FlinkLogicalCalc.class)
+
.anyInputs()))
+ .description("RemoteCalcConditionProjectionCseRule")
+ .build();
+
+ @Value.Default
+ default RemoteCallFinder remoteCallFinder() {
+ return new PythonRemoteCallFinder();
+ }
+
+ /** Sets {@link #remoteCallFinder()}. */
+ Config withRemoteCallFinder(RemoteCallFinder callFinder);
+
+ @Override
+ default RemoteCalcConditionProjectionCseRule toRule() {
+ return new RemoteCalcConditionProjectionCseRule(this);
+ }
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcCseUtil.java
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcCseUtil.java
new file mode 100644
index 00000000000..e6bddec8c24
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcCseUtil.java
@@ -0,0 +1,125 @@
+/*
+ * 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.flink.table.planner.plan.rules.logical;
+
+import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalCalc;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexShuttle;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Shared helpers for the remote (e.g. Python) call common sub-expression
elimination rules, namely
+ * {@link RemoteCalcProjectionCseRule} and {@link
RemoteCalcConditionProjectionCseRule}.
+ *
+ * <p>Keeping the reusability predicate in one place ensures both rules agree
on which calls may
+ * safely share a single evaluation.
+ */
+class RemoteCalcCseUtil {
+
+ private RemoteCalcCseUtil() {}
+
+ /** Expands the local refs of a calc's projection into self-contained
expression trees. */
+ static List<RexNode> expandProjects(FlinkLogicalCalc calc) {
+ RexProgram program = calc.getProgram();
+ return program.getProjectList().stream()
+ .map(program::expandLocalRef)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Returns true if the node is itself a remote call whose result may be
reused. A
+ * non-deterministic call must be evaluated once per occurrence, so it
never qualifies.
+ */
+ static boolean isReusableRemoteCall(RexNode node, RemoteCallFinder
callFinder) {
+ return callFinder.isRemoteCall(node) && isReusable(node);
+ }
+
+ /**
+ * Returns true if the node contains a remote call and its result may be
reused. Unlike {@link
+ * #isReusableRemoteCall}, the remote call may be nested inside the
expression tree.
+ */
+ static boolean containsReusableRemoteCall(RexNode node, RemoteCallFinder
callFinder) {
+ return callFinder.containsRemoteCall(node) && isReusable(node);
+ }
+
+ /**
+ * Maps each field a calc forwards unchanged to the output position at
which it is exposed.
+ *
+ * <p>Only plain {@link RexInputRef} projections are forwarding; the first
occurrence wins when a
+ * field is projected more than once.
+ */
+ static Map<Integer, Integer> forwardedFieldPositions(List<RexNode>
projects) {
+ Map<Integer, Integer> positions = new HashMap<>();
+ for (int i = 0; i < projects.size(); i++) {
+ RexNode project = projects.get(i);
+ if (project instanceof RexInputRef) {
+ positions.putIfAbsent(((RexInputRef) project).getIndex(), i);
+ }
+ }
+ return positions;
+ }
+
+ /**
+ * Rewrites an expression written against a calc's input so that it is
written against that
+ * calc's output instead, or returns {@code null} if it reads a field the
calc does not forward.
+ *
+ * <p>Splitting a Calc changes what an input ref means: the lower Calc
addresses the original
+ * input while the upper one addresses the lower Calc's output, and the
lower Calc forwards only
+ * the fields still needed above. Two expressions that print identically
may therefore read
+ * different columns, so they must be brought into a common frame of
reference before being
+ * compared for reuse. A {@code null} result means no valid comparison
exists, because the upper
+ * Calc cannot express the expression at all.
+ */
+ static RexNode translateToOutputFrame(
+ RexNode node, Map<Integer, Integer> forwardedFieldPositions) {
+ try {
+ return node.accept(
+ new RexShuttle() {
+ @Override
+ public RexNode visitInputRef(RexInputRef inputRef) {
+ Integer position =
forwardedFieldPositions.get(inputRef.getIndex());
+ if (position == null) {
+ throw new NotForwardedException();
+ }
+ return new RexInputRef(position,
inputRef.getType());
+ }
+ });
+ } catch (NotForwardedException e) {
+ return null;
+ }
+ }
+
+ private static boolean isReusable(RexNode node) {
+ return ShortcutUtils.isDeterministicThroughProgram(node, null);
+ }
+
+ /** Signals that an expression reads a field which the calc does not
forward. */
+ private static class NotForwardedException extends RuntimeException {
+ private NotForwardedException() {
+ super(null, null, false, false);
+ }
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcProjectionCseRule.java
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcProjectionCseRule.java
new file mode 100644
index 00000000000..547ced1819c
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcProjectionCseRule.java
@@ -0,0 +1,237 @@
+/*
+ * 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.flink.table.planner.plan.rules.logical;
+
+import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalCalc;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.sql.validate.SqlValidatorUtil;
+import org.immutables.value.Value;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Rule that deduplicates identical remote (e.g. Python) calls repeated in the
projection of a {@link
+ * FlinkLogicalCalc}.
+ *
+ * <p>The underlying {@link RexProgram} already shares structurally identical
expressions through
+ * {@link org.apache.calcite.rex.RexLocalRef}s. However, the remote calc
translation expands those
+ * local refs into independent expression trees, which re-introduces the
duplication and makes the
+ * same UDF be shipped to the remote worker once per occurrence. This rule
makes the sharing explicit
+ * in the plan by splitting the calc into two:
+ *
+ * <pre>
+ * Calc(projection=[pyFunc(a, b), pyFunc(a, b)])
+ * </pre>
+ *
+ * <p>becomes
+ *
+ * <pre>
+ * TopCalc(projection=[$0, $0])
+ * BottomCalc(projection=[pyFunc(a, b) AS f0])
+ * </pre>
+ *
+ * <p>The bottom calc keeps one occurrence of every distinct remote call, and
the top calc is a pure
+ * {@link RexInputRef} projection restoring the original output schema.
+ *
+ * <p>Only deterministic calls are deduplicated; a non-deterministic call must
be evaluated
+ * independently for each occurrence.
+ */
[email protected]
+public class RemoteCalcProjectionCseRule extends
RelRule<RemoteCalcProjectionCseRule.Config> {
+
+ protected RemoteCalcProjectionCseRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ FlinkLogicalCalc calc = call.rel(0);
+ RexProgram program = calc.getProgram();
+
+ // Conditions are pushed away by RemoteCalcPushConditionRule
beforehand, and duplicates
+ // shared with a condition are handled by
RemoteCalcConditionProjectionCseRule.
+ if (program.getCondition() != null) {
+ return false;
+ }
+
+ List<RexNode> projects = RemoteCalcCseUtil.expandProjects(calc);
+ RemoteCallFinder callFinder = config.remoteCallFinder();
+
+ // Only a projection already normalized by
RemoteCalcRewriteProjectionRule is handled, i.e.
+ // it consists of plain input refs and top-level remote calls only.
+ if (projects.stream().noneMatch(callFinder::isRemoteCall)) {
+ return false;
+ }
+ if (!projects.stream()
+ .allMatch(p -> p instanceof RexInputRef ||
callFinder.isRemoteCall(p))) {
+ return false;
+ }
+
+ return findDuplicates(projects, callFinder) != null;
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ FlinkLogicalCalc calc = call.rel(0);
+ RelNode input = calc.getInput();
+ RexBuilder rexBuilder = call.builder().getRexBuilder();
+ RemoteCallFinder callFinder = config.remoteCallFinder();
+
+ List<RexNode> projects = RemoteCalcCseUtil.expandProjects(calc);
+ int[] originalToUnique = findDuplicates(projects, callFinder);
+ if (originalToUnique == null) {
+ return;
+ }
+
+ // The bottom calc keeps the first occurrence of every distinct
expression.
+ List<RexNode> bottomProjects = new ArrayList<>();
+ Map<Integer, Integer> uniqueIndexToBottomIndex = new LinkedHashMap<>();
+ for (int i = 0; i < projects.size(); i++) {
+ if (originalToUnique[i] == i) {
+ uniqueIndexToBottomIndex.put(i, bottomProjects.size());
+ bottomProjects.add(projects.get(i));
+ }
+ }
+
+ List<String> bottomFieldNames =
+ SqlValidatorUtil.uniquify(
+ java.util.stream.IntStream.range(0,
bottomProjects.size())
+ .mapToObj(i -> "f" + i)
+ .collect(Collectors.toList()),
+
rexBuilder.getTypeFactory().getTypeSystem().isSchemaCaseSensitive());
+
+ FlinkLogicalCalc bottomCalc =
+ new FlinkLogicalCalc(
+ calc.getCluster(),
+ calc.getTraitSet(),
+ input,
+ RexProgram.create(
+ input.getRowType(),
+ bottomProjects,
+ null,
+ bottomFieldNames,
+ rexBuilder));
+
+ // The top calc only forwards the shared results back to their
original positions.
+ RelDataType bottomRowType = bottomCalc.getRowType();
+ List<RexNode> topProjects = new ArrayList<>();
+ for (int i = 0; i < projects.size(); i++) {
+ int bottomIndex =
uniqueIndexToBottomIndex.get(originalToUnique[i]);
+ topProjects.add(
+ new RexInputRef(
+ bottomIndex,
bottomRowType.getFieldList().get(bottomIndex).getType()));
+ }
+
+ call.transformTo(
+ calc.copy(
+ calc.getTraitSet(),
+ bottomCalc,
+ RexProgram.create(
+ bottomRowType,
+ topProjects,
+ null,
+ calc.getRowType(),
+ rexBuilder)));
+ }
+
+ /**
+ * Maps every projection index to the index of the first projection
computing the same value.
+ *
+ * @return the mapping, or {@code null} when there is nothing to
deduplicate
+ */
+ private int[] findDuplicates(List<RexNode> projects, RemoteCallFinder
callFinder) {
+ Map<RexNode, Integer> firstOccurrence = new LinkedHashMap<>();
+ int[] originalToUnique = new int[projects.size()];
+ boolean hasDuplicate = false;
+
+ for (int i = 0; i < projects.size(); i++) {
+ RexNode project = projects.get(i);
+ // Forwarded fields are cheap and already shared, so only remote
calls are considered.
+ boolean canReuse = RemoteCalcCseUtil.isReusableRemoteCall(project,
callFinder);
+ Integer existing = canReuse ? firstOccurrence.get(project) : null;
+ if (existing != null) {
+ originalToUnique[i] = existing;
+ hasDuplicate = true;
+ } else {
+ if (canReuse) {
+ firstOccurrence.put(project, i);
+ }
+ originalToUnique[i] = i;
+ }
+ }
+
+ return hasDuplicate ? originalToUnique : null;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof RemoteCalcProjectionCseRule)) {
+ return false;
+ }
+ RemoteCalcProjectionCseRule other = (RemoteCalcProjectionCseRule) obj;
+ return super.equals(other)
+ && config.remoteCallFinder()
+ .getClass()
+ .equals(other.config.remoteCallFinder().getClass());
+ }
+
+ @Override
+ public int hashCode() {
+ return super.hashCode() * 31 +
config.remoteCallFinder().getClass().hashCode();
+ }
+
+ //
-------------------------------------------------------------------------
+
+ /** Rule configuration. */
+ @Value.Immutable(singleton = false)
+ public interface Config extends RelRule.Config {
+ Config DEFAULT =
+ ImmutableRemoteCalcProjectionCseRule.Config.builder()
+ .operandSupplier(b0 ->
b0.operand(FlinkLogicalCalc.class).anyInputs())
+ .description("RemoteCalcProjectionCseRule")
+ .build();
+
+ @Value.Default
+ default RemoteCallFinder remoteCallFinder() {
+ return new PythonRemoteCallFinder();
+ }
+
+ /** Sets {@link #remoteCallFinder()}. */
+ Config withRemoteCallFinder(RemoteCallFinder callFinder);
+
+ @Override
+ default RemoteCalcProjectionCseRule toRule() {
+ return new RemoteCalcProjectionCseRule(this);
+ }
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/PythonCalcCseTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/PythonCalcCseTest.java
new file mode 100644
index 00000000000..15b7a92ea81
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/PythonCalcCseTest.java
@@ -0,0 +1,126 @@
+/*
+ * 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.flink.table.planner.plan.stream.sql;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import
org.apache.flink.table.planner.runtime.utils.JavaUserDefinedScalarFunctions.NonDeterministicPythonScalarFunction;
+import
org.apache.flink.table.planner.runtime.utils.JavaUserDefinedScalarFunctions.PythonScalarFunction;
+import org.apache.flink.table.planner.utils.JavaStreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for Python UDF common sub-expression elimination in the planner,
covering {@link
+ *
org.apache.flink.table.planner.plan.rules.logical.RemoteCalcProjectionCseRule}
(duplicates within
+ * a projection) and {@link
+ *
org.apache.flink.table.planner.plan.rules.logical.RemoteCalcConditionProjectionCseRule}
+ * (duplicates shared between a condition and a projection).
+ */
+class PythonCalcCseTest extends TableTestBase {
+
+ private final JavaStreamTableTestUtil util = javaStreamTestUtil();
+
+ @BeforeEach
+ void setup() {
+ util.addTableSource(
+ "MyTable",
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("b", DataTypes.INT())
+ .column("c", DataTypes.INT())
+ .build());
+ util.tableEnv()
+ .createTemporarySystemFunction("pyFunc1", new
PythonScalarFunction("pyFunc1"));
+ util.tableEnv()
+ .createTemporarySystemFunction("pyFunc2", new
PythonScalarFunction("pyFunc2"));
+ util.tableEnv()
+ .createTemporarySystemFunction(
+ "pyFuncNonDet", new
NonDeterministicPythonScalarFunction("pyFuncNonDet"));
+ }
+
+ //
-------------------------------------------------------------------------
+ // Duplicates within a projection
+ //
-------------------------------------------------------------------------
+
+ @Test
+ void testDuplicatedCallsInProjection() {
+ util.verifyExecPlan("SELECT pyFunc1(a, b), pyFunc1(a, b), pyFunc1(a,
b) FROM MyTable");
+ }
+
+ @Test
+ void testDuplicatedCallsWithForwardedField() {
+ util.verifyExecPlan("SELECT a, pyFunc1(a, b), pyFunc1(a, b) FROM
MyTable");
+ }
+
+ @Test
+ void testDistinctCallsAreNotDeduplicated() {
+ util.verifyExecPlan("SELECT pyFunc1(a, b), pyFunc1(b, a), pyFunc2(a,
b) FROM MyTable");
+ }
+
+ @Test
+ void testNonDeterministicCallsAreNotDeduplicated() {
+ util.verifyExecPlan("SELECT pyFuncNonDet(a, b), pyFuncNonDet(a, b)
FROM MyTable");
+ }
+
+ //
-------------------------------------------------------------------------
+ // Duplicates shared between a condition and a projection
+ //
-------------------------------------------------------------------------
+
+ @Test
+ void testSameUdfInConditionAndProjection() {
+ util.verifyExecPlan(
+ "SELECT pyFunc1(a, b) + 1, pyFunc1(a, b) + 2 FROM MyTable
WHERE pyFunc1(a, b) > 0");
+ }
+
+ @Test
+ void testDifferentUdfInConditionAndProjection() {
+ util.verifyExecPlan("SELECT pyFunc1(a, b) FROM MyTable WHERE
pyFunc2(a, c) > 0");
+ }
+
+ @Test
+ void testNestedUdfInProjectionWithSameInCondition() {
+ util.verifyExecPlan(
+ "SELECT pyFunc2(pyFunc1(a, b), c), pyFunc1(a, b) FROM MyTable
WHERE pyFunc1(a, b) > 0");
+ }
+
+ @Test
+ void testDuplicatedCallsInProjectionWithCondition() {
+ util.verifyExecPlan(
+ "SELECT pyFunc1(a, b), pyFunc1(a, b) FROM MyTable WHERE
pyFunc1(a, b) > 0");
+ }
+
+ /**
+ * The same function with different arguments must not be shared. After
the condition split both
+ * calls can be printed as {@code pyFunc1($0, $1)} even though they read
different columns, so
+ * comparing them without accounting for the frame of reference would
silently return the
+ * condition's result.
+ */
+ @Test
+ void testSameUdfWithDifferentArgsInConditionAndProjection() {
+ util.verifyExecPlan("SELECT pyFunc1(b, c) FROM MyTable WHERE
pyFunc1(a, b) > 0");
+ }
+
+ @Test
+ void testSameUdfWithSwappedArgsInConditionAndProjection() {
+ util.verifyExecPlan("SELECT pyFunc1(b, a) FROM MyTable WHERE
pyFunc1(a, b) > 0");
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedScalarFunctions.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedScalarFunctions.java
index dfd11550fb3..0b93bd42882 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedScalarFunctions.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedScalarFunctions.java
@@ -175,6 +175,40 @@ public class JavaUserDefinedScalarFunctions {
}
}
+ /** Test for non-deterministic Python Scalar Function. */
+ public static class NonDeterministicPythonScalarFunction extends
ScalarFunction
+ implements PythonFunction {
+ private final String name;
+
+ public NonDeterministicPythonScalarFunction(String name) {
+ this.name = name;
+ }
+
+ public int eval(Integer i, Integer j) {
+ return i + j;
+ }
+
+ @Override
+ public boolean isDeterministic() {
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ return name;
+ }
+
+ @Override
+ public byte[] getSerializedPythonFunction() {
+ return new byte[0];
+ }
+
+ @Override
+ public PythonEnv getPythonEnv() {
+ return new PythonEnv(PythonEnv.ExecType.PROCESS);
+ }
+ }
+
/** Test for Python Scalar Function. */
public static class BooleanPythonScalarFunction extends ScalarFunction
implements PythonFunction {
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/PythonCalcCseTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/PythonCalcCseTest.xml
new file mode 100644
index 00000000000..3f3f708fae0
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/PythonCalcCseTest.xml
@@ -0,0 +1,208 @@
+<?xml version="1.0" ?>
+<!--
+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.
+-->
+<Root>
+ <TestCase name="testDifferentUdfInConditionAndProjection">
+ <Resource name="sql">
+ <![CDATA[SELECT pyFunc1(a, b) FROM MyTable WHERE pyFunc2(a, c) > 0]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[pyFunc1($0, $1)])
++- LogicalFilter(condition=[>(pyFunc2($0, $2), 0)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+PythonCalc(select=[pyFunc1(a, b) AS EXPR$0])
++- Calc(select=[a, b], where=[(f0 > 0)])
+ +- PythonCalc(select=[a, b, pyFunc2(a, c) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testDistinctCallsAreNotDeduplicated">
+ <Resource name="sql">
+ <![CDATA[SELECT pyFunc1(a, b), pyFunc1(b, a), pyFunc2(a, b) FROM
MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[pyFunc1($0, $1)], EXPR$1=[pyFunc1($1, $0)],
EXPR$2=[pyFunc2($0, $1)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+PythonCalc(select=[pyFunc1(a, b) AS EXPR$0, pyFunc1(b, a) AS EXPR$1,
pyFunc2(a, b) AS EXPR$2])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testDuplicatedCallsInProjection">
+ <Resource name="sql">
+ <![CDATA[SELECT pyFunc1(a, b), pyFunc1(a, b), pyFunc1(a, b) FROM
MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[pyFunc1($0, $1)], EXPR$1=[pyFunc1($0, $1)],
EXPR$2=[pyFunc1($0, $1)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+Calc(select=[f0 AS EXPR$0, f0 AS EXPR$1, f0 AS EXPR$2])
++- PythonCalc(select=[pyFunc1(a, b) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testDuplicatedCallsInProjectionWithCondition">
+ <Resource name="sql">
+ <![CDATA[SELECT pyFunc1(a, b), pyFunc1(a, b) FROM MyTable WHERE
pyFunc1(a, b) > 0]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[pyFunc1($0, $1)], EXPR$1=[pyFunc1($0, $1)])
++- LogicalFilter(condition=[>(pyFunc1($0, $1), 0)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+Calc(select=[f0 AS EXPR$0, f0 AS EXPR$1], where=[(f0 > 0)])
++- PythonCalc(select=[pyFunc1(a, b) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testDuplicatedCallsWithForwardedField">
+ <Resource name="sql">
+ <![CDATA[SELECT a, pyFunc1(a, b), pyFunc1(a, b) FROM MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], EXPR$1=[pyFunc1($0, $1)], EXPR$2=[pyFunc1($0, $1)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+Calc(select=[f0 AS a, f1 AS EXPR$1, f1 AS EXPR$2])
++- PythonCalc(select=[a AS f0, pyFunc1(a, b) AS f1])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testNestedUdfInProjectionWithSameInCondition">
+ <Resource name="sql">
+ <![CDATA[SELECT pyFunc2(pyFunc1(a, b), c), pyFunc1(a, b) FROM MyTable
WHERE pyFunc1(a, b) > 0]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[pyFunc2(pyFunc1($0, $1), $2)], EXPR$1=[pyFunc1($0, $1)])
++- LogicalFilter(condition=[>(pyFunc1($0, $1), 0)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+Calc(select=[f00 AS EXPR$0, f0 AS EXPR$1])
++- PythonCalc(select=[f0, pyFunc2(f0, c) AS f00])
+ +- Calc(select=[f0, c], where=[(f0 > 0)])
+ +- PythonCalc(select=[a, b, c, pyFunc1(a, b) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database,
MyTable]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testNonDeterministicCallsAreNotDeduplicated">
+ <Resource name="sql">
+ <![CDATA[SELECT pyFuncNonDet(a, b), pyFuncNonDet(a, b) FROM MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[pyFuncNonDet($0, $1)], EXPR$1=[pyFuncNonDet($0, $1)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+PythonCalc(select=[pyFuncNonDet(a, b) AS EXPR$0, pyFuncNonDet(a, b) AS EXPR$1])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testSameUdfInConditionAndProjection">
+ <Resource name="sql">
+ <![CDATA[SELECT pyFunc1(a, b) + 1, pyFunc1(a, b) + 2 FROM MyTable WHERE
pyFunc1(a, b) > 0]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[+(pyFunc1($0, $1), 1)], EXPR$1=[+(pyFunc1($0, $1), 2)])
++- LogicalFilter(condition=[>(pyFunc1($0, $1), 0)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+Calc(select=[(f0 + 1) AS EXPR$0, (f0 + 2) AS EXPR$1], where=[(f0 > 0)])
++- PythonCalc(select=[a, b, pyFunc1(a, b) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testSameUdfWithDifferentArgsInConditionAndProjection">
+ <Resource name="sql">
+ <![CDATA[SELECT pyFunc1(b, c) FROM MyTable WHERE pyFunc1(a, b) > 0]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[pyFunc1($1, $2)])
++- LogicalFilter(condition=[>(pyFunc1($0, $1), 0)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+PythonCalc(select=[pyFunc1(b, c) AS EXPR$0])
++- Calc(select=[b, c], where=[(f0 > 0)])
+ +- PythonCalc(select=[b, c, pyFunc1(a, b) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testSameUdfWithSwappedArgsInConditionAndProjection">
+ <Resource name="sql">
+ <![CDATA[SELECT pyFunc1(b, a) FROM MyTable WHERE pyFunc1(a, b) > 0]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[pyFunc1($1, $0)])
++- LogicalFilter(condition=[>(pyFunc1($0, $1), 0)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+PythonCalc(select=[pyFunc1(b, a) AS EXPR$0])
++- Calc(select=[b, a], where=[(f0 > 0)])
+ +- PythonCalc(select=[b, a, pyFunc1(a, b) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+</Root>
