[FLINK-37965][FLINK-37889][FLINK-37966] Add PhysicalMultiJoinRule, FlinkLogicalMultiJoin, LogicalJoinToMultiJoinRule and Semantic + Restore + relPlan + execPlan tests #26687

gustavodemorais · 2025-06-17T08:57:28Z

What is the purpose of the change

Add PhysicalMultiJoinRule, FlinkLogicalMultiJoin, LogicalJoinToMultiJoinRule + e2e Semantic, Restore, relPlan, execPlan and explain tests.

This covers FLINK-37965-FLINK-37889-FLINK-37966

This is the smallest group of changes that makes sense and enables us to test e2e. Out of scope to try not to increase the the PR size even more:

use right join to left join rule + tests for it it can be added in another PR.
RelTimeIndicatorConverter
Extend JoinToMultiJoinRule to check for common join key

Brief change log

Add PhysicalMultiJoinRule
Add FlinkLogicalMultiJoin
Add LogicalJoinToMultiJoinRule

Verifying this change

Added Semantic tests
Added Restore tests
Added verifyRelPlan tests
Added verifyExecPlan tests
Added verifyExplain 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? (no)
If yes, how is the feature documented? (docs / JavaDocs)

flinkbot · 2025-06-17T09:04:55Z

CI report:

44fe2a3 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

twalthr

Excellent PR @gustavodemorais! Well tested, well documented, and scoped. I mostly added nit comments.

...r/src/main/java/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalMultiJoin.java

...in/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinForReorderRule.java

twalthr · 2025-06-17T14:15:47Z

...in/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinForReorderRule.java

@@ -99,24 +99,24 @@
 * @see CoreRules#JOIN_TO_MULTI_JOIN
 */
 @Value.Enclosing
-public class FlinkJoinToMultiJoinRule extends RelRule<FlinkJoinToMultiJoinRule.Config>
+public class JoinToMultiJoinForReorderRule extends RelRule<JoinToMultiJoinForReorderRule.Config>


update JavaDoc?

This rule is copied from {@link org.apache.calcite.rel.rules.JoinToMultiJoinRule}

This is the rule used for reordering and I just tried to give it a better name. The rule is mostly copied from {@link org.apache.calcite.rel.rules.JoinToMultiJoinRule} with some modifications. The javadoc is still accurate sinc.

That said, I fixed the typos I've found.

...ner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java

twalthr · 2025-06-17T14:19:41Z

...ner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java

+    @Override
+    public boolean matches(RelOptRuleCall call) {
+        final Join origJoin = call.rel(0);
+        if (origJoin.getJoinType() != JoinRelType.INNER


I remember we had some issues with the multi way join rules and FOR SYSTEM TIME AS OF, is it guaranteed that lookups joins are not merged into multijoin?

Potentially also interval joins should not be pulled into the multijoin.

Yes, we avoid matching temporal joins. I basically had done the same we do for binary joins, see

flink/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalMultiJoinRule.java

Line 71 in d87ff8c

return false;

.

I took a look again and I agree this doesn't look complete. I've added the necessary checks in the logical rule + tests for it, see e0ccc9c

If this look good to you I could add this + tests for the reorder rule in another ticket

twalthr · 2025-06-17T14:44:51Z

...c/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinTestPrograms.java

+                                            Row.ofKind(RowKind.INSERT, "1", "order0", "ProdB"),
+                                            Row.ofKind(RowKind.INSERT, "1", "order1", "ProdA"),
+                                            Row.ofKind(RowKind.INSERT, "2", "order2", "ProdB"),
+                                            Row.ofKind(RowKind.INSERT, "3", "order3", "ProdC"),
+                                            Row.ofKind(RowKind.INSERT, "4", "order4", "ProdD"),
+                                            Row.ofKind(RowKind.INSERT, "6", "order6", "ProdF"),
+                                            Row.ofKind(RowKind.INSERT, "7", "order7", "ProdG"),
+                                            Row.ofKind(RowKind.INSERT, "9", "order9", "ProdA"))


looks like the ids are mostly perfectly ordered and also always at column 0, I suggest to reorder some and move columns also to the middle or end.

I've shuffled rows and the columns for the four way complex join for restore and semantic test. Also shuffled columns for the rel/exec/explain tests 👍 570da7b

...r/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java

...way-complex-updating-join-with-restore/plan/four-way-complex-updating-join-with-restore.json

twalthr · 2025-06-17T14:54:38Z

...c-multi-join_1/three-way-inner-join-with-restore/plan/three-way-inner-join-with-restore.json

+        "rightFieldIndex" : 0
+      } ]
+    },
+    "inputUpsertKeys" : [ [ ], [ ], [ ] ],


