fsk119 commented on pull request #15307: URL: https://github.com/apache/flink/pull/15307#issuecomment-812502909
> @fsk119 After the change, my partial predicate match no longer works: > > ```java > @Test > public void testPushdownAcrossWatermarkPartialPredicateMatch() { > String ddl3 = "CREATE TABLE WithWatermark (" > + " name STRING,\n" > + " event_time TIMESTAMP(3),\n" > + " WATERMARK FOR event_time as event_time - INTERVAL '5' SECOND" > + ") WITH (\n" > + " 'connector' = 'values',\n" > + " 'bounded' = 'true',\n" > + " 'filterable-fields' = 'name',\n" > + " 'enable-watermark-push-down' = 'false',\n" > + " 'disable-lookup' = 'true'" > + ")"; > > util.tableEnv().executeSql(ddl3); > util.verifyRelPlan( > "SELECT * FROM WithWatermark WHERE LOWER(name) = 'foo' AND name IS NOT NULL"); > } > ``` > > Expected: > > ``` > Calc(select=[name, event_time], where=[IS NOT NULL(name)]) > +- WatermarkAssigner(rowtime=[event_time], watermark=[-(event_time, 5000:INTERVAL SECOND)]) > +- TableSourceScan(table=[[default_catalog, default_database, WithWatermark, filter=[equals(lower(name), 'foo')]]], fields=[name, event_time]) > ``` > > Actual: > > ``` > FlinkLogicalCalc(select=[name, event_time]) > +- FlinkLogicalCalc(select=[name, event_time], where=[AND(=(LOWER(name), _UTF-16LE'foo':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"), IS NOT NULL(name))]) > +- FlinkLogicalWatermarkAssigner(rowtime=[event_time], watermark=[-($1, 5000:INTERVAL SECOND)]) > +- FlinkLogicalTableSourceScan(table=[[default_catalog, default_database, WithWatermark]], fields=[name, event_time]) > ``` > > Partial match is not being pushed down. We shouldn't push down the filter if watermark assigner exists. If we push down the filter without watermark assigner push down, the scan will only emit the records meet the condition. -- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org