[FLINK-39261][table] Add FROM_CHANGELOG built-in process table function

Author: raminqaf

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

@@ -30,9 +30,119 @@ Flink SQL provides built-in process table functions (PTFs) for working with chan
 
 | Function | Description |
 |:---------|:------------|
+| [FROM_CHANGELOG](#from_changelog) | Converts an append-only table with operation codes into a (potentially updating) dynamic table |
 | [TO_CHANGELOG](#to_changelog) | Converts a dynamic table into an append-only table with explicit operation codes |
 
-<!-- Placeholder for future FROM_CHANGELOG function -->
+## FROM_CHANGELOG
+
+The `FROM_CHANGELOG` PTF converts an append-only table with an explicit operation code column into a (potentially updating) dynamic table. Each input row is expected to have a string column that indicates the change operation. The operation column is interpreted by the engine and removed from the output.
+
+This is useful when consuming Change Data Capture (CDC) streams from systems like Debezium where events arrive as flat append-only records with an explicit operation field. It's also useful to be used in combination with the TO_CHANGELOG function, when converting the append-only table back into an updating table after doing some specific transformation to the events.
+
+Note: This version requires that your CDC data encodes updates using a full image (i.e. providing separate events for before and after the update). Please double-check whether your source provides both UPDATE_BEFORE and UPDATE_AFTER events. FROM_CHANGELOG is a very powerful function but might produce incorrect results in subsequent operations and tables, if not configured correctly.
+
+### Syntax
+
+```sql
+SELECT * FROM FROM_CHANGELOG(
+  input => TABLE source_table,
+  [op => DESCRIPTOR(op_column_name),]
+  [op_mapping => MAP[
+      'c, r', 'INSERT',
+      'ub', 'UPDATE_BEFORE',
+      'ua', 'UPDATE_AFTER',
+      'd', 'DELETE'
+  ]]
+)
+```
+
+### Parameters
+
+| Parameter    | Required | Description                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
+|:-------------|:---------|:------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| `input`      | Yes      | The input table. Must be append-only.                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
+| `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 codes to Flink change operation names. Keys are user-defined codes (e.g., `'c'`, `'u'`, `'d'`), values are Flink 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. |
+
+#### Default op_mapping
+
+When `op_mapping` is omitted, the following standard names are used. They allow a reverse conversion from TO_CHANGELOG by default.
+
+| Input code         | Change operation  |
+|:-------------------|:------------------|
+| `'INSERT'`         | INSERT            |
+| `'UPDATE_BEFORE'`  | UPDATE_BEFORE     |
+| `'UPDATE_AFTER'`   | UPDATE_AFTER      |
+| `'DELETE'`         | DELETE            |
+
+### Output Schema
+
+The output contains all input columns except the operation code (e.g., op) column, which is interpreted by Flink's SQL engine and removed. Each output row carries the appropriate change operation (INSERT, UPDATE_BEFORE, UPDATE_AFTER, or DELETE).
+
+```
+[all_input_columns_without_op]
+```
+
+### Examples
+
+#### Basic usage with standard op names
+
+```sql
+-- Input (append-only):
+-- +I[id:1, op:'INSERT',        name:'Alice']
+-- +I[id:1, op:'UPDATE_BEFORE', name:'Alice']
+-- +I[id:1, op:'UPDATE_AFTER',  name:'Alice2']
+-- +I[id:2, op:'DELETE',        name:'Bob']
+
+SELECT * FROM FROM_CHANGELOG(
+  input => TABLE cdc_stream
+)
+
+-- Output (updating table):
+-- +I[id:1, name:'Alice']
+-- -U[id:1, name:'Alice']
+-- +U[id:1, name:'Alice2']
+-- -D[id:2, name:'Bob']
+
+-- Table state after all events:
+-- | id | name   |
+-- |----|--------|
+-- | 1  | Alice2 |
+```
+
+#### Custom operation column name
+
+```sql
+-- Source schema: id INT, operation STRING, name STRING
+SELECT * FROM FROM_CHANGELOG(
+  input => TABLE cdc_stream,
+  op => DESCRIPTOR(operation)
+)
+-- The operation column named 'operation' is used instead of 'op'
+```
+
+#### Table API
+
+```java
+Table cdcStream = ...;
+
+// Default: reads 'op' column with standard change operation names
+Table result = cdcStream.fromChangelog();
+
+// With custom op column name
+Table result = cdcStream.fromChangelog(
+    descriptor("operation").asArgument("op")
+);
+
+// With custom op_mapping
+Table result = cdcStream.fromChangelog(
+    descriptor("op").asArgument("op"),
+    map("c, r", "INSERT",
+        "ub", "UPDATE_BEFORE",
+        "ua", "UPDATE_AFTER",
+        "d", "DELETE").asArgument("op_mapping")
+);
+```
 
 ## TO_CHANGELOG
 

flink-python/pyflink/table/tests/test_table_completeness.py

@@ -42,6 +42,7 @@ def excluded_methods(cls):
             'asArgument',
             'process',
             'partitionBy',
+            'fromChangelog',
         }
 
     @classmethod

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

@@ -1454,4 +1454,31 @@ default TableResult executeInsert(
      * @return an append-only {@link Table} with an {@code op} column prepended to the input columns
      */
     Table toChangelog(Expression... arguments);
+
+    /**
+     * 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);
 }

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

@@ -72,6 +72,8 @@ public final class BuiltInFunctionDefinition implements SpecializedFunction {
 
     private final SqlCallSyntax sqlCallSyntax;
 
+    private final @Nullable ChangelogModeStrategy changelogModeStrategy;
+
     private final String sqlName;
 
     private BuiltInFunctionDefinition(
@@ -84,7 +86,8 @@ private BuiltInFunctionDefinition(
             boolean isDeterministic,
             boolean isRuntimeProvided,
             String runtimeClass,
-            boolean isInternal) {
+            boolean isInternal,
+            @Nullable ChangelogModeStrategy changelogModeStrategy) {
         this.name = checkNotNull(name, "Name must not be null.");
         this.sqlName = sqlName;
         this.version = isInternal ? null : version;
@@ -95,6 +98,7 @@ private BuiltInFunctionDefinition(
         this.runtimeClass = runtimeClass;
         this.isInternal = isInternal;
         this.sqlCallSyntax = sqlCallSyntax;
+        this.changelogModeStrategy = changelogModeStrategy;
         validateFunction(this.name, this.version, this.isInternal);
     }
 
@@ -131,6 +135,14 @@ public boolean isInternal() {
         return isInternal;
     }
 
+    /**
+     * 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<ChangelogModeStrategy> getChangelogModeStrategy() {
+        return Optional.ofNullable(changelogModeStrategy);
+    }
+
     public String getQualifiedName() {
         if (isInternal) {
             return name;
@@ -253,6 +265,8 @@ public static final class Builder {
 
         private SqlCallSyntax sqlCallSyntax = SqlCallSyntax.FUNCTION;
 
+        private @Nullable ChangelogModeStrategy changelogModeStrategy;
+
         public Builder() {
             // default constructor to allow a fluent definition
         }
@@ -399,6 +413,15 @@ public Builder sqlName(String name) {
             return this;
         }
 
+        /**
+         * 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 changelogModeStrategy(ChangelogModeStrategy changelogModeStrategy) {
+            this.changelogModeStrategy = changelogModeStrategy;
+            return this;
+        }
+
         public BuiltInFunctionDefinition build() {
             return new BuiltInFunctionDefinition(
                     name,
@@ -410,7 +433,8 @@ public BuiltInFunctionDefinition build() {
                     isDeterministic,
                     isRuntimeProvided,
                     runtimeClass,
-                    isInternal);
+                    isInternal,
+                    changelogModeStrategy);
         }
     }
 }

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

@@ -27,6 +27,7 @@
 import org.apache.flink.table.api.JsonType;
 import org.apache.flink.table.api.JsonValueOnEmptyOrError;
 import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.connector.ChangelogMode;
 import org.apache.flink.table.expressions.TimeIntervalUnit;
 import org.apache.flink.table.expressions.TimePointUnit;
 import org.apache.flink.table.expressions.ValueLiteralExpression;
@@ -106,6 +107,7 @@
 import static org.apache.flink.table.types.inference.TypeStrategies.varyingString;
 import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.ARRAY_ELEMENT_ARG;
 import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.ARRAY_FULLY_COMPARABLE;
+import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.FROM_CHANGELOG_INPUT_TYPE_STRATEGY;
 import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.INDEX;
 import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.JSON_ARGUMENT;
 import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.ML_PREDICT_INPUT_TYPE_STRATEGY;
@@ -115,6 +117,7 @@
 import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.percentage;
 import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.percentageArray;
 import static org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.ARRAY_APPEND_PREPEND;
+import static org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.FROM_CHANGELOG_OUTPUT_TYPE_STRATEGY;
 import static org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.ML_PREDICT_OUTPUT_TYPE_STRATEGY;
 import static org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.TO_CHANGELOG_OUTPUT_TYPE_STRATEGY;
 
@@ -809,6 +812,30 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
                             "org.apache.flink.table.runtime.functions.ptf.ToChangelogFunction")
                     .build();
 
+    public static final BuiltInFunctionDefinition FROM_CHANGELOG =
+            BuiltInFunctionDefinition.newBuilder()
+                    .name("FROM_CHANGELOG")
+                    .kind(PROCESS_TABLE)
+                    .staticArguments(
+                            StaticArgument.table(
+                                    "input",
+                                    Row.class,
+                                    false,
+                                    EnumSet.of(
+                                            StaticArgumentTrait.TABLE,
+                                            StaticArgumentTrait.ROW_SEMANTIC_TABLE)),
+                            StaticArgument.scalar("op", DataTypes.DESCRIPTOR(), true),
+                            StaticArgument.scalar(
+                                    "op_mapping",
+                                    DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()),
+                                    true))
+                    .changelogModeStrategy(ctx -> ChangelogMode.all())
+                    .inputTypeStrategy(FROM_CHANGELOG_INPUT_TYPE_STRATEGY)
+                    .outputTypeStrategy(FROM_CHANGELOG_OUTPUT_TYPE_STRATEGY)
+                    .runtimeClass(
+                            "org.apache.flink.table.runtime.functions.ptf.FromChangelogFunction")
+                    .build();
+
     public static final BuiltInFunctionDefinition GREATEST =
             BuiltInFunctionDefinition.newBuilder()
                     .name("GREATEST")

flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/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.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.functions.ChangelogFunction.ChangelogContext;
+import org.apache.flink.table.types.inference.TypeStrategy;
+
+/**
+ * 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 ChangelogFunction} interface.
+ */
+@Internal
+public interface ChangelogModeStrategy {
+
+    /** Infers the output {@link ChangelogMode} based on the given {@link ChangelogContext}. */
+    ChangelogMode inferChangelogMode(ChangelogContext changelogContext);
+}