[flink] Support FLIP-510 key-only deletes on primary-key table sink by irodriguezclaveria · Pull Request #8452 · apache/paimon

irodriguezclaveria · 2026-07-03T15:45:44Z

Support key-only deletes on primary-key table sink (FLIP-510)

Advertise the FLIP-510 keyOnlyDeletes capability on primary-key table sinks so the Flink planner can drop the upstream ChangelogNormalize node when the source produces deletes by key.

This is especially useful for CDC sources, where the UPDATE_AFTER already carries the full row: dropping ChangelogNormalize removes its state and overhead with no loss of correctness.

Gated behind a new table option sink.key-only-deletes.enabled (default false), so existing plans are unchanged unless explicitly opted in. The API only exists in Flink 2.1+, so the behavior is wired through ChangelogModeUtils in paimon-flink1-common (no-op) and paimon-flink2-common (sets the flag), keeping paimon-flink-common compiling against both Flink 1.x and 2.x.

This PR has been tested in a real environment and drops correctly the ChangelogNormalize.

yunfengzhou-hub

Thanks for the PR. Left some comments as below.

yunfengzhou-hub · 2026-07-08T03:43:42Z

+
+    /** FLIP-510 key-only deletes, available from Flink 2.1. */
+    public static ChangelogMode.Builder enableKeyOnlyDeletes(ChangelogMode.Builder builder) {
+        return builder.keyOnlyDeletes(true);


This may throw NoSuchMethodException for Flink 2.0. Maybe an additional stub code is needed for paimon-flink-2.0 module.

yunfengzhou-hub · 2026-07-08T03:46:23Z

    }

