[FLINK-37814][table-planner] Add FlinkRightJoinToLeftJoinRule #26627

SteveStevenpoor · 2025-06-03T12:52:39Z

This PR is part of multi-way join implementation. See FLIP-516

What is the purpose of the change

StreamingMultiJoinOperator introduced in this PR supports only LEFT JOINs. Since in multi-way join left and right joins are practically the same we need a rule to change right join to the left ones before transforming join chain to single multijoin node.

Brief change log

Added FlinkRightJoinToLeftJoinRule

Verifying this change

Added simple test for rel plan verification after applying the rule

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? (not applicable)

SteveStevenpoor · 2025-06-03T12:54:12Z

@gustavodemorais Hey! Please, take a look at this. Also I'm waiting for the ticket with join to multi join rule since i'm also almost done with it.

flinkbot · 2025-06-03T12:59:04Z

CI report:

7182aa3 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

… need of multi-way join operator

davidradl · 2025-06-04T16:27:23Z

...ain/java/org/apache/flink/table/planner/plan/rules/logical/FlinkRightJoinToLeftJoinRule.java

+        super(config);
+    }
+
+    @Deprecated // to be removed before 2.0


I assume we are copying code that has deprecated methods that should have been removed for version 2.

Do we need to add these deprecated methods?

At the least we should remove the comment as 2.0 is now out - so anything that was not removed will not be until another version change.

Thank you for the remark! I removed these methods because in this case they are not really needed.

gustavodemorais · 2025-06-05T08:52:56Z

...ala/org/apache/flink/table/planner/plan/rules/logical/FlinkRightJoinToLeftJoinRuleTest.scala

