[FLINK-38511][table] Join that consumes cdc source without delete may be converted to delta join #27111

xuyangzhong · 2025-10-14T03:19:50Z

What is the purpose of the change

Join with the following pattern can be converted to delta join:

All two cdc sources output changelog mode without delete
Pks of these two cdc sources all contain join key, and then upstream nodes of join do not require sending update before

Brief change log

Support to optimize to delta join when join can match the optimization pattern.
Add UT, IT and Harness test for this optimization.

Verifying this change

Tests are added to verify this pr

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? no
If yes, how is the feature documented?

flinkbot · 2025-10-14T03:26:06Z

CI report:

df25ab0 UNKNOWN
cc55f42 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

… be converted to delta join

xuyangzhong · 2025-10-22T08:39:51Z

...anner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/DeltaJoinITCase.scala

-    val expected = List("+I[1.0, 1, 2022-02-02T02:02:02, 1, 1.0, 2022-02-02T02:02:22]")
-
-    // could not optimize into delta join because join keys do not contain indexes strictly
-    assertThatThrownBy(


I removed these expected failed tests because these exceptions have been tested in DeltaJoinTest. And I hope the IT tests can purely test the dataset, not for plan.

Au-Miner · 2025-10-23T07:02:00Z

...src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeltaJoin.java

    private static final String FIELD_NAME_RIGHT_JOIN_KEYS = "rightJoinKeys";

    private static final String FIELD_NAME_LOOKUP_RIGHT_TABLE_JOIN_SPEC =
            "lookupRightTableJoinSpec";


I don't quite understand why the DELTA_JOIN_TRANSFORMATION here doesn't use deltaJoin but to use delta-join.

Actually, I'm not entirely sure either. 🧐 I referenced CommonExecLookupJoin and CommonExecSink, and I noticed that the transformations use "-" as separators, while the field names for JSON are in camel case. So, I decided to follow that convention as well.

...src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeltaJoin.java

Au-Miner · 2025-10-23T07:21:51Z

...ink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/DeltaJoinUtil.java

+            return false;
+        }
+
+        // if this join output cdc records, the non-equiv condition must be applied on upsert key


if this join output cdc records and has non-equiv condition, upsert key must contain non-equiv condition

Besides, I don't quite understand why there is such a restriction here

You raised a very important issue. You can refer FlinkChangelogModeInferenceProgram.scala to understand the problem better. The essence of the issue is whether we can allow the upstream to omit sending Update Before(UB) when a filter is present. We can break down the filtering condition into two scenarios: 1. Filtering only on upsert key columns, and 2. Filtering on non-upsert keys. The distinction between these two cases is that in scenario 1, we can enable further optimization for the upstream and avoid sending UB.

Currently, in the FlinkChangelogModeInferenceProgram, we only check the filter on the Calc node. However, there are similar filters present on the join node (which are divided into join keys and non-equivalent join conditions), and this aspect has been overlooked. To prevent expanding the scope of this pull request, I only ensure that a delta join will not be optimized here (since Delta Join cannot consume UB messages) without altering the existing logic for streaming join.

As for why we don't need to address the join key, that has already been handled here: FlinkChangelogModeInferenceProgram.scala.

By the way, the comment here should emphasize that the columns in the non-equivalent condition must all come from the same set of upsert keys. Therefore, I will modify it to:
If this join outputs cdc records and has non-equiv condition, the reference columns in the non-equiv condition must come from the same set of upsert keys.

The reason I did not reuse the logic of isNonUpsertKeyCondition from FlinkChangelogModeInferenceProgram is that it only considers a single upsert key (as seen in this line). In this case, any set of upsert keys can be applicable.

Au-Miner · 2025-10-23T07:31:52Z

...le-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DeltaJoinTest.scala

  }

+  @Test
+  def testFilterOnNonUpsertKeysAfterJoinWithCdcSourceWithoutDelete(): Unit = {


Can we set deltaJoin to force and use assertThatThrownBy to display the error reason in the test for failing to convert to deltaJoin

Currently, we are unable to provide detailed error logs to explain why optimization to Delta Join is not supported. For more information, please refer to FLINK-37954. As for now, I think it's acceptable to display the original Join in the plan instead of the DeltaJoin. WDYT?

Ok, I think it's indeed worth further improvement

Au-Miner · 2025-10-23T08:22:52Z

Thanks for advancing the feature. Let me leave some comments

Au-Miner

Thanks for addressing the comment. It LGTM.

xuyangzhong mentioned this pull request Oct 14, 2025

[FLINK-38504][table] Add restore tests for delta join #27117

Open

xuyangzhong force-pushed the delta_join_support_update2 branch from 939312f to 1092702 Compare October 22, 2025 08:35

xuyangzhong marked this pull request as ready for review October 22, 2025 08:35

xuyangzhong force-pushed the delta_join_support_update2 branch from 1092702 to c65f4d2 Compare October 22, 2025 08:36

[FLINK-38511][table] Join that consumes cdc source without delete may…

3d28928

… be converted to delta join

xuyangzhong force-pushed the delta_join_support_update2 branch from c65f4d2 to 3d28928 Compare October 22, 2025 08:37

xuyangzhong commented Oct 22, 2025

View reviewed changes

Au-Miner reviewed Oct 23, 2025

View reviewed changes

address comment

cc55f42

xuyangzhong force-pushed the delta_join_support_update2 branch from 6cf7fdf to cc55f42 Compare October 23, 2025 11:02

github-actions bot added the community-reviewed PR has been reviewed by the community. label Oct 23, 2025

Au-Miner approved these changes Oct 24, 2025

View reviewed changes

xuyangzhong merged commit 5543195 into apache:master Oct 27, 2025

xuyangzhong mentioned this pull request Oct 28, 2025

[FLINK-38556][table] Support filter and project between source and delta join #27159

Merged

Uh oh!

[FLINK-38511][table] Join that consumes cdc source without delete may be converted to delta join #27111

[FLINK-38511][table] Join that consumes cdc source without delete may be converted to delta join #27111

Uh oh!

Conversation

xuyangzhong commented Oct 14, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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 Oct 14, 2025 • 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!

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!

xuyangzhong Oct 23, 2025 • 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!

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!

Au-Miner commented Oct 23, 2025

Uh oh!

Au-Miner left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

xuyangzhong commented Oct 14, 2025 •

edited

Loading

flinkbot commented Oct 14, 2025 •

edited

Loading

xuyangzhong Oct 23, 2025 •

edited

Loading