[FLINK-36707][table] Integrate PTF type inference into Calcite

twalthr · Jan 9, 2025 · d5568aa · d5568aa
1 parent 42e2593
commit d5568aa
Show file tree

Hide file tree

Showing 33 changed files with 1,460 additions and 626 deletions.
diff --git a/...-common/src/main/java/org/apache/flink/table/types/extraction/FunctionResultTemplate.java b/...-common/src/main/java/org/apache/flink/table/types/extraction/FunctionResultTemplate.java
@@ -21,7 +21,6 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.inference.StateTypeStrategy;
-import org.apache.flink.table.types.inference.StateTypeStrategyWrapper;
 import org.apache.flink.table.types.inference.TypeStrategies;
 import org.apache.flink.table.types.inference.TypeStrategy;
 
@@ -122,7 +121,7 @@ private void checkSingleStateEntry() {
         }
 
         private static StateTypeStrategy createStateTypeStrategy(DataType dataType) {
-            return StateTypeStrategyWrapper.of(TypeStrategies.explicit(dataType));
+            return StateTypeStrategy.of(TypeStrategies.explicit(dataType));
         }
 
         private static TypeStrategy createTypeStrategy(DataType dataType) {

diff --git a/...-common/src/main/java/org/apache/flink/table/types/extraction/TypeInferenceExtractor.java b/...-common/src/main/java/org/apache/flink/table/types/extraction/TypeInferenceExtractor.java
@@ -38,7 +38,6 @@
 import org.apache.flink.table.types.inference.InputTypeStrategies;
 import org.apache.flink.table.types.inference.InputTypeStrategy;
 import org.apache.flink.table.types.inference.StateTypeStrategy;
-import org.apache.flink.table.types.inference.StateTypeStrategyWrapper;
 import org.apache.flink.table.types.inference.StaticArgument;
 import org.apache.flink.table.types.inference.TypeInference;
 import org.apache.flink.table.types.inference.TypeStrategies;
@@ -321,7 +320,7 @@ private static LinkedHashMap<String, StateTypeStrategy> translateStateTypeStrate
                                         e -> e.getKey().toInputTypeStrategy(),
                                         e -> e.getValue().toAccumulatorTypeStrategy()));
         final StateTypeStrategy accumulatorStrategy =
-                StateTypeStrategyWrapper.of(TypeStrategies.mapping(mappings));
+                StateTypeStrategy.of(TypeStrategies.mapping(mappings));
         final Set<String> stateNames =
                 stateMapping.values().stream()
                         .map(FunctionStateTemplate::toAccumulatorStateName)

diff --git a/.../flink-table-common/src/main/java/org/apache/flink/table/types/inference/CallContext.java b/.../flink-table-common/src/main/java/org/apache/flink/table/types/inference/CallContext.java
@@ -23,6 +23,8 @@
 import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.catalog.DataTypeFactory;
 import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.table.functions.TableSemantics;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.LogicalType;
 
