[FLINK-39349][table] Support TO_CHANGELOG retract/upsert stream to upsert stream with deletion flag (4.2.2.1) by gustavodemorais · Pull Request #27847 · apache/flink

gustavodemorais · 2026-03-29T15:56:57Z

What is the purpose of the change

Extends TO_CHANGELOG's op_mapping parameter to support comma-separated RowKind names as keys, enabling multiple RowKinds to map to the same output code. This implements FLIP-564 section 4.2.2.1 - the deletion flag pattern (Kafka Connect
style):

  SELECT * FROM TO_CHANGELOG(                                                                                                                                                                                                                       
    input => TABLE t PARTITION BY id,                                                                                                                                                                                                               
    op => DESCRIPTOR(deleted),                                                                                                                                                                                                                      
    op_mapping => MAP['INSERT, UPDATE_AFTER', 'false', 'DELETE', 'true']                                                                                                                                                                            
  )

Brief change log

Support comma-separated RowKind names in op_mapping keys (validation and runtime)
Validate duplicate RowKinds across entries at planning time
Add semantic tests for deletion flag pattern and duplicate RowKind validation
Document the new capability and add examples (SQL, Table API, JavaDoc)

Verifying this change

Added DELETION_FLAG semantic test: retract source with comma-separated keys, verifies correct mapping and that unmapped UPDATE_BEFORE is dropped
- Added DUPLICATE_ROW_KIND validation test: verifies planning fails when a RowKind appears in multiple entries

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)

…ELOG op_mapping

flinkbot · 2026-03-29T16:01:38Z

CI report:

e47fadb Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

twalthr · 2026-03-30T07:26:08Z

+                            throwOnFailure,
+                            String.format(
+                                    "Invalid target mapping for argument 'op_mapping'. "
+                                            + "Unknown RowKind: '%s'. Valid values are: %s.",


RowKind is a Java class. And is confusing to SQL persona. This is why it called op:

Suggested change

+ "Unknown RowKind: '%s'. Valid values are: %s.",

+ "Unknown change operation: '%s'. Valid values are: %s.",

twalthr · 2026-03-30T07:26:18Z

+                            throwOnFailure,
+                            String.format(
+                                    "Invalid target mapping for argument 'op_mapping'. "
+                                            + "Duplicate RowKind: '%s'.",


Suggested change

+ "Duplicate RowKind: '%s'.",

+ "Duplicate change operation: '%s'.",

please update all other locations

…ser-facing text

gustavodemorais · 2026-03-30T08:13:32Z

Thanks for the review, @twalthr 🙂 Addressed your comments, take a look

featzhang · 2026-03-30T08:15:13Z

-        final Map<RowKind, String> map = new EnumMap<>(RowKind.class);
-        opMapping.forEach((name, code) -> map.put(RowKind.valueOf(name), code));
-        return map;
+        final Map<RowKind, String> result = new EnumMap<>(RowKind.class);


Recommend adding logging for conflicts or invalid mappings to help users debug configuration issues.

Hey, for TO_CHANGELOG invalid mappings are already caught at planning time by ToChangelogTypeStrategy, the user will see a validation error

featzhang · 2026-03-30T08:16:24Z

            TableTestProgram.of(
                            "to-changelog-invalid-op-mapping",
-                            "fails when op_mapping has invalid RowKind name")
+                            "fails when op_mapping has invalid change operation name")


Recommend adding notes on why UPDATE_BEFORE is dropped and practical use cases of deletion flag pattern.

featzhang · 2026-03-30T08:17:37Z

+     * be valid and appear at most once across all entries.
+     */
+    @SuppressWarnings("rawtypes")
+    private static Optional<List<DataType>> validateOpMappingKeys(


Recommend adding comments about whitespace and case handling for better user understanding.

I've updated the comment to be more specific about this e8d4956

featzhang · 2026-03-30T08:18:07Z

     *     map("INSERT", "I", "UPDATE_AFTER", "U").asArgument("op_mapping")
     * );
+     *
+     * // Deletion flag pattern: comma-separated keys map multiple change operations to the same code


uggest enhancing Javadoc to clarify how multiple RowKinds mapping to the same output may affect UPDATE_BEFORE behavior.

Thanks for the review, Feat Zhang. The existing JavaDoc and documentation already explain that unmapped operations are dropped. Adding UPDATE_BEFORE-specific details to the generic toChangelog() JavaDoc would be too narrow, since it depends on the use case

…ng validation JavaDoc

twalthr

Thanks @gustavodemorais for the additional improvements. Will merge now...

This closes apache#27847.

gustavodemorais added 2 commits March 29, 2026 17:48

[FLINK-39349][table] Support comma-separated RowKind keys in TO_CHANG…

097a978

…ELOG op_mapping

[FLINK-39349][table] Document deletion flag pattern for TO_CHANGELOG

5c04b7f

twalthr reviewed Mar 30, 2026

View reviewed changes

[FLINK-39349][table] Use 'change operation' instead of 'RowKind' in u…

305a6f0

…ser-facing text

gustavodemorais requested a review from twalthr March 30, 2026 08:13

twalthr approved these changes Mar 30, 2026

View reviewed changes

featzhang reviewed Mar 30, 2026

View reviewed changes

github-actions Bot added the community-reviewed PR has been reviewed by the community. label Mar 30, 2026

[FLINK-39349][table] Clarify whitespace and case handling in op_mappi…

e47fadb

…ng validation JavaDoc

gustavodemorais force-pushed the FLINK-39349 branch from e8d4956 to e47fadb Compare March 30, 2026 09:00

twalthr approved these changes Mar 30, 2026

View reviewed changes

twalthr merged commit 4734953 into apache:master Mar 30, 2026

featzhang pushed a commit to featzhang/flink that referenced this pull request Mar 30, 2026

[FLINK-39349][table] Support deletion flag in TO_CHANGELOG

50e6f71

This closes apache#27847.

och5351 pushed a commit to och5351/flink that referenced this pull request Mar 31, 2026

[FLINK-39349][table] Support deletion flag in TO_CHANGELOG

5b9a544

This closes apache#27847.

Samrat002 pushed a commit to Samrat002/flink that referenced this pull request Mar 31, 2026

[FLINK-39349][table] Support deletion flag in TO_CHANGELOG

4488c2c

This closes apache#27847.

	+ "Unknown RowKind: '%s'. Valid values are: %s.",
	+ "Unknown change operation: '%s'. Valid values are: %s.",

	+ "Duplicate RowKind: '%s'.",
	+ "Duplicate change operation: '%s'.",

Conversation

gustavodemorais commented Mar 29, 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 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

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 30, 2026

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!

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

flinkbot commented Mar 29, 2026 •

edited

Loading