dianfu commented on code in PR #28998:
URL: https://github.com/apache/flink/pull/28998#discussion_r3900490898
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/PythonCallCseResult.java:
##########
@@ -0,0 +1,82 @@
+/*
Review Comment:
Could you move PythonCallDeduplicator and PythonCallCseResult to the utils
package?
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/python/PythonFunctionInfo.java:
##########
@@ -36,11 +36,65 @@ public class PythonFunctionInfo implements Serializable {
private final PythonFunction pythonFunction;
/**
- * The input arguments, it could be an input offset of the input row or
the execution result of
- * another python function described as PythonFunctionInfo.
+ * The input arguments. It could be one of the following:
+ *
+ * <ul>
+ * <li>{@link Integer} – an input offset of the input row
Review Comment:
What about introduce interface PythonFunctionInput, subclasses InputRef,
ResultRef, ConstantInput and also making PythonFunctionInfo extends
PythonFunctionInput to make the inputs more clear?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecPythonCalc.java:
##########
@@ -192,14 +211,14 @@ private OneInputTransformation<RowData, RowData>
createPythonOneInputTransformat
}
private Tuple2<int[], PythonFunctionInfo[]>
extractPythonScalarFunctionInfos(
- List<RexCall> rexCalls, ClassLoader classLoader) {
+ List<RexCall> rexCalls, ClassLoader classLoader, Map<RexCall,
Integer> refMap) {
Review Comment:
What about reorder the parameters: rexCalls, refMap, classLoader?
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/python/PythonFunctionInfo.java:
##########
@@ -36,11 +36,65 @@ public class PythonFunctionInfo implements Serializable {
private final PythonFunction pythonFunction;
/**
- * The input arguments, it could be an input offset of the input row or
the execution result of
- * another python function described as PythonFunctionInfo.
+ * The input arguments. It could be one of the following:
+ *
+ * <ul>
+ * <li>{@link Integer} – an input offset of the input row
+ * <li>{@link PythonFunctionInfo} – the execution result of another
python function (nested
+ * call)
+ * <li>{@code byte[]} – a constant value
+ * <li>{@link ResultRef} – a reference to the result of a previously
computed function in the
+ * flattened UDF list (used for cross-subtree CSE)
+ * </ul>
*/
private Object[] inputs;
+ /**
+ * This function's position in the operator output, or {@code -1} when its
result is only
+ * referenced by another function and must not be emitted.
+ */
+ private int outputPosition = -1;
Review Comment:
I think we should not introduce field outputPosition in this class. It makes
class PythonFunctionInfo difficult to understand. PythonFunctionInfo only
stands how a UDF could be computed, it doesn't stand how it will be used.
##########
flink-python/pyflink/table/tests/test_udf.py:
##########
@@ -926,6 +926,135 @@ def test_create_and_drop_function(self):
self.assertTrue('add_one_func' not in
t_env.list_user_defined_functions())
self.assertTrue('subtract_one_func' not in
t_env.list_user_defined_functions())
+ def test_python_local_ref_reuse(self):
Review Comment:
Could we keep this IT case focused on nested CSE? It runs in stream, batch,
and embedded-thread modes, so the first five cases add 15 jobs while re-testing
the existing top-level/condition CSE.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/PythonCallCseResult.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.nodes.exec.common;
+
+import org.apache.flink.annotation.Internal;
+
+import org.apache.calcite.rex.RexCall;
+
+import java.util.List;
+import java.util.Map;
+
+/** Encapsulates the result of Python UDF call Common Sub-expression
Elimination (CSE). */
+@Internal
+public class PythonCallCseResult {
+
+ /**
+ * The flattened Python UDF calls to be evaluated, in execution order.
+ *
+ * <p>Nested call trees are flattened so that a sub-expression shared
between calls appears
+ * exactly once. For example {@code SELECT udf1(x), udf2(udf1(x))} yields
{@code [udf1(x),
+ * udf2(<ref to udf1(x)>)]}.
+ */
+ private final List<RexCall> uniqueCalls;
Review Comment:
How about renaming uniqueCalls to deduplicatedCalls?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/PythonCallDeduplicator.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.nodes.exec.common;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.table.planner.plan.utils.PythonUtil;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+
+/**
+ * Utility for Python UDF Common Sub-expression Elimination (CSE) of nested
calls.
+ *
+ * <p>Nested Python UDF call trees are flattened in post-order and
deduplicated by structural
+ * equivalence, so that a sub-expression shared between calls is evaluated
only once by the Python
+ * worker. Duplicates between whole projection entries are already removed in
the planner by {@code
+ * RemoteCalcProjectionCseRule}, so this only concerns sub-expressions.
+ */
+@Internal
+public class PythonCallDeduplicator {
+
+ /**
+ * Recursively collects all deterministic Python UDF calls from a call
tree in DFS post-order.
+ *
+ * <p>Post-order ensures child results are computed before parents that
reference them via
+ * refIndex. Non-deterministic children are NOT flattened to prevent
incorrect sharing.
+ */
+ @VisibleForTesting
+ static List<RexCall> collectAllPythonUdfCalls(RexCall root) {
Review Comment:
Could be declared as private? It's marked as VisibleForTesting, however, I
have not found any test case which refers it.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/PythonCallDeduplicator.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.nodes.exec.common;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.table.planner.plan.utils.PythonUtil;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+
+/**
+ * Utility for Python UDF Common Sub-expression Elimination (CSE) of nested
calls.
+ *
+ * <p>Nested Python UDF call trees are flattened in post-order and
deduplicated by structural
+ * equivalence, so that a sub-expression shared between calls is evaluated
only once by the Python
+ * worker. Duplicates between whole projection entries are already removed in
the planner by {@code
+ * RemoteCalcProjectionCseRule}, so this only concerns sub-expressions.
+ */
+@Internal
+public class PythonCallDeduplicator {
+
+ /**
+ * Recursively collects all deterministic Python UDF calls from a call
tree in DFS post-order.
+ *
+ * <p>Post-order ensures child results are computed before parents that
reference them via
+ * refIndex. Non-deterministic children are NOT flattened to prevent
incorrect sharing.
+ */
+ @VisibleForTesting
+ static List<RexCall> collectAllPythonUdfCalls(RexCall root) {
+ List<RexCall> result = new ArrayList<>();
+ for (RexNode operand : root.getOperands()) {
+ if (operand instanceof RexCall &&
PythonUtil.isPythonCall((RexCall) operand)) {
+ RexCall childCall = (RexCall) operand;
+ // Only flatten deterministic child calls for CSE.
+ // Non-deterministic calls must remain nested to avoid
incorrect sharing.
+ if (ShortcutUtils.isDeterministicThroughProgram(childCall,
null)) {
+ result.addAll(collectAllPythonUdfCalls(childCall));
+ }
+ }
+ }
+ result.add(root);
+ return result;
+ }
+
+ /**
+ * Flattens the given Python UDF call trees and deduplicates the resulting
calls.
+ *
+ * <p>Flattening all trees into a single list enables cross-subtree reuse:
e.g. in {@code SELECT
+ * udf1(x), udf2(udf1(x))}, the inner {@code udf1(x)} is evaluated only
once and {@code udf2}
+ * receives its result by reference.
+ */
+ public static PythonCallCseResult deduplicate(List<RexCall>
pythonRexCalls) {
+ // Flatten: collect all Python UDF calls from all projection trees in
post-order, so a
+ // nested sub-expression is always evaluated before the call
referencing it. The root of
+ // each tree is the last element of its own sub-list.
+ List<RexCall> allCalls = new ArrayList<>();
+ int[] rootPositions = new int[pythonRexCalls.size()];
+ for (int i = 0; i < pythonRexCalls.size(); i++) {
+ List<RexCall> subtreeCalls =
collectAllPythonUdfCalls(pythonRexCalls.get(i));
+ rootPositions[i] = allCalls.size() + subtreeCalls.size() - 1;
+ allCalls.addAll(subtreeCalls);
+ }
+
+ // Deduplicate the flattened list by structural equivalence,
preserving post-order.
+ LinkedHashMap<RexCall, Integer> callToIndex = new LinkedHashMap<>();
+ List<RexCall> uniqueCalls = new ArrayList<>();
+ int[] allToUnique = new int[allCalls.size()];
+ for (int i = 0; i < allCalls.size(); i++) {
+ RexCall call = allCalls.get(i);
+ boolean canReuse =
ShortcutUtils.isDeterministicThroughProgram(call, null);
+ Integer existing = canReuse ? callToIndex.get(call) : null;
+ if (existing != null) {
+ allToUnique[i] = existing;
+ continue;
+ }
+ int newPos = uniqueCalls.size();
+ if (canReuse) {
+ callToIndex.put(call, newPos);
+ }
+ uniqueCalls.add(call);
+ allToUnique[i] = newPos;
+ }
+
+ // Flattening adds entries for nested sub-expressions, and post-order
means a top-level
+ // result is not necessarily last, so record where each projection
entry ended up. Only
+ // those positions form the operator output.
+ int[] outputIndices = new int[pythonRexCalls.size()];
+ for (int i = 0; i < pythonRexCalls.size(); i++) {
+ outputIndices[i] = allToUnique[rootPositions[i]];
+ }
+
+ // Build refMap for sub-expression cross-referencing. putIfAbsent
preserves the first
+ // occurrence index, ensuring a parent references its own child rather
than a later
+ // structurally-equal duplicate.
+ LinkedHashMap<RexCall, Integer> refMap = new LinkedHashMap<>();
+ for (int i = 0; i < uniqueCalls.size(); i++) {
Review Comment:
It seems that refMap could be computed according to uniqueCalls. If so,
maybe the constructor of PythonCallCseResult should only accept uniqueCalls
and outputIndices.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