@@ -0,0 +1,79 @@
+/*


Can you please write the test in .java? We're trying to move away from scala and implement new things in java

gustavodemorais · 2025-06-05T09:17:46Z

...ala/org/apache/flink/table/planner/plan/rules/logical/FlinkRightJoinToLeftJoinRuleTest.scala

+
+  @BeforeEach
+  def setup(): Unit = {
+    util.buildBatchProgram(FlinkBatchProgram.DEFAULT_REWRITE)


Could we use streaming for the setup?

gustavodemorais · 2025-06-05T09:18:00Z

...ala/org/apache/flink/table/planner/plan/rules/logical/FlinkRightJoinToLeftJoinRuleTest.scala

+
+  @Test
+  def testJoinChainIsConverted(): Unit = {
+    val sqlQuery = "SELECT * FROM (SELECT * FROM T1 JOIN T2 ON a = c) RIGHT JOIN T3 ON a = f"


gustavodemorais · 2025-06-05T09:18:52Z

...ala/org/apache/flink/table/planner/plan/rules/logical/FlinkRightJoinToLeftJoinRuleTest.scala

+    val sqlQuery = "SELECT b, d FROM T1 RIGHT JOIN T2 ON a = c"
+    util.verifyRelPlan(sqlQuery)
+  }
+}


Can you please add one test for multiple chained right joins? T1 RIGHT JOIN T2 RIGHT JOIN T3

gustavodemorais · 2025-06-05T09:25:44Z

...urces/org/apache/flink/table/planner/plan/rules/logical/FlinkRightJoinToLeftJoinRuleTest.xml

+		<Resource name="optimized rel plan">
+			<![CDATA[
+LogicalProject(b=[$1], d=[$3])
+- LogicalProject(a=[$3], b=[$4], c=[$0], d=[$1], e=[$2])


Why are indexes here for the same key (d=[$1] and d=[$3]) different?

I don't think we want to actually reorder the projected fields since the user expects the query to be returned as they wrote it? What we actually want is to make sure the indexes for the projected fields are correct

When we call transformTo on RelOptRuleCall calcite verifies that output record before transformation is the same as after applying. Without projection it will look like this:

Type mismatch:
rowtype of original rel: RecordType(INTEGER a, BIGINT b, INTEGER c, BIGINT d, BIGINT e) NOT NULL
rowtype of new rel: RecordType(INTEGER c, BIGINT d, BIGINT e, INTEGER a, BIGINT b) NOT NULL
Difference:
c: INTEGER -> BIGINT
d: BIGINT -> INTEGER

Since I swap left and right input output record field order is changed. So I had to add this projection to return field order back to normal.

RelNode project = reorderProjectedFields(left, right, relBuilder);

Say we have T1 with field a, T2 with field b and T3 with field c
(T1 join T2) right join T3 without multijoin will return abc
but with multijoin it will be converted to T3 left join (T1 join T2) ===> multijoin(T3, T1, T2) which will return cab.
So the projection is necessary. And I can handle it in JoinToMultiJoinRule.

I think I could create a child class from calcite's Join, something like ConvertedJoin, to override getRowType. But I decided to avoid it. Let me know what do you think!

That makes sense! Thanks for the reply :)

Let me know when you address the other points. If you have a draft PR for the other ticket, send it my way and I'll try to work this weekend. I have the PR for the exec node open and might have to slightly change the joinAttributeMap coming from the logical node #26647 (comment)

gustavodemorais · 2025-06-05T09:40:23Z

@gustavodemorais Hey! Please, take a look at this. Also I'm waiting for the ticket with join to multi join rule since i'm also almost done with it.

Here's the other ticket https://issues.apache.org/jira/browse/FLINK-37889

SteveStevenpoor · 2025-06-09T09:03:05Z

Hey, @gustavodemorais! I added some tests, made it java instead of scala and changed batch program to stream program.

gustavodemorais

Thanks for the change and the contribution, @SteveStevenpoor! Two last nits and apart from that it looks good to me 👍

...ain/java/org/apache/flink/table/planner/plan/rules/logical/FlinkRightJoinToLeftJoinRule.java

gustavodemorais · 2025-06-09T09:21:37Z

...cala/org/apache/flink/table/planner/plan/rules/logical/FlinkRightJoinToLeftJoinRuleTest.java

@@ -0,0 +1,139 @@
+/*


Now that this is .java, please move it to the java test path instead of scala. Smth like flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/FlinkRightJoinToLeftJoinRuleTest.java

Thank's, I've missed it! Now it's in java tests path.

gustavodemorais · 2025-06-09T12:17:08Z

The PR looks good to me. Let's get a committer to review it and we should be good to go

...ain/java/org/apache/flink/table/planner/plan/rules/logical/FlinkRightJoinToLeftJoinRule.java

flinkbot added the component=<none> label Jun 3, 2025

SteveStevenpoor mentioned this pull request Jun 4, 2025

[FLIP-516][FLINK-37481] Add new multi way join operator #26313

Merged

[FLINK-37814][table-planner] Add FlinkRightJoinToLeftJoinRule for the…

d74d786

… need of multi-way join operator

SteveStevenpoor force-pushed the FLINK-37814 branch from 09738f8 to d74d786 Compare June 4, 2025 10:15

davidradl reviewed Jun 4, 2025

View reviewed changes

[FLINK-37814][table-planner] Remove deprecated methods

7c8dc2c

gustavodemorais reviewed Jun 5, 2025

View reviewed changes

[FLINK-37814][table-planner] Move scala tests to java

a93d238

gustavodemorais reviewed Jun 9, 2025

View reviewed changes

[FLINK-37889][table-planner] Refactor tests

7182aa3

gustavodemorais approved these changes Jun 9, 2025

View reviewed changes

dawidwys approved these changes Jun 9, 2025

View reviewed changes

...ain/java/org/apache/flink/table/planner/plan/rules/logical/FlinkRightJoinToLeftJoinRule.java Show resolved Hide resolved

dawidwys merged commit 58d3cb4 into apache:master Jun 9, 2025

[FLINK-37814][table-planner] Add FlinkRightJoinToLeftJoinRule #26627

[FLINK-37814][table-planner] Add FlinkRightJoinToLeftJoinRule #26627

Uh oh!

Conversation

SteveStevenpoor commented Jun 3, 2025

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!

SteveStevenpoor commented Jun 3, 2025

Uh oh!

flinkbot commented Jun 3, 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!

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!

Choose a reason for hiding this comment

Uh oh!

gustavodemorais commented Jun 5, 2025

Uh oh!

SteveStevenpoor commented Jun 9, 2025

Uh oh!

gustavodemorais left a comment

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!

gustavodemorais commented Jun 9, 2025

Uh oh!

Uh oh!

Uh oh!

flinkbot commented Jun 3, 2025 •

edited

Loading