[FLINK-39261][table] Introduce ChangelogModeStrategy interface for BuiltInFunctionDefinition

Author: raminqaf

flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinition.java

@@ -21,9 +21,8 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.table.api.TableException;
 import org.apache.flink.table.catalog.DataTypeFactory;
-import org.apache.flink.table.connector.ChangelogMode;
-import org.apache.flink.table.functions.ChangelogFunction.ChangelogContext;
 import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ChangelogModeStrategy;
 import org.apache.flink.table.types.inference.InputTypeStrategy;
 import org.apache.flink.table.types.inference.StaticArgument;
 import org.apache.flink.table.types.inference.TypeInference;
@@ -35,7 +34,6 @@
 import java.util.Arrays;
 import java.util.Locale;
 import java.util.Optional;
-import java.util.function.Function;
 import java.util.regex.Pattern;
 
 import static org.apache.flink.util.Preconditions.checkArgument;
@@ -75,7 +73,7 @@ public final class BuiltInFunctionDefinition implements SpecializedFunction {
 
     private final SqlCallSyntax sqlCallSyntax;
 
-    private final @Nullable Function<ChangelogContext, ChangelogMode> changelogModeResolver;
+    private final @Nullable ChangelogModeStrategy changelogModeStrategy;
 
     private final String sqlName;
 
@@ -90,7 +88,7 @@ private BuiltInFunctionDefinition(
             boolean isRuntimeProvided,
             String runtimeClass,
             boolean isInternal,
-            @Nullable Function<ChangelogContext, ChangelogMode> changelogModeResolver) {
+            @Nullable ChangelogModeStrategy changelogModeStrategy) {
         this.name = checkNotNull(name, "Name must not be null.");
         this.sqlName = sqlName;
         this.version = isInternal ? null : version;
@@ -101,7 +99,7 @@ private BuiltInFunctionDefinition(
         this.runtimeClass = runtimeClass;
         this.isInternal = isInternal;
         this.sqlCallSyntax = sqlCallSyntax;
-        this.changelogModeResolver = changelogModeResolver;
+        this.changelogModeStrategy = changelogModeStrategy;
         validateFunction(this.name, this.version, this.isInternal);
     }
 
@@ -139,11 +137,11 @@ public boolean isInternal() {
     }
 
     /**
-     * Returns the optional changelog mode resolver for built-in PTFs that emit updates (e.g.,
+     * Returns the optional {@link ChangelogModeStrategy} for built-in PTFs that emit updates (e.g.,
      * FROM_CHANGELOG). The planner uses this to determine the output changelog mode.
      */
-    public Optional<Function<ChangelogContext, ChangelogMode>> getChangelogModeResolver() {
-        return Optional.ofNullable(changelogModeResolver);
+    public Optional<ChangelogModeStrategy> getChangelogModeStrategy() {
+        return Optional.ofNullable(changelogModeStrategy);
     }
 
     public String getQualifiedName() {
@@ -268,7 +266,7 @@ public static final class Builder {
 
         private SqlCallSyntax sqlCallSyntax = SqlCallSyntax.FUNCTION;
 
-        private @Nullable Function<ChangelogContext, ChangelogMode> changelogModeResolver;
+        private @Nullable ChangelogModeStrategy changelogModeStrategy;
 
         public Builder() {
             // default constructor to allow a fluent definition
@@ -417,12 +415,11 @@ public Builder sqlName(String name) {
         }
 
         /**
-         * Sets a resolver that determines the output {@link ChangelogMode} for this built-in
-         * function. Only needed for functions that emit updates (e.g., FROM_CHANGELOG).
+         * Sets the {@link ChangelogModeStrategy} that determines the output changelog mode for this
+         * built-in PTF. Only needed for PTFs that emit updates (e.g., FROM_CHANGELOG).
          */
-        public Builder changelogModeResolver(
-                Function<ChangelogContext, ChangelogMode> changelogModeResolver) {
-            this.changelogModeResolver = changelogModeResolver;
+        public Builder changelogModeStrategy(ChangelogModeStrategy changelogModeStrategy) {
+            this.changelogModeStrategy = changelogModeStrategy;
             return this;
         }
 
@@ -438,7 +435,7 @@ public BuiltInFunctionDefinition build() {
                     isRuntimeProvided,
                     runtimeClass,
                     isInternal,
-                    changelogModeResolver);
+                    changelogModeStrategy);
         }
     }
 }

flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java

@@ -829,7 +829,7 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
                                     "op_mapping",
                                     DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()),
                                     true))
-                    .changelogModeResolver(ctx -> ChangelogMode.all())
+                    .changelogModeStrategy(ctx -> ChangelogMode.all())
                     .inputTypeStrategy(FROM_CHANGELOG_INPUT_TYPE_STRATEGY)
                     .outputTypeStrategy(FROM_CHANGELOG_OUTPUT_TYPE_STRATEGY)
                     .runtimeClass(

flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/ChangelogModeStrategy.java

@@ -0,0 +1,37 @@
+/*
+ * 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.types.inference;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.functions.ChangelogFunction.ChangelogContext;
+
+/**
+ * Strategy for determining the output {@link ChangelogMode} of a built-in process table function.
+ *
+ * <p>Similar to {@link TypeStrategy}, this is used to declare changelog semantics in the function
+ * definition rather than implementing the {@link
+ * org.apache.flink.table.functions.ChangelogFunction} interface directly.
+ */
+@Experimental
+public interface ChangelogModeStrategy {
+
+    /** Infers the output {@link ChangelogMode} based on the given {@link ChangelogContext}. */
+    ChangelogMode inferChangelogMode(ChangelogContext changelogContext);
+}

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

@@ -1713,11 +1713,12 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
         val changelogMode = changelogFunction.getChangelogMode(changelogContext)
         verifyPtfTableArgsForUpdates(call, changelogMode)
         toTraitSet(changelogMode)
-      case builtIn: BuiltInFunctionDefinition if builtIn.getChangelogModeResolver.isPresent =>
+      case builtIn: BuiltInFunctionDefinition if builtIn.getChangelogModeStrategy.isPresent =>
         val inputChangelogModes = children.map(toChangelogMode(_, None, None))
         val changelogContext =
           toPtfChangelogContext(process, inputChangelogModes, requiredChangelogMode)
-        val changelogMode = builtIn.getChangelogModeResolver.get().apply(changelogContext)
+        val changelogMode =
+          builtIn.getChangelogModeStrategy.get().inferChangelogMode(changelogContext)
         verifyPtfTableArgsForUpdates(call, changelogMode)
         toTraitSet(changelogMode)
       case _ =>