watermark_strategy_expression

Author: st1page

rfcs/0002-watermark-filter.md

@@ -235,6 +235,69 @@ WINDOW w AS (
 )
 ```
 
+### watermark_strategy_expression
+
+> the section is updated on 31 Oct
+
+the old synatx is
+
+```plain
+WATERMARK(orders, time_column, timeout: interval)
+```
+
+and we will change it to
+
+```plain
+WATERMARK(orders, time_column, watermark_strategy_expression)
+```
+
+And the concept `watermark_strategy_expression` refers [FlinkSQL](https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/create/#watermark). The expression is evaluated for every record and update the watermark if the result greater than the current watermark.
+The new synatx is more friendly for flinkSQL user and offer potential to make user define their strategy
+
+```sql
+CREATE SOURCE `orders` (
+  `id` BIGINT,
+  `order_time` TIMESTAMP,
+  `price` DECIMAL,
+  `customer_id` BIGINT,
+) WITH (
+  'connector' = ...,
+);
+
+-- normal timeout watermark
+with watermarked_orders as (
+  WATERMARK(orders, order_time, order - INTERVAL '1' MINUTE), 
+) 
+
+-- normal timeout watermark with simple check
+with watermarked_orders as (
+  WATERMARK(
+  orders, 
+  order_time, 
+  CASE
+  -- we have not determined the design about `PROC_TIME()`
+    WHEN order_time > PROC_TIME() THEN Null
+    ELSE order - INTERVAL '1' MINUTE
+  END), 
+
+  -- normal timeout watermark with removing outliers
+with watermarked_orders as (
+  WATERMARK(
+  (
+    -- we have not determined the window funtion on unordered stream
+    select *, 
+      max(order_time) as win_max_time 
+        OVER(BETWEEN 10 PRECEDING AND CURRENT ROW)
+    from orders
+  ), 
+  order_time, 
+  CASE
+    WHEN order_time == win_max_time THEN Null
+    ELSE order - INTERVAL '1' MINUTE
+  END), 
+) 
+```
+
 ## Future possibilities
 
 ### Should we support extra columns in Sort?