Project pushdown not work for lookup source

@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)
}

<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>

=========  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])