ProjectWatermarkAssignerTransposeRule field pruning causes wrong watermark index to be accessed

[Yuval Itzchakov]

Hi,

Flink 1.12.1

Scala 2.12.12

While attempting to fix a serialization bug I previously wrote about, I temporarily disabled projection pushdown for my custom source implementation. I then proceeded to run the application only to encounter a ClassCastException, which after debugging was caused by the fact the WatermarkGenerator auto-generated code was trying to access the rowtime under a wrong index (0 instead of 1).

My program is as follows:

TableSource -> SQL query over table -> Sink

Debugging the issue, I encountered ProjectWatermarkAssignerTransposeRule which resolves unused fields in the table, prunes them and re-orders the fields with new indexes (https://github.com/apache/flink/blob/master/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/rules/logical/ProjectWatermarkAssignerTransposeRule.java)

Now, since I did not use projection pushdown (which would have probably fixed the issue and caused the re-ordering of the field indexes), the wrong idx was accessed.

You can see the reordering in the plan:

Initially:

         +- LogicalProject(event_time=[$1])
            +- LogicalWatermarkAssigner(rowtime=[event_time], watermark=[$1])
               +- LogicalTableScan(table=[[foo-catalog, default-db, foo-table]])

After optimization:

         +- FlinkLogicalWatermarkAssigner(rowtime=[event_time], watermark=[$0])
            +- FlinkLogicalTableSourceScan(table=[[foo-catalog, default-db, foo-table, project=[event_time])

Is this the desired behavior? I was quite surprised by this and the fact that the pruning happens regardless of if the underlying table is a TableSource or if it actually filters out these unused fields from the result query.

--