YuvalItzchakov commented on pull request #15307:
URL: https://github.com/apache/flink/pull/15307#issuecomment-812499504
@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])
```
--
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
