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

Author: raminqaf

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

@@ -30,9 +30,103 @@ 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
+// Default: reads 'op' column with standard change operation names
+Table result = cdcStream.fromChangelog();
+```
 
 ## 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

@@ -21,6 +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.InputTypeStrategy;
 import org.apache.flink.table.types.inference.StaticArgument;
@@ -33,6 +35,7 @@
 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;
@@ -72,6 +75,8 @@ public final class BuiltInFunctionDefinition implements SpecializedFunction {
 
     private final SqlCallSyntax sqlCallSyntax;
 
+    private final @Nullable Function<ChangelogContext, ChangelogMode> changelogModeResolver;
+
     private final String sqlName;
 
     private BuiltInFunctionDefinition(
@@ -84,7 +89,8 @@ private BuiltInFunctionDefinition(
             boolean isDeterministic,
             boolean isRuntimeProvided,
             String runtimeClass,
-            boolean isInternal) {
+            boolean isInternal,
+            @Nullable Function<ChangelogContext, ChangelogMode> changelogModeResolver) {
         this.name = checkNotNull(name, "Name must not be null.");
         this.sqlName = sqlName;
         this.version = isInternal ? null : version;
@@ -95,6 +101,7 @@ private BuiltInFunctionDefinition(
         this.runtimeClass = runtimeClass;
         this.isInternal = isInternal;
         this.sqlCallSyntax = sqlCallSyntax;
+        this.changelogModeResolver = changelogModeResolver;
         validateFunction(this.name, this.version, this.isInternal);
     }
 
@@ -131,6 +138,14 @@ public boolean isInternal() {
         return isInternal;
     }
 
+    /**
+     * Returns the optional changelog mode resolver 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 String getQualifiedName() {
         if (isInternal) {
             return name;
@@ -253,6 +268,8 @@ public static final class Builder {
 
         private SqlCallSyntax sqlCallSyntax = SqlCallSyntax.FUNCTION;
 
+        private @Nullable Function<ChangelogContext, ChangelogMode> changelogModeResolver;
+
         public Builder() {
             // default constructor to allow a fluent definition
         }
@@ -399,6 +416,16 @@ public Builder sqlName(String name) {
             return this;
         }
 
+        /**
+         * Sets a resolver that determines the output {@link ChangelogMode} 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;
+            return this;
+        }
+
         public BuiltInFunctionDefinition build() {
             return new BuiltInFunctionDefinition(
                     name,
@@ -410,7 +437,8 @@ public BuiltInFunctionDefinition build() {
                     isDeterministic,
                     isRuntimeProvided,
                     runtimeClass,
-                    isInternal);
+                    isInternal,
+                    changelogModeResolver);
         }
     }
 }

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))
+                    .changelogModeResolver(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")