[FLINK-25396] lookupjoin source table for pre-partitioning

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Duplicate
• Component/s: Table SQL / Planner
• Assignee: Unassigned
• Reporter: HunterXHunter
• Created: 21/Dec/21 02:29
• Updated: 21/Dec/21 02:50
• Resolved: 21/Dec/21 02:48

Description

When we perform external associations, we need to partition by key first, so that the same key is in a task, which can reduce the number of queries and make the external data cached by each task more scattered rather than full

Example：select * from sourceTable t1 LEFT JOIN lookuptable FOR SYSTEM_TIME AS OF t1.proctime as t2 ON t1.msg = t2.word

Execution Plan like：

== Optimized Execution Plan ==
Calc(select=[topic, offset, rowtime, msg, uid, PROCTIME_MATERIALIZE(proctime) AS proctime, word])
+- LookupJoin(table=[default_catalog.default_database.hbaselookup], joinType=[LeftOuterJoin], async=[false], lookup=[word=msg], select=[topic, offset, rowtime, msg, uid, proctime, word])
   +- Calc(select=[CAST(topic) AS topic, CAST(offset) AS offset, Reinterpret(rowtime) AS rowtime, msg, uid, PROCTIME() AS proctime])
      +- TableSourceScan(table=[[default_catalog, default_database, test, watermark=[-($0, 10000:INTERVAL SECOND)]]], fields=[rowtime, msg, uid, topic, offset])

After I made the optimization, I added a hint configuration(lookup.join.pre-partition) and added a rule to generate an exchange. so that I can pre-partition by the join key when obtaining external data synchronously

select * from test t1 LEFT JOIN hbaselookup /*+ OPTIONS('lookup.join.pre-partition'='true') */ FOR SYSTEM_TIME AS OF t1.proctime as t2 ON t1.msg = t2.word

== Optimized Execution Plan ==
Calc(select=[topic, offset, rowtime, msg, uid, PROCTIME_MATERIALIZE(proctime) AS proctime, word])
+- LookupJoin(table=[default_catalog.default_database.hbaselookup], joinType=[LeftOuterJoin], async=[false], lookup=[word=msg], select=[topic, offset, rowtime, msg, uid, proctime, word])
   +- Exchange(distribution=[hash[msg]])
      +- Calc(select=[CAST(topic) AS topic, CAST(offset) AS offset, Reinterpret(rowtime) AS rowtime, msg, uid, PROCTIME() AS proctime])
         +- TableSourceScan(table=[[default_catalog, default_database, test, watermark=[-($0, 10000:INTERVAL SECOND)]]], fields=[rowtime, msg, uid, topic, offset])
 

 

Issue Links

• duplicates: FLINK-23687 Introduce partitioned lookup join to enforce input of LookupJoin to hash shuffle by lookup keys