[FLINK-39261][table] Add Row Semantic for retract and Set Semantic for upsert mode

Author: raminqaf

docs/content/docs/sql/reference/queries/changelog.md

@@ -43,17 +43,19 @@ This is useful when consuming Change Data Capture (CDC) streams from systems lik
 
 ```sql
 SELECT * FROM FROM_CHANGELOG(
-  input => TABLE source_table PARTITION BY key_col,
+  input => TABLE source_table [PARTITION BY key_col],
   [op => DESCRIPTOR(op_column_name),]
   [op_mapping => MAP['c, r', 'INSERT', 'u', 'UPDATE_AFTER', 'd', 'DELETE']]
 )
 ```
 
+`PARTITION BY` is optional when the mapping includes `UPDATE_BEFORE` (retract mode). It is required when the mapping produces upsert mode (no `UPDATE_BEFORE`), because downstream operators need a key for updates and deletes. When provided, records are distributed by the partition key for parallel processing.
+
 ### Parameters
 
 | Parameter    | Required | Description |
 |:-------------|:---------|:------------|
-| `input`      | Yes      | The input table. Must be append-only and include `PARTITION BY` for parallel execution. |
+| `input`      | Yes      | The input table. Must be append-only. `PARTITION BY` is optional for retract mode (with `UPDATE_BEFORE`) and required for upsert mode (without `UPDATE_BEFORE`). |
 | `op`         | No       | A `DESCRIPTOR` with a single column name for the operation code column. Defaults to `op`. The column must exist in the input table and be of type STRING. |
 | `op_mapping` | No       | A `MAP<STRING, STRING>` mapping user-defined operation codes to change operation names. Keys are user codes (e.g., `'c'`, `'u'`, `'d'`), values are change operation names (`INSERT`, `UPDATE_BEFORE`, `UPDATE_AFTER`, `DELETE`). Keys can contain comma-separated codes to map multiple codes to the same operation (e.g., `'c, r'`). When provided, only mapped codes are forwarded - unmapped codes are dropped. Each change operation may appear at most once across all entries. |
 
@@ -73,7 +75,7 @@ When `op_mapping` is omitted, the following standard names are used:
 The output columns are ordered as:
 
 ```
-[partition_key_columns, remaining_columns_without_op]
+[all_input_columns_without_op]
 ```
 
 The op column is removed from the output. Output rows carry the appropriate change operation (INSERT, UPDATE_BEFORE, UPDATE_AFTER, or DELETE).
@@ -90,7 +92,7 @@ The op column is removed from the output. Output rows carry the appropriate chan
 -- +I[id:2, op:'DELETE',        name:'Bob']
 
 SELECT * FROM FROM_CHANGELOG(
-  input => TABLE cdc_stream PARTITION BY id
+  input => TABLE cdc_stream
 )
 
 -- Output (updating table):
