You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "SteveStevenpoor (via GitHub)" <gi...@apache.org> on 2026/03/25 10:43:03 UTC
[PR] [FLINK-39324][table-planner] Allow MultiJoin to respect IS_NOT_DISTICT_FROM when extracting join keys [flink]
SteveStevenpoor opened a new pull request, #27827:
URL: https://github.com/apache/flink/pull/27827
## What is the purpose of the change
### Example
Consider the following query:
```sql
INSERT INTO ResultSink
SELECT u.user_id, u.name, o.order_id, p.payment_id, p.price
FROM Users u
JOIN Orders o
ON u.user_id IS NOT DISTINCT FROM o.user_id
JOIN Payments p
ON o.user_id IS NOT DISTINCT FROM p.user_id
```
### Before fix:
MultiJoin fails to extract a common join key, resulting in two MultiJoin nodes and singleton distribution:
```bash
== Optimized Execution Plan ==
Sink(table=[default_catalog.default_database.ResultSink], fields=[user_id, name, order_id, payment_id, price])
+- Calc(select=[user_id, name, order_id, payment_id, price])
+- MultiJoin(commonJoinKey=[noCommonJoinKey], joinTypes=[INNER], inputUniqueKeys=[noUniqueKey, noUniqueKey], joinConditions=[IS NOT DISTINCT FROM(user_id0, user_id1)], select=[user_id,name,order_id,user_id0,payment_id,price,user_id1], rowType=[RecordType(VARCHAR(2147483647) user_id, VARCHAR(2147483647) name, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id0, VARCHAR(2147483647) payment_id, INTEGER price, VARCHAR(2147483647) user_id1)])
:- Exchange(distribution=[single])
: +- MultiJoin(commonJoinKey=[noCommonJoinKey], joinTypes=[INNER], inputUniqueKeys=[noUniqueKey, noUniqueKey], joinConditions=[IS NOT DISTINCT FROM(user_id, user_id0)], select=[user_id,name,order_id,user_id0], rowType=[RecordType(VARCHAR(2147483647) user_id, VARCHAR(2147483647) name, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id0)])
: :- Exchange(distribution=[single])
: : +- Calc(select=[user_id, name])
: : +- MiniBatchAssigner(interval=[1000ms], mode=[ProcTime])
: : +- TableSourceScan(table=[[default_catalog, default_database, Users]], fields=[user_id, name, cash])
: +- Exchange(distribution=[single])
: +- Calc(select=[order_id, user_id])
: +- MiniBatchAssigner(interval=[1000ms], mode=[ProcTime])
: +- TableSourceScan(table=[[default_catalog, default_database, Orders]], fields=[order_id, user_id, product])
+- Exchange(distribution=[single])
+- MiniBatchAssigner(interval=[1000ms], mode=[ProcTime])
+- TableSourceScan(table=[[default_catalog, default_database, Payments]], fields=[payment_id, price, user_id])
```
### After fix:
Joins are flattened into a single MultiJoin node:
```bash
== Optimized Execution Plan ==
Sink(table=[default_catalog.default_database.ResultSink], fields=[user_id, name, order_id, payment_id, price])
+- Calc(select=[user_id, name, order_id, payment_id, price])
+- MultiJoin(commonJoinKey=[user_id], joinTypes=[INNER, INNER], inputUniqueKeys=[noUniqueKey, noUniqueKey, noUniqueKey], joinConditions=[IS NOT DISTINCT FROM(user_id, user_id0), IS NOT DISTINCT FROM(user_id0, user_id1)], select=[user_id,name,order_id,user_id0,payment_id,price,user_id1], rowType=[RecordType(VARCHAR(2147483647) user_id, VARCHAR(2147483647) name, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id0, VARCHAR(2147483647) payment_id, INTEGER price, VARCHAR(2147483647) user_id1)])
:- Exchange(distribution=[hash[user_id]])
: +- Calc(select=[user_id, name])
: +- MiniBatchAssigner(interval=[1000ms], mode=[ProcTime])
: +- TableSourceScan(table=[[default_catalog, default_database, Users]], fields=[user_id, name, cash])
:- Exchange(distribution=[hash[user_id]])
: +- Calc(select=[order_id, user_id])
: +- MiniBatchAssigner(interval=[1000ms], mode=[ProcTime])
: +- TableSourceScan(table=[[default_catalog, default_database, Orders]], fields=[order_id, user_id, product])
+- Exchange(distribution=[hash[user_id]])
+- MiniBatchAssigner(interval=[1000ms], mode=[ProcTime])
+- TableSourceScan(table=[[default_catalog, default_database, Payments]], fields=[payment_id, price, user_id])
```
## Brief change log
- *Little addition to join attribute map creation*
- *Added some tests*
## Verifying this change
This change added tests and can be verified as follows:
- *Added test with IS_NOT_DISTINCT_FROM to MultiJoinTest*
- *Added test with IS_NOT_DISTINCT_FROM to MultiJoinSemanticTests*
Also some tests with IS_NOT_DISTINCT_FROM will be available after [FLINK-38551](https://github.com/apache/flink/pull/27415) is merged.
## 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39324][table-planner] Allow MultiJoin to respect IS_NOT_DISTICT_FROM when extracting join keys [flink]
Posted by "Au-Miner (via GitHub)" <gi...@apache.org>.
Au-Miner commented on code in PR #27827:
URL: https://github.com/apache/flink/pull/27827#discussion_r3115533055
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinTestPrograms.java:
##########
@@ -133,6 +133,56 @@ public class MultiJoinTestPrograms {
+ "INNER JOIN Payments p ON u.user_id = p.user_id")
.build();
+ public static final TableTestProgram MULTI_JOIN_THREE_WAY_INNER_JOIN_WITH_IS_NOT_DISTINCT =
Review Comment:
The new semantic test lacks a state recovery scenario, and testing can be added in `MultiJoinRestoreTest`
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinTestPrograms.java:
##########
@@ -133,6 +133,56 @@ public class MultiJoinTestPrograms {
+ "INNER JOIN Payments p ON u.user_id = p.user_id")
.build();
+ public static final TableTestProgram MULTI_JOIN_THREE_WAY_INNER_JOIN_WITH_IS_NOT_DISTINCT =
+ TableTestProgram.of(
+ "three-way-inner-join-with-is-not-distinct",
+ "three way inner join with is not distinct")
+ .setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
Review Comment:
Aligning with other MultiJoinSemanticTests#programs, `.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)` can be deleted
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39324][table-planner] Allow MultiJoin to respect IS_NOT_DISTICT_FROM when extracting join keys [flink]
Posted by "SteveStevenpoor (via GitHub)" <gi...@apache.org>.
SteveStevenpoor commented on PR #27827:
URL: https://github.com/apache/flink/pull/27827#issuecomment-4497501363
Hey, @snuyanzin! Could you check this pls?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39324][table-planner] Allow MultiJoin to respect IS_NOT_DISTICT_FROM when extracting join keys [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #27827:
URL: https://github.com/apache/flink/pull/27827#issuecomment-4125543634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3258bfde922655b5278347dcd1b916f93f931ca4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3258bfde922655b5278347dcd1b916f93f931ca4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3258bfde922655b5278347dcd1b916f93f931ca4 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39324][table-planner] Allow MultiJoin to respect IS_NOT_DISTICT_FROM when extracting join keys [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #27827:
URL: https://github.com/apache/flink/pull/27827#issuecomment-4497543267
let's also check with @gustavodemorais who contributed a lot to multijoin
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org