+    @Test
+    public void testKeyOnlyDeletesEnabled() throws Exception {


The test coverage seems a little not enough. Maybe we can add some test cases that actually generate a Flink Plan from a SQL query, check whether ChangelogNormalizer has actually been removed through Table#explain, and check the correctness of result data.

yunfengzhou-hub · 2026-07-08T03:57:34Z


            if (options.get(MERGE_ENGINE) == MergeEngine.PARTIAL_UPDATE
                    && new CoreOptions(options).definedAggFunc()) {
                return requestedMode;


In the current implementation, if some prior conditions like this have made this method return in advance and ChangelogModeUtils.enableKeyOnlyDeletes is not invoked, users won't receive any notification, until the Flink job is started and users found that the ChangelogNormalizer operator is still there. Thus how about adding the following behavior?

When users set sink.key-only-deletes.enabled to true, but prior conditions lead to a different ChangelogMode, throw an exception or at least generate a warning log.

Apart from true and false, add a third option, auto, for sink.key-only-deletes.enabled, When set to auto, paimon will try its best to enable the optimization when condition is met, but it is not guaranteed. In this case, no exception would be thrown if it failed to achieve the optimization.

Also document these limitations in the description of the option.

yunfengzhou-hub · 2026-07-08T03:59:53Z

            return requestedMode;
        } else {
            Options options = Options.fromMap(table.options());
            if (options.get(CHANGELOG_PRODUCER) == ChangelogProducer.INPUT) {


It might be better to check whether these changelog producers and merge engines support enabling sink.key-only-deletes.enabled, and add tests to verify the correctness of results with different combinations of changelog producers and merge engines.

Thanks, good point. So far I have validated it with deduplicate + none, which works correctly.
For the other combinations, this is my understanding — could you confirm?

Changelog producers:

input: the docs say it expects the input to already be a complete changelog, so with key-only deletes it would forward incomplete deletes. Should we treat it as not compatible here?
lookup / full-compaction: both reconstruct the complete changelog, so I would expect them to work with key-only deletes.

Merge engines:

partial-update: I think it works as long as delete handling is enabled (remove-record-on-delete), since it merges and removes by key.
aggregate: this one probably does not work, because it needs the previous values to retract on updates and deletes, and a key-only delete does not carry them. Would you agree it should be blocked?
first-row: it ignores updates and deletes by design, so a key-only delete would have no effect. I understand it does not really apply here.

For the combinations that do not apply, I could document them as unsupported. I am not sure yet whether we also need a check or a test for that, or if documentation would be enough — what would you prefer?

For the other cases I will test adding the corresponding junits/IT.

Thanks

I think your idea is basically correct, that key-only deletes cannot work with input changelog producer or merge engines with aggregation semantics.

Apart from our discussion here, a complete test for all possible combinations with ParameterizedTest could help provide a better guarantee for the supported situations in the long term.

yunfengzhou-hub · 2026-07-08T04:04:22Z

+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "If true, a primary-key table sink advertises the FLIP-510 key-only "


I personally recommend not mentioning FLIP-510 in user document. It is mostly an API for developers. For users, mentioning the semantics change and performance benefits should be enough.

If you think this information is helpful for developers, It might be more proper to reference FLIP-510 in JavaDocs and comments.

irodriguezclaveria · 2026-07-08T08:29:56Z

Hi @yunfengzhou-hub , thanks for the review, I'm going to check your comments and I'll let you know when the changes are done. Thanks again.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

irodriguezclaveria · 2026-07-12T21:17:46Z

Hi @yunfengzhou-hub ,

Let me explain you the changes that I did:

On test coverage (generate a Flink plan and check ChangelogNormalize is removed via Table#explain + check data correctness):

I added end-to-end data correctness tests in KeyOnlyDeletesITCase:

One test with deduplicate + changelog-producer=none + sink.key-only-deletes.enabled=true, where the source sends updates only as +U (no -U) and deletes with only the primary key. It checks the final table content is correct.
A parameterized test over {deduplicate, partial-update} × {lookup, full-compaction} that checks both the batch result and the streaming reconstructed changelog (the -U/+U pair with the before-image, and the -D with the full row).

About checking ChangelogNormalize removal with explain: I tried it, but I could not make it observable. The option only changes the sink ChangelogMode. In a normal INSERT INTO paimon SELECT ... FROM source, the plan has no ChangelogNormalize in either case, because the Paimon PK sink already accepts upsert. So the plan does not change with the option on/off. The real effect of the capability is on the sink ChangelogMode, which is already covered by the unit test ChangelogModeTest.

On throwing an exception / warning when a prior condition changes the ChangelogMode:

Done. When sink.key-only-deletes.enabled=true but a prior condition prevents the optimization (no primary key, changelog-producer=input, merge-engine=aggregation, or partial-update with aggregation functions), we now log a warning and keep the job running (no exception). I also documented these limitations in the option description.

On the auto option (true / false / auto):

I thought about it, but I think true/false is enough. The current true already works as "best effort": it enables the optimization where possible, and where a prior condition prevents it, it just skips it and logs a warning — it never throws. So true already has the "auto" semantics. The only difference from the proposed auto would be hiding the warning, and the warning is exactly what tells the user why the optimization did not apply. So I would prefer to keep true/false.

On the combinations (input / aggregation / etc.):

I agree with your understanding: key-only deletes cannot work with input changelog producer or with aggregation merge engines. The code already blocks these cases. I also added a test in ChangelogModeTest to check that the option is ignored when changelog-producer=input.

On not mentioning FLIP-510 in the user docs:

Done. I removed FLIP-510 from the option description (user docs). I kept the reference only in JavaDocs/comments for developers, as you suggested.

Thanks!

yunfengzhou-hub

Thanks for the update. Generally LGTM except one small comment as below.

yunfengzhou-hub · 2026-07-14T03:28:41Z

+            <td><h5>sink.key-only-deletes.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>If true, a primary-key table sink advertises the key-only (partial) deletes capability, allowing the Flink planner to drop the upstream ChangelogNormalize node when the source produces deletes by key. Requires Flink 2.1+; no effect on Flink 1.x. Does not apply when the table has no primary key, when 'changelog-producer' is 'input', or when 'merge-engine' is 'aggregation' or 'partial-update' with aggregation functions; in those cases a warning is logged. Disabled by default.</td>


no effect on Flink 1.x.

Also no effect on Flink 2.0.

irodriguezclaveria · 2026-07-14T11:03:24Z

Hi @yunfengzhou-hub , thanks to review again! the change is done !

irodriguezclaveria added 8 commits July 3, 2026 16:49

add flink support FLIP-510

a68f115

Merge branch 'master' into add_support_to_flip_510

e170299

add parameter to use FLIP-510

85d1d96

add test to check regreassion

31c9efb

fix error

b67ee16

Re-trigger CI

7dc29b6

Re-trigger CI

7ff80ce

fix documentation

6ad1814

irodriguezclaveria marked this pull request as ready for review July 7, 2026 10:43

irodriguezclaveria changed the title ~~add flink support FLIP-510~~ [flink] Support FLIP-510 key-only deletes on primary-key table sink Jul 7, 2026

yunfengzhou-hub reviewed Jul 8, 2026

View reviewed changes

irodriguezclaveria and others added 2 commits July 12, 2026 20:07

add changes requests in the PRi

7a80e0e

Re-trigger CI

57e829a

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

irodriguezclaveria requested a review from yunfengzhou-hub July 13, 2026 08:42

yunfengzhou-hub approved these changes Jul 14, 2026

View reviewed changes

irodriguezclaveria added 2 commits July 14, 2026 09:23

fix documentation

e9f878c

Re-trigger CI

94185a6

Uh oh!

Conversation

irodriguezclaveria commented Jul 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

yunfengzhou-hub left a comment

Choose a reason for hiding this comment

Uh oh!

yunfengzhou-hub Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

yunfengzhou-hub Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

yunfengzhou-hub Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

yunfengzhou-hub Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

irodriguezclaveria Jul 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yunfengzhou-hub Jul 9, 2026

Choose a reason for hiding this comment

Uh oh!

yunfengzhou-hub Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

irodriguezclaveria commented Jul 8, 2026

Uh oh!

irodriguezclaveria commented Jul 12, 2026

Uh oh!

yunfengzhou-hub left a comment

Choose a reason for hiding this comment

Uh oh!

yunfengzhou-hub Jul 14, 2026

Choose a reason for hiding this comment

Uh oh!

irodriguezclaveria commented Jul 14, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

irodriguezclaveria commented Jul 3, 2026 •

edited

Loading

irodriguezclaveria Jul 8, 2026 •

edited

Loading