[FLINK-39268][table] Expand and reuse local refs in CalcCodeGenerator

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/FunctionDefinitionQueryable.java

@@ -0,0 +1,30 @@
+/*
+ * 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.functions.sql;
+
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
+
+/**
+ * Function which is not bridged by {@link BridgingSqlFunction} and which is not legacy {@link
+ * TableFunction} however whose {@link FunctionDefinition} might be queried.
+ */
+public interface FunctionDefinitionQueryable {
+    FunctionDefinition getFunctionDefinition();
+}

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/SqlJsonArrayFunctionWrapper.java

@@ -18,6 +18,9 @@
 
 package org.apache.flink.table.planner.functions.sql;
 
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionDefinition;
+
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.sql.SqlOperatorBinding;
 import org.apache.calcite.sql.fun.SqlJsonArrayFunction;
@@ -29,7 +32,8 @@
  * This class is a wrapper class for the {@link SqlJsonArrayFunction} but using the {@code
  * VARCHAR_NOT_NULL} return type inference.
  */
-class SqlJsonArrayFunctionWrapper extends SqlJsonArrayFunction {
+class SqlJsonArrayFunctionWrapper extends SqlJsonArrayFunction
+        implements FunctionDefinitionQueryable {
 
     @Override
     public RelDataType inferReturnType(SqlOperatorBinding opBinding) {
@@ -49,4 +53,9 @@ public RelDataType inferReturnType(SqlOperatorBinding opBinding) {
     public SqlReturnTypeInference getReturnTypeInference() {
         return VARCHAR_NOT_NULL;
     }
+
+    @Override
+    public FunctionDefinition getFunctionDefinition() {
+        return BuiltInFunctionDefinitions.JSON_ARRAY;
+    }
 }

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/SqlJsonObjectFunctionWrapper.java

@@ -18,6 +18,9 @@
 
 package org.apache.flink.table.planner.functions.sql;
 
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionDefinition;
+
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.sql.SqlOperatorBinding;
 import org.apache.calcite.sql.fun.SqlJsonObjectFunction;
@@ -29,7 +32,8 @@
  * This class is a wrapper class for the {@link SqlJsonObjectFunction} but using the {@code
  * VARCHAR_NOT_NULL} return type inference.
  */
-class SqlJsonObjectFunctionWrapper extends SqlJsonObjectFunction {
+class SqlJsonObjectFunctionWrapper extends SqlJsonObjectFunction
+        implements FunctionDefinitionQueryable {
 
     @Override
     public RelDataType inferReturnType(SqlOperatorBinding opBinding) {
@@ -49,4 +53,9 @@ public RelDataType inferReturnType(SqlOperatorBinding opBinding) {
     public SqlReturnTypeInference getReturnTypeInference() {
         return VARCHAR_NOT_NULL;
     }
+
+    @Override
+    public FunctionDefinition getFunctionDefinition() {
+        return BuiltInFunctionDefinitions.JSON_OBJECT;
+    }
 }

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/SqlJsonQueryFunctionWrapper.java

@@ -19,6 +19,8 @@
 package org.apache.flink.table.planner.functions.sql;
 
 import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionDefinition;
 
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.sql.SqlCallBinding;
@@ -42,7 +44,8 @@
  * This class is a wrapper class for the {@link SqlJsonQueryFunction} but using the {@code
  * VARCHAR_FORCE_NULLABLE} return type inference.
  */
-class SqlJsonQueryFunctionWrapper extends SqlJsonQueryFunction {
+class SqlJsonQueryFunctionWrapper extends SqlJsonQueryFunction
+        implements FunctionDefinitionQueryable {
     private final SqlReturnTypeInference returnTypeInference;
 
     SqlJsonQueryFunctionWrapper() {
@@ -142,4 +145,9 @@ private static RelDataType explicitTypeSpec(SqlOperatorBinding opBinding) {
         }
         return null;
     }
+
+    @Override
+    public FunctionDefinition getFunctionDefinition() {
+        return BuiltInFunctionDefinitions.JSON_QUERY;
+    }
 }

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/SqlJsonValueFunctionWrapper.java

@@ -18,6 +18,9 @@
 
 package org.apache.flink.table.planner.functions.sql;
 
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionDefinition;
+
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.sql.SqlJsonValueReturning;
 import org.apache.calcite.sql.SqlOperatorBinding;
@@ -35,7 +38,8 @@
  * VARCHAR_FORCE_NULLABLE} return type inference by default. It also supports specifying return type
  * with the RETURNING keyword just like the original {@link SqlJsonValueFunction}.
  */
-class SqlJsonValueFunctionWrapper extends SqlJsonValueFunction {
+class SqlJsonValueFunctionWrapper extends SqlJsonValueFunction
+        implements FunctionDefinitionQueryable {
 
     private final SqlReturnTypeInference returnTypeInference;
 
@@ -80,4 +84,9 @@ private static RelDataType explicitTypeSpec(SqlOperatorBinding opBinding) {
         }
         return null;
     }
+
+    @Override
+    public FunctionDefinition getFunctionDefinition() {
+        return BuiltInFunctionDefinitions.JSON_VALUE;
+    }
 }

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecCalc.java

@@ -32,6 +32,7 @@
 import org.apache.flink.table.planner.plan.nodes.exec.SingleTransformationTranslator;
 import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
 import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
 import org.apache.flink.table.runtime.operators.CodeGenOperatorFactory;
 import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
 import org.apache.flink.table.types.logical.RowType;
@@ -99,10 +100,11 @@ protected Transformation<RowData> translateToPlanInternal(
         final CodeGenOperatorFactory<RowData> substituteStreamOperator =
                 CalcCodeGenerator.generateCalcOperator(
                         ctx,
-                        inputTransform,
+                        (RowType) inputEdge.getOutputType(),
                         (RowType) getOutputType(),
                         JavaScalaConversionUtil.toScala(projection),
                         JavaScalaConversionUtil.toScala(Optional.ofNullable(this.condition)),
+                        ShortcutUtils.unwrapTypeFactory(planner),
                         retainHeader,
                         getClass().getSimpleName());
         return ExecNodeUtil.createOneInputTransformation(

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java

@@ -502,7 +502,8 @@ protected StreamOperatorFactory<RowData> createAsyncLookupJoin(
                             JavaScalaConversionUtil.toScala(projectionOnTemporalTable),
                             filterOnTemporalTable,
                             projectionOutputRelDataType,
-                            tableSourceRowType);
+                            tableSourceRowType,
+                            ShortcutUtils.unwrapTypeFactory(relBuilder));
             asyncFunc =
                     new AsyncLookupJoinWithCalcRunner(
                             generatedFuncWithType.tableFunc(),
@@ -647,7 +648,8 @@ protected ProcessFunction<RowData, RowData> createSyncLookupJoinFunction(
                             JavaScalaConversionUtil.toScala(projectionOnTemporalTable),
                             filterOnTemporalTable,
                             projectionOutputRelDataType,
-                            tableSourceRowType);
+                            tableSourceRowType,
+                            ShortcutUtils.unwrapTypeFactory(relBuilder));
 
             processFunc =
                     new LookupJoinWithCalcRunner(

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/DeltaJoinTree.java

@@ -228,7 +228,8 @@ private DeltaJoinRuntimeTree.Node convert2RuntimeTreeInternal(
                                                 node.filter,
                                                 rowTypePassThroughCalc,
                                                 rowTypeBeforeCalc,
-                                                generatedCalcName))
+                                                generatedCalcName,
+                                                typeFactory))
                         .orElse(null);
 
         if (node instanceof BinaryInputNode) {

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeltaJoin.java

@@ -574,7 +574,8 @@ private static LookupHandlerBase generateLookupHandler(
                             JavaScalaConversionUtil.toScala(projectionOnTemporalTable),
                             filterOnTemporalTable,
                             lookupSidePassThroughCalcRowType,
-                            lookupTableSourceRowType);
+                            lookupTableSourceRowType,
+                            typeFactory);
         }
 
         Preconditions.checkState(!generatedFetcherCollector.containsKey(lookupTableOrdinal));

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/ShortcutUtils.java

@@ -31,6 +31,7 @@
 import org.apache.flink.table.planner.delegation.PlannerBase;
 import org.apache.flink.table.planner.expressions.RexNodeExpression;
 import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
+import org.apache.flink.table.planner.functions.sql.FunctionDefinitionQueryable;
 import org.apache.flink.table.planner.functions.utils.TableSqlFunction;
 
 import org.apache.calcite.plan.Context;
@@ -40,13 +41,20 @@
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexFieldAccess;
+import org.apache.calcite.rex.RexLocalRef;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlOperatorBinding;
 import org.apache.calcite.tools.RelBuilder;
 
 import javax.annotation.Nullable;
 
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
 /**
  * Utilities for quick access of commonly used instances (like {@link FlinkTypeFactory}) without
  * long chains of getters or casting like {@code (FlinkTypeFactory)
@@ -147,14 +155,18 @@ public static DataTypeFactory unwrapDataTypeFactory(RelBuilder relBuilder) {
             return null;
         }
         final RexCall call = (RexCall) rexNode;
-        if (!(call.getOperator() instanceof BridgingSqlFunction)) {
+        final SqlOperator operator = call.getOperator();
+        if (!(operator instanceof BridgingSqlFunction)) {
+            if (operator instanceof FunctionDefinitionQueryable) {
+                return ((FunctionDefinitionQueryable) operator).getFunctionDefinition();
+            }
             // legacy
-            if (call.getOperator() instanceof TableSqlFunction) {
-                return ((TableSqlFunction) call.getOperator()).udtf();
+            if (operator instanceof TableSqlFunction) {
+                return ((TableSqlFunction) operator).udtf();
             }
             return null;
         }
-        return ((BridgingSqlFunction) call.getOperator()).getDefinition();
+        return ((BridgingSqlFunction) operator).getDefinition();
     }
 
     public static @Nullable FunctionDefinition unwrapFunctionDefinition(SqlOperator operator) {
@@ -169,6 +181,56 @@ public static boolean isFunctionKind(SqlOperator operator, FunctionKind kind) {
         return functionDefinition != null && functionDefinition.getKind() == kind;
     }
 
+    public static boolean isOneOfFunctionDefinitions(
+            RexNode rexNode, FunctionDefinition... expectedDefinitions) {
+        if (!(rexNode instanceof RexCall)) {
+            return false;
+        }
+        final RexCall call = (RexCall) rexNode;
+        final FunctionDefinition unwrapped = unwrapFunctionDefinition(call);
+        for (FunctionDefinition expected : expectedDefinitions) {
+            if (unwrapped == expected) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    public static boolean isDeterministicThroughProgram(
+            RexNode node, @Nullable List<RexNode> exprs) {
+        if (exprs == null) {
+            return RexUtil.isDeterministic(node);
+        }
+        return isDeterministicThroughProgram(node, exprs, new HashSet<>());
+    }
+
+    private static boolean isDeterministicThroughProgram(
+            RexNode node, List<RexNode> exprs, Set<Integer> visited) {
+        if (node instanceof RexCall) {
+            final RexCall call = (RexCall) node;
+            if (!call.getOperator().isDeterministic()) {
+                return false;
+            }
+            for (RexNode operand : call.getOperands()) {
+                if (!isDeterministicThroughProgram(operand, exprs, visited)) {
+                    return false;
+                }
+            }
+            return true;
+        }
+        if (node instanceof RexLocalRef) {
+            final int idx = ((RexLocalRef) node).getIndex();
+            // already on the stack: skip rather than recurse forever
+            return !visited.add(idx)
+                    || isDeterministicThroughProgram(exprs.get(idx), exprs, visited);
+        }
+        if (node instanceof RexFieldAccess) {
+            return isDeterministicThroughProgram(
+                    ((RexFieldAccess) node).getReferenceExpr(), exprs, visited);
+        }
+        return true;
+    }
+
     public static @Nullable BridgingSqlFunction unwrapBridgingSqlFunction(RexCall call) {
         final SqlOperator operator = call.getOperator();
         if (operator instanceof BridgingSqlFunction) {