[FLINK-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) by gustavodemorais · Pull Request #27777 · apache/flink

gustavodemorais · 2026-03-16T16:17:42Z

What is the purpose of the change

This is the first implementation of TO_CHANGELOG. The idea is to implement the use cases one by one interactively. With this ticket we will write the first version which allows flat mapping of upsert stream.

Implement TO_CHANGELOG built-in PTF as specified in FLIP-564, section 4.2.2.2 (retract/upsert stream to upsert stream, flat mode).

TO_CHANGELOG converts a dynamic table (retract/upsert stream) into an append-only stream with an explicit operation code column. Each input row is emitted as an INSERT-only row with a string op column indicating the original RowKind. This is stateless - each event maps directly to one output record.

Brief change log

- Built-in function definition and type strategies
- Runtime implementation
- Table API interface
- MAP literal support in OperatorBindingCallContext and CallBindingCallContext
- Plan tests, semantic tests (SQL, Table API, convenience API)
- Use-case test: LAG on upsert stream (previously impossible without TO_CHANGELOG, like other aggregation functions)
- Documentation

Verifying this change

Plan tests
Semantic tests

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (docs / JavaDocs)

…ations

…finition

flinkbot · 2026-03-16T16:21:35Z

CI report:

35fd839 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

twalthr

Feedback for 24a9bed

twalthr · 2026-03-16T16:42:13Z

...src/main/java/org/apache/flink/table/planner/functions/inference/CallBindingCallContext.java

+        final List<SqlNode> operands = mapCall.getOperandList();
+        final Map<String, String> map = new LinkedHashMap<>();
+        for (int i = 0; i < operands.size(); i += 2) {
+            final String key = SqlLiteral.unchain(operands.get(i)).getValueAs(String.class);


What happens if this unchaining fails? A map can only be literal if all children are also literal, otherwise the getArgumentValue must return Optional.empty()

how well is null supported in keys and values? officially we support them in Flink.

Let's catch exceptions here and also check for nulls

twalthr · 2026-03-16T16:42:50Z

...src/main/java/org/apache/flink/table/planner/functions/inference/CallBindingCallContext.java

    }

    @Override
    public boolean isArgumentLiteral(int pos) {


update isArgumentLiteral as well? A map can only be literal if all children are also literal

twalthr

Feedback for 6efff33

twalthr · 2026-03-16T17:25:56Z

...src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java

+        if (opDescriptor.isPresent() && opDescriptor.get().getNames().size() != 1) {
+            return callContext.fail(
+                    throwOnFailure,
+                    "OP descriptor must contain exactly one column name "


The descriptor for argument 'op' must contain exactly one column name

twalthr · 2026-03-16T17:26:52Z

...src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java

+                if (!(key instanceof String) || !VALID_ROW_KIND_NAMES.contains(key)) {
+                    return callContext.fail(
+                            throwOnFailure,
+                            "Invalid RowKind name '%s' in op_mapping. Valid names are: %s.",


"Invalid target mapping for argument 'op_mapping'."

twalthr

Feedback for 8d295e1

twalthr · 2026-03-16T17:29:11Z

...-planner/src/main/java/org/apache/flink/table/planner/functions/ptf/ToChangelogFunction.java

+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.ptf;


put into org.apache.flink.table.runtime.functions.ptf in table-runtime

twalthr · 2026-03-16T17:31:36Z

...-planner/src/main/java/org/apache/flink/table/planner/functions/ptf/ToChangelogFunction.java

+ * partition key columns automatically.
+ */
+@Internal
+public class ToChangelogFunction extends ProcessTableFunction<Row> {


BuiltInProcessTableFunction similar to BuiltInScalarFunction

You probably mean we should add a new BuiltInProcessTableFunction similar to BuiltInScalarFunction as start using it as a base class?

twalthr · 2026-03-16T17:33:07Z

...-planner/src/main/java/org/apache/flink/table/planner/functions/ptf/ToChangelogFunction.java

+        collect(buildOutputRow(input, opCode));
+    }
+
+    private Row buildOutputRow(final Row input, final String opCode) {


once you use BuiltIn stack, everything should be RowData. as efficiently as possible with as little conversion as possible

Done. It was necessary to do one more change to the framework to support this: TypeInferenceUtil.inferInputTypes() (line 484) returns semantics.dataType() for table arguments, which always has Row as conversion class - it ignores the conversionClass specified in StaticArgument.table(). Adding bridgedTo(expectedArg.getConversionClass()).

#27777 (comment)

twalthr

Thank you @gustavodemorais. Sorry for being so picky here but this is the first real built-in PTF so we need to prepare a good template for the future.

twalthr · 2026-03-16T17:35:17Z

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


+        @Override
+        public Table toChangelog() {
+            return process("TO_CHANGELOG");


don't hard code the name here. use the built in function definition. I know this was an issue for ml predict, but we should fix it properly now.

twalthr · 2026-03-16T17:38:15Z

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

        }

+        @Override
+        public Table toChangelog() {


add op_mapping parameter and op, same functionality as in sql. to avoid to many overloads, we can accept Expression and force people to use .asArgument.

twalthr · 2026-03-16T17:38:49Z

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

+
+    /**
+     * Converts this dynamic table into an append-only stream with an explicit operation code column
+     * using the built-in {@link BuiltInFunctionDefinitions#TO_CHANGELOG TO_CHANGELOG} PTF.


Remove references to BuiltInFunctionDefinitions#TO_CHANGELOG TO_CHANGELOG. BuiltInFunctionDefinitions is rather internal.

…ion and other feedback

gustavodemorais · 2026-03-17T17:08:51Z

Hey @twalthr, thanks for the review! I've addressed all your comments here ea476c3

gustavodemorais added 6 commits March 16, 2026 16:59

[FLINK-39259][table] Add MAP literal support to CallContext implement…

24a9bed

…ations

[FLINK-39259][table] Add TO_CHANGELOG type strategies and built-in de…

6efff33

…finition

[FLINK-39259][table] Add TO_CHANGELOG runtime implementation

8d295e1

[FLINK-39259][table] Add TO_CHANGELOG Table API

059c846

[FLINK-39259][table] Add TO_CHANGELOG plan and semantic tests

d97f6bc

[FLINK-39259][table] Add TO_CHANGELOG documentation

06ac092

twalthr reviewed Mar 16, 2026

View reviewed changes

[FLINK-39259][table] Use RowData for TO_CHANGELOG runtime, move funct…

ea476c3

…ion and other feedback

gustavodemorais requested a review from twalthr March 17, 2026 17:10

[FLINK-39259][table] Add REQUIRE_UPDATE_BEFORE and retract stream tests

35fd839

Conversation

gustavodemorais commented Mar 16, 2026

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Mar 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gustavodemorais commented Mar 17, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

flinkbot commented Mar 16, 2026 •

edited

Loading