@@ -122,7 +124,7 @@ SELECT * FROM FROM_CHANGELOG(
 
 ```sql
 SELECT * FROM FROM_CHANGELOG(
-  input => TABLE cdc_stream PARTITION BY id,
+  input => TABLE cdc_stream,
   op => DESCRIPTOR(operation)
 )
 -- The operation column named 'operation' is used instead of 'op'
@@ -131,12 +133,11 @@ SELECT * FROM FROM_CHANGELOG(
 #### Table API
 
 ```java
-// Default: reads 'op' column with standard change operation names
-Table result = cdcStream.partitionBy($("id")).fromChangelog();
+// Default (retract mode): reads 'op' column with standard change operation names
+Table result = cdcStream.fromChangelog();
 
-// Debezium-style mapping
-Table result = cdcStream.partitionBy($("id")).fromChangelog(
-    descriptor("__op").asArgument("op"),
+// Upsert mode requires PARTITION BY — use the generic process() method
+Table result = cdcStream.partitionBy($("id")).process("FROM_CHANGELOG",
     map("c, r", "INSERT", "u", "UPDATE_AFTER", "d", "DELETE").asArgument("op_mapping")
 );
 ```

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java

@@ -1323,6 +1323,33 @@ default TableResult executeInsert(
      */
     PartitionedTable partitionBy(Expression... fields);
 
+    /**
+     * Converts this append-only table with an explicit operation code column into a dynamic table
+     * using the built-in {@code FROM_CHANGELOG} process table function.
+     *
+     * <p>Each input row is expected to have a string operation code column (default: {@code "op"})
+     * that indicates the change operation (e.g., INSERT, UPDATE_AFTER, UPDATE_BEFORE, DELETE). The
+     * output table is a dynamic table backed by a changelog stream.
+     *
+     * <p>Optional arguments can be passed using named expressions:
+     *
+     * <pre>{@code
+     * // Default: reads 'op' column with standard change operation names
+     * table.fromChangelog();
+     *
+     * // Custom op column name and mapping (Debezium-style codes)
+     * table.fromChangelog(
+     *     descriptor("__op").asArgument("op"),
+     *     map("c, r", "INSERT", "u", "UPDATE_AFTER", "d", "DELETE").asArgument("op_mapping")
+     * );
+     * }</pre>
+     *
+     * @param arguments optional named arguments for {@code op} and {@code op_mapping}
+     * @return a dynamic {@link Table} with the op column removed and proper change operation
+     *     semantics
+     */
+    Table fromChangelog(Expression... arguments);
+
     /**
      * Converts this table object into a named argument.
      *

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableImpl.java

@@ -497,6 +497,11 @@ public PartitionedTable partitionBy(Expression... fields) {
         return new PartitionedTableImpl(this, Arrays.asList(fields));
     }
 
+    @Override
+    public Table fromChangelog(Expression... arguments) {
+        return process(BuiltInFunctionDefinitions.FROM_CHANGELOG.getName(), (Object[]) arguments);
+    }
+
     @Override
     public ApiExpression asArgument(String name) {
         return createArgumentExpression(operationTree, tableEnvironment, name);

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

@@ -137,15 +137,6 @@ public boolean isInternal() {
         return isInternal;
     }
 
-    /**
-     * Returns the optional changelog mode resolver for declaring the output changelog mode of a
-     * built-in process table function (e.g., FROM_CHANGELOG). Receives the {@link ChangelogContext}
-     * so it can inspect function arguments to dynamically determine the changelog mode.
-     */
-    public Optional<Function<ChangelogContext, ChangelogMode>> getChangelogModeResolver() {
-        return Optional.ofNullable(changelogModeResolver);
-    }
-
     public String getQualifiedName() {
         if (isInternal) {
             return name;

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

@@ -823,7 +823,8 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
                                     false,
                                     EnumSet.of(
                                             StaticArgumentTrait.TABLE,
-                                            StaticArgumentTrait.SET_SEMANTIC_TABLE)),
+                                            StaticArgumentTrait.SET_SEMANTIC_TABLE,
+                                            StaticArgumentTrait.OPTIONAL_PARTITION_BY)),
                             StaticArgument.scalar("op", DataTypes.DESCRIPTOR(), true),
                             StaticArgument.scalar(
                                     "op_mapping",

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

@@ -63,8 +63,10 @@
  * </ol>
  *
  * <p>Emitting changelogs is only valid for PTFs that take table arguments with set semantics (see
- * {@link ArgumentTrait#SET_SEMANTIC_TABLE}). In case of upserts, the upsert key must be equal to
- * the PARTITION BY key.
+ * {@link ArgumentTrait#SET_SEMANTIC_TABLE}). When using {@code OPTIONAL_PARTITION_BY}, the
+ * PARTITION BY clause can be omitted for retract mode (with {@link RowKind#UPDATE_BEFORE}), since
+ * the stream is self-describing. In case of upserts, the upsert key must be equal to the PARTITION
+ * BY key.
  *
  * <p>It is perfectly valid for a {@link ChangelogFunction} implementation to return a fixed {@link
  * ChangelogMode}, regardless of the {@link ChangelogContext}. This approach may be appropriate when

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

@@ -36,7 +36,6 @@
 import org.apache.flink.table.types.inference.TypeStrategy;
 import org.apache.flink.table.types.logical.LogicalTypeFamily;
 import org.apache.flink.types.ColumnList;
-import org.apache.flink.types.RowKind;
 
 import java.util.HashSet;
 import java.util.List;
@@ -166,6 +165,16 @@ private static Optional<List<DataType>> validateInputs(
             }
         }
 
+        // Upsert mode (no UPDATE_BEFORE in mapping) requires PARTITION BY for key-based routing
+        final boolean isUpsertMode =
+                opMapping.isPresent() && !containsUpdateBefore(opMapping.get());
+        final boolean hasPartitionBy = tableSemantics.partitionByColumns().length > 0;
+        if (isUpsertMode && !hasPartitionBy) {
+            throw new ValidationException(
+                    "Upsert changelog mode (without UPDATE_BEFORE) requires a PARTITION BY clause. "
+                            + "Either add PARTITION BY or include UPDATE_BEFORE in the op_mapping.");
+        }
+
         return Optional.of(callContext.getArgumentDataTypes());
     }
 
@@ -223,6 +232,13 @@ private static String resolveOpColumnName(final CallContext callContext) {
                 .orElse(DEFAULT_OP_COLUMN_NAME);
     }
 
+    /** Returns true if the op_mapping values contain UPDATE_BEFORE. */
+    @SuppressWarnings("rawtypes")
+    private static boolean containsUpdateBefore(final Map opMapping) {
+        return opMapping.values().stream()
+                .anyMatch(v -> v instanceof String && "UPDATE_BEFORE".equals(((String) v).trim()));
+    }
+
     private static List<Field> buildOutputFields(
             final TableSemantics tableSemantics, final String opColumnName) {
         final Set<Integer> partitionKeys =
@@ -246,18 +262,16 @@ private static List<Field> buildOutputFields(
      * (default) or includes UPDATE_BEFORE, returns retract mode (all). Otherwise, returns upsert
      * mode (no UPDATE_BEFORE).
      */
-    @SuppressWarnings({"unchecked", "rawtypes"})
+    @SuppressWarnings("rawtypes")
     public static ChangelogMode resolveChangelogMode(final ChangelogContext changelogContext) {
         final Optional<Map> opMapping = changelogContext.getArgumentValue(2, Map.class);
         if (opMapping.isEmpty()) {
             // Default mapping includes UPDATE_BEFORE -> retract mode
             return ChangelogMode.all();
         }
-        final boolean hasUpdateBefore =
-                ((Map<String, String>) opMapping.get())
-                        .values().stream()
-                                .anyMatch(v -> RowKind.UPDATE_BEFORE.name().equals(v.trim()));
-        return hasUpdateBefore ? ChangelogMode.all() : ChangelogMode.upsert(false);
+        return containsUpdateBefore(opMapping.get())
+                ? ChangelogMode.all()
+                : ChangelogMode.upsert(false);
     }
 
     private FromChangelogTypeStrategy() {}

flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/FromChangelogInputTypeStrategyTest.java

@@ -46,10 +46,15 @@ class FromChangelogInputTypeStrategyTest extends InputTypeStrategiesTestBase {
     @Override
     protected Stream<TestSpec> testData() {
         return Stream.of(
-                // Valid: all three arguments with Debezium-style mapping
-                TestSpec.forStrategy("Valid with all arguments", FROM_CHANGELOG_INPUT_TYPE_STRATEGY)
+                // Valid: upsert mapping with PARTITION BY
+                TestSpec.forStrategy(
+                                "Valid upsert with partition by",
+                                FROM_CHANGELOG_INPUT_TYPE_STRATEGY)
                         .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE, MAP_TYPE)
-                        .calledWithTableSemanticsAt(0, new TableSemanticsMock(TABLE_TYPE))
+                        .calledWithTableSemanticsAt(
+                                0,
+                                new TableSemanticsMock(
+                                        TABLE_TYPE, new int[] {0}, new int[0], -1, null))
                         .calledWithLiteralAt(1, ColumnList.of(List.of("op")))
                         .calledWithLiteralAt(
                                 2,
@@ -126,6 +131,21 @@ protected Stream<TestSpec> testData() {
                         .calledWithTableSemanticsAt(0, new TableSemanticsMock(TABLE_TYPE))
                         .calledWithLiteralAt(1, ColumnList.of(List.of("op")))
                         .calledWithLiteralAt(2, Map.of("c", "INSERT", "r", "INSERT"))
-                        .expectErrorMessage("Duplicate change operation: 'INSERT'"));
+                        .expectErrorMessage("Duplicate change operation: 'INSERT'"),
+
+                // Error: upsert mode without PARTITION BY
+                TestSpec.forStrategy(
+                                "Upsert without PARTITION BY", FROM_CHANGELOG_INPUT_TYPE_STRATEGY)
+                        .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE, MAP_TYPE)
+                        .calledWithTableSemanticsAt(0, new TableSemanticsMock(TABLE_TYPE))
+                        .calledWithLiteralAt(1, ColumnList.of(List.of("op")))
+                        .calledWithLiteralAt(
+                                2,
+                                Map.of(
+                                        "c", "INSERT",
+                                        "u", "UPDATE_AFTER",
+                                        "d", "DELETE"))
+                        .expectErrorMessage(
+                                "Upsert changelog mode (without UPDATE_BEFORE) requires a PARTITION BY"));
     }
 }

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

@@ -1713,7 +1713,10 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
         val changelogContext =
           toPtfChangelogContext(process, inputChangelogModes, requiredChangelogMode)
         val changelogMode = changelogFunction.getChangelogMode(changelogContext)
-        if (!changelogMode.containsOnly(RowKind.INSERT)) {
+        if (
+          !changelogMode.containsOnly(RowKind.INSERT) &&
+          !changelogMode.contains(RowKind.UPDATE_BEFORE)
+        ) {
           verifyPtfTableArgsForUpdates(call)
         }
         toTraitSet(changelogMode)
@@ -1722,6 +1725,15 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
     }
   }
 
+  /**
+   * Verifies that PTFs with upsert output (without UPDATE_BEFORE) use set semantics.
+   *
+   * Retract mode (with UPDATE_BEFORE) is self-describing — each update carries both the old and new
+   * value, so downstream can process it without a key. Row semantics is safe.
+   *
+   * Upsert mode (without UPDATE_BEFORE) requires a key to look up previous values, so set semantics
+   * with PARTITION BY is required.
+   */
   private def verifyPtfTableArgsForUpdates(call: RexCall): Unit = {
     StreamPhysicalProcessTableFunction
       .getProvidedInputArgs(call)
@@ -1730,7 +1742,7 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
         tableArg =>
           if (tableArg.is(StaticArgumentTrait.ROW_SEMANTIC_TABLE)) {
             throw new ValidationException(
-              s"PTFs that take table arguments with row semantics don't support updating output. " +
+              s"PTFs that take table arguments with row semantics don't support upsert output. " +
                 s"Table argument '${tableArg.getName}' of function '${call.getOperator.toString}' " +
                 s"must use set semantics.")
           }

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/FromChangelogSemanticTests.java

@@ -43,8 +43,9 @@ public List<TableTestProgram> programs() {
                 FromChangelogTestPrograms.DEBEZIUM_MAPPING,
                 FromChangelogTestPrograms.UNMAPPED_CODES_DROPPED,
                 FromChangelogTestPrograms.CUSTOM_OP_NAME,
-                FromChangelogTestPrograms.TABLE_API_DEFAULT,
-                FromChangelogTestPrograms.MISSING_PARTITION_BY
+                FromChangelogTestPrograms.TABLE_API_DEFAULT
+                // TODO: UPSERT_WITHOUT_PARTITION_BY validation is tested in
+                // FromChangelogInputTypeStrategyTest
                 // TODO: enable after TO_CHANGELOG switches to row semantics (PR #27911)
                 // FromChangelogTestPrograms.ROUND_TRIP
                 );