some comment as able: #26683

In this case, I also added @JsonInclude(JsonInclude.Include.NON_EMPTY) to inputUpsertkeys. However, we'll still have it in the compiled plan because it's not an empty list and has the "relevant information" that we have three inputs.

We can add a workaround like using value = JsonInclude.Include.CUSTOM or some custom logic in the ExecNode but I wasn't sure if it's worth it for the readability for this case. It's always going to be a relatively short list of "empty lists". Wdyt?

Not worth the effort. I agree.

...he/flink/table/runtime/operators/join/stream/keyselector/AttributeBasedJoinKeyExtractor.java

twalthr · 2025-06-18T12:46:05Z

...c/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinTestPrograms.java

-                                            "+I[3, Nomad, order3, payment3, New York]")
+                                            "+I[1, Gus, order0, payment1, London]"
+                                            // "+I[1, Gus, order1, payment1, London]" // TODO
+                                            // Gustavo Why is this being consumed after the restore?


This should not end up in the Flink code base. Could you figure it out?

I've removed it, since you're squashing the PR then it won't land in the codebase nor in the history

twalthr

LGTM for the first version. Thanks @gustavodemorais!

gustavodemorais · 2025-06-18T13:14:35Z

LGTM for the first version. Thanks @gustavodemorais!

Thanks for precise and useful reviews, @twalthr!

Yeah, let's start with this initial version and I'm constantly mapping new tasks under FLINK-37859 and continuing the work to make this production ready.

gustavodemorais added 3 commits June 17, 2025 10:41

[FLINK-37965][table] Add PhysicalMultiJoinRule and FlinkLogicalMultiJoin

f8f4e27

[FLINK-37889][table] Add LogicalJoinToMultiJoinRule

8140d18

[FLINK-37966][table] Add semantic tests and restore tests

4bcdb6c

[FLINK-37966][table] Add verifyRelPlan and verifyExecPlan tests

d87ff8c

gustavodemorais force-pushed the FLINK-37965-FLINK-37889-FLINK-37966 branch from fdf0bc7 to d87ff8c Compare June 17, 2025 13:52

twalthr reviewed Jun 17, 2025

View reviewed changes

gustavodemorais added 4 commits June 17, 2025 21:17

[FLINK-37966][table] Add tests to check to check temporal joins

e0ccc9c

[FLINK-37966][table] Refactor comments and multiple code nits

1e03689

[FLINK-37966][table] Generate restore tests' artifacts again

5a478e6

[FLINK-37966][table] Rename test .xml for renamed reorder rule

23a207d

gustavodemorais mentioned this pull request Jun 18, 2025

[FLINK-37889] [table-planner] Add JoinToMultiJoinRule #26689

Open

gustavodemorais added 2 commits June 18, 2025 13:15

[FLINK-37966][table] Shuffled primary columns and rows for tests

570da7b

[FLINK-37966][table] Add tests with WHERE clause

44fe2a3

twalthr reviewed Jun 18, 2025

View reviewed changes

twalthr approved these changes Jun 18, 2025

View reviewed changes

[FLINK-37965][FLINK-37889][FLINK-37966] Add PhysicalMultiJoinRule, FlinkLogicalMultiJoin, LogicalJoinToMultiJoinRule and Semantic + Restore + relPlan + execPlan tests #26687

Are you sure you want to change the base?

[FLINK-37965][FLINK-37889][FLINK-37966] Add PhysicalMultiJoinRule, FlinkLogicalMultiJoin, LogicalJoinToMultiJoinRule and Semantic + Restore + relPlan + execPlan tests #26687

Conversation

gustavodemorais commented Jun 17, 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 Jun 17, 2025 • 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!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gustavodemorais Jun 17, 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!

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!

gustavodemorais Jun 18, 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!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gustavodemorais Jun 17, 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!

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!

gustavodemorais commented Jun 18, 2025

Uh oh!

Uh oh!

gustavodemorais commented Jun 17, 2025 •

edited

Loading

flinkbot commented Jun 17, 2025 •

edited

Loading

gustavodemorais Jun 17, 2025 •

edited

Loading

gustavodemorais Jun 18, 2025 •

edited

Loading

gustavodemorais Jun 17, 2025 •

edited

Loading