[FLINK-29138] Project pushdown not work for lookup source

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.16.0, 1.14.6, 1.15.3
• Component/s: Table SQL / Planner
• Labels: pull-request-available
• Assignee: lincoln lee
• Reporter: lincoln lee
• Created: 30/Aug/22 09:02
• Updated: 04/Jul/23 07:41
• Resolved: 05/Sep/22 02:37

Description

Current tests: LookupJoinTest#testJoinTemporalTableWithProjectionPushDown

@Test
def testJoinTemporalTableWithProjectionPushDown(): Unit = {
val sql =
"""
|SELECT T.*, D.id
|FROM MyTable AS T
|JOIN LookupTable FOR SYSTEM_TIME AS OF T.proctime AS D
|ON T.a = D.id
""".stripMargin

util.verifyExecPlan(sql)
}

the optimized plan doesn't print the selected columns from lookup source, but actually it didn't push the project into lookup source (still select all columns from source), this is not as expected

<Resource name="optimized exec plan">
<![CDATA[
Calc(select=[a, b, c, PROCTIME_MATERIALIZE(proctime) AS proctime, rowtime, id])
+- LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[InnerJoin], lookup=[id=a], select=[a, b, c, proctime, rowtime, id])
+- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime])
]]>
</Resource>

 

incorrect intermediate optimization result

=========  logical_rewrite ========
 optimize result: 
FlinkLogicalJoin(condition=[=($0, $5)], joinType=[inner])
:- FlinkLogicalDataStreamTableScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime])
+- FlinkLogicalSnapshot(period=[$cor0.proctime])
   +- FlinkLogicalCalc(select=[id])
      +- FlinkLogicalTableSourceScan(table=[[default_catalog, default_database, LookupTable]], fields=[id, name, age])


=========  time_indicator ========
 optimize result: 
FlinkLogicalCalc(select=[a, b, c, PROCTIME_MATERIALIZE(proctime) AS proctime, rowtime, id])
+- FlinkLogicalJoin(condition=[=($0, $5)], joinType=[inner])
   :- FlinkLogicalDataStreamTableScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime])
   +- FlinkLogicalSnapshot(period=[$cor0.proctime])
      +- FlinkLogicalCalc(select=[id])
         +- FlinkLogicalTableSourceScan(table=[[default_catalog, default_database, LookupTable]], fields=[id, name, age])

 

plan comparison after fix

Issue Links

• relates to: FLINK-20840 Projection pushdown doesn't work in temporal(lookup) join
• links to: GitHub Pull Request #20717
• links to: GitHub Pull Request #20727
• links to: GitHub Pull Request #20729