feat: derive monotonicity for date_truc with timezone when unit larger than day

src/expr/impl/src/scalar/date_trunc.rs

@@ -34,6 +34,7 @@ const MILLENNIUM: &str = "millennium";
 
 #[function("date_trunc(varchar, timestamp) -> timestamp")]
 pub fn date_trunc_timestamp(field: &str, ts: Timestamp) -> Result<Timestamp> {
+    // NOTE(st1page): please also modify the `MonotonicityAnalyzer::date_unit_larger_than_day` in frontend if this changes
     Ok(match field.to_ascii_lowercase().as_str() {
         MICROSECONDS => ts.truncate_micros(),
         MILLISECONDS => ts.truncate_millis(),

src/frontend/planner_test/tests/testdata/input/temporal_filter.yaml

@@ -123,6 +123,12 @@
     select * from t where ts + interval '1 hour' > date_trunc('day', now());
   expected_outputs:
     - stream_plan
+- name: date_trunc_with_timezone
+  sql: |
+    create table t (ts timestamp with time zone, a int);
+    select * from t where ts + interval '1 hour' > date_trunc('day', now(), 'Asia/Singapore');
+  expected_outputs:
+    - stream_plan
 - name: Non-trivial now expression 2
   sql: |
     create table t (ts timestamp with time zone, a int);

src/frontend/planner_test/tests/testdata/output/temporal_filter.yaml

@@ -476,6 +476,19 @@
         └─StreamExchange { dist: Broadcast }
           └─StreamProject { exprs: [DateTrunc('day':Varchar, now, 'UTC':Varchar) as $expr2], output_watermarks: [$expr2] }
             └─StreamNow { output: [now] }
+- name: date_trunc_with_timezone
+  sql: |
+    create table t (ts timestamp with time zone, a int);
+    select * from t where ts + interval '1 hour' > date_trunc('day', now(), 'Asia/Singapore');
+  stream_plan: |-
+    StreamMaterialize { columns: [ts, a, t._row_id(hidden)], stream_key: [t._row_id], pk_columns: [t._row_id], pk_conflict: NoCheck }
+    └─StreamProject { exprs: [t.ts, t.a, t._row_id] }
+      └─StreamDynamicFilter { predicate: ($expr1 > $expr2), output_watermarks: [$expr1], output: [t.ts, t.a, $expr1, t._row_id], cleaned_by_watermark: true }
+        ├─StreamProject { exprs: [t.ts, t.a, AddWithTimeZone(t.ts, '01:00:00':Interval, 'UTC':Varchar) as $expr1, t._row_id] }
+        │ └─StreamTableScan { table: t, columns: [t.ts, t.a, t._row_id], stream_scan_type: ArrangementBackfill, stream_key: [t._row_id], pk: [_row_id], dist: UpstreamHashShard(t._row_id) }
+        └─StreamExchange { dist: Broadcast }
+          └─StreamProject { exprs: [DateTrunc('day':Varchar, now, 'Asia/Singapore':Varchar) as $expr2], output_watermarks: [$expr2] }
+            └─StreamNow { output: [now] }
 - name: Non-trivial now expression 2
   sql: |
     create table t (ts timestamp with time zone, a int);

src/frontend/src/optimizer/property/monotonicity.rs

@@ -186,6 +186,23 @@ impl MonotonicityAnalyzer {
             tz_is_utc // conservative
         }
 
+        fn date_unit_larger_than_day(date_unit: Option<&str>) -> bool {
+            const DAY: &str = "day";
+            const WEEK: &str = "week";
+            const MONTH: &str = "month";
+            const QUARTER: &str = "quarter";
+            const YEAR: &str = "year";
+            const DECADE: &str = "decade";
+            const CENTURY: &str = "century";
+            const MILLENNIUM: &str = "millennium";
+            date_unit.map_or(false, |date_unit| {
+                matches!(
+                    date_unit.to_ascii_lowercase().as_str(),
+                    DAY | WEEK | MONTH | QUARTER | YEAR | DECADE | CENTURY | MILLENNIUM
+                )
+            })
+        }
+
         match func_call.func_type() {
             ExprType::Unspecified => unreachable!(),
             ExprType::Add => match self.visit_binary_op(func_call.inputs()) {
@@ -257,7 +274,13 @@ impl MonotonicityAnalyzer {
                             .as_literal()
                             .and_then(|literal| literal.get_data().as_ref())
                             .map(|tz| tz.as_utf8().as_ref());
-                        if time_zone_is_without_dst(time_zone) {
+                        let date_unit = func_call.inputs()[0]
+                            .as_literal()
+                            .and_then(|literal| literal.get_data().as_ref())
+                            .map(|tz| tz.as_utf8().as_ref());
+                        if time_zone_is_without_dst(time_zone)
+                            || date_unit_larger_than_day(date_unit)
+                        {
                             any
                         } else {
                             Inherent(Unknown)