@@ -44,7 +46,8 @@ public interface CallContext {
 
     /**
      * Returns {@code true} if the argument at the given position is a literal and {@code null},
-     * {@code false} otherwise.
+     * {@code false} otherwise. If the argument is declared as optional and has no value, true is
+     * returned.
      *
      * <p>Use {@link #isArgumentLiteral(int)} before to check if the argument is actually a literal.
      */
@@ -61,6 +64,17 @@ public interface CallContext {
      */
     <T> Optional<T> getArgumentValue(int pos, Class<T> clazz);
 
+    /**
+     * Returns information about the table that has been passed to a table argument.
+     *
+     * <p>Semantics are only available for table arguments (i.e. arguments of a {@link
+     * ProcessTableFunction} that are annotated with {@code @ArgumentHint(TABLE_AS_SET)} or
+     * {@code @ArgumentHint(TABLE_AS_ROW)}).
+     */
+    default Optional<TableSemantics> getTableSemantics(int pos) {
+        return Optional.empty();
+    }
+
     /**
      * Returns the function's name usually referencing the function in a catalog.
      *

diff --git a/...s/inference/StateTypeStrategyWrapper.java → ...s/inference/DefaultStateTypeStrategy.java b/...s/inference/StateTypeStrategyWrapper.java → ...s/inference/DefaultStateTypeStrategy.java
@@ -18,28 +18,24 @@
 
 package org.apache.flink.table.types.inference;
 
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.util.Preconditions;
 
 import java.util.Objects;
 import java.util.Optional;
 
 /** A helper class that wraps a {@link TypeStrategy} into a {@link StateTypeStrategy}. */
-@PublicEvolving
-public class StateTypeStrategyWrapper implements StateTypeStrategy {
+@Internal
+class DefaultStateTypeStrategy implements StateTypeStrategy {
 
     private final TypeStrategy typeStrategy;
 
-    private StateTypeStrategyWrapper(TypeStrategy typeStrategy) {
+    DefaultStateTypeStrategy(TypeStrategy typeStrategy) {
         this.typeStrategy =
                 Preconditions.checkNotNull(typeStrategy, "Type strategy must not be null.");
     }
 
-    public static StateTypeStrategyWrapper of(TypeStrategy typeStrategy) {
-        return new StateTypeStrategyWrapper(typeStrategy);
-    }
-
     @Override
     public Optional<DataType> inferType(CallContext callContext) {
         return typeStrategy.inferType(callContext);
@@ -50,8 +46,8 @@ public boolean equals(Object o) {
         if (this == o) {
             return true;
         }
-        if (o instanceof StateTypeStrategyWrapper) {
-            return Objects.equals(typeStrategy, ((StateTypeStrategyWrapper) o).typeStrategy);
+        if (o instanceof DefaultStateTypeStrategy) {
+            return Objects.equals(typeStrategy, ((DefaultStateTypeStrategy) o).typeStrategy);
         }
         if (o instanceof TypeStrategy) {
             return Objects.equals(typeStrategy, o);

diff --git a/...-table-common/src/main/java/org/apache/flink/table/types/inference/StateTypeStrategy.java b/...-table-common/src/main/java/org/apache/flink/table/types/inference/StateTypeStrategy.java
@@ -23,5 +23,10 @@
 /** Strategy for inferring a function call's intermediate result data type (i.e. state entry). */
 @PublicEvolving
 public interface StateTypeStrategy extends TypeStrategy {
+
+    static StateTypeStrategy of(TypeStrategy typeStrategy) {
+        return new DefaultStateTypeStrategy(typeStrategy);
+    }
+
     // marker interface which will be filled with additional contracts in the future
 }
diff --git a/...ink-table-common/src/main/java/org/apache/flink/table/types/inference/StaticArgument.java b/...ink-table-common/src/main/java/org/apache/flink/table/types/inference/StaticArgument.java
@@ -173,6 +173,10 @@ public EnumSet<StaticArgumentTrait> getTraits() {
         return traits;
     }
 
+    public boolean is(StaticArgumentTrait trait) {
+        return traits.contains(trait);
+    }
+
     @Override
     public String toString() {
         final StringBuilder s = new StringBuilder();
@@ -181,13 +185,18 @@ public String toString() {
         // (myTypedTable ROW<i INT> {TABLE BY ROW})
         // (myUntypedTable {TABLE BY ROW})
         s.append(name);
+        s.append(" =>");
         if (dataType != null) {
             s.append(" ");
             s.append(dataType);
         }
         if (!traits.equals(EnumSet.of(StaticArgumentTrait.SCALAR))) {
             s.append(" ");
-            s.append(traits.stream().map(Enum::name).collect(Collectors.joining(", ", "{", "}")));
+            s.append(
+                    traits.stream()
+                            .map(Enum::name)
+                            .map(n -> n.replace('_', ' '))
+                            .collect(Collectors.joining(", ", "{", "}")));
         }
         return s.toString();
     }
@@ -218,7 +227,7 @@ private void checkName() {
             throw new ValidationException(
                     String.format(
                             "Invalid argument name '%s'. An argument must follow "
-                                    + "the pattern [a-zA-Z_$][a-zA-Z_$0-9].",
+                                    + "the pattern [a-zA-Z_$][a-zA-Z_$0-9]*.",
                             name));
         }
     }