feat(frontend): cast/cmp with session timezone

e2e_test/batch/functions/session_timezone.slt

@@ -0,0 +1,59 @@
+statement ok
+set timezone = "us/pacific";
+
+# Cast date to timestamptz
+query T
+select '2022-01-01'::date::timestamp with time zone;
+----
+2022-01-01 08:00:00+00:00
+
+# Cast timestamp to timestamptz
+query T
+select '2022-01-01 00:00:00'::timestamp::timestamp with time zone;
+----
+2022-01-01 08:00:00+00:00
+
+# Cast timestamptz to timestamp
+query T
+select '2022-01-01 00:00:00+00:00'::timestamp with time zone::timestamp;
+----
+2021-12-31 16:00:00
+
+# Cast timestamptz to date
+query T
+select '2022-01-01 00:00:00+00:00'::timestamp with time zone::date;
+----
+2021-12-31
+
+# Cast timestamptz to time
+query T
+select '2022-01-01 00:00:00+00:00'::timestamp with time zone::time;
+----
+16:00:00
+
+# Compare timestamp with timestamptz
+query T
+select '2022-01-01 00:00:00-08:00'::timestamp with time zone = '2022-01-01 00:00:00'::timestamp;
+----
+t
+
+# Compare date with timestamptz
+query T
+select '2022-01-01 00:00:00-08:00'::timestamp with time zone = '2022-01-01'::date;
+----
+t
+
+# Cast varchar without timezone to timestamptz
+query T
+select '2022-01-01 00:00:00'::timestamp with time zone;
+----
+2022-01-01 08:00:00+00:00
+
+# Cast timestamptz to varchar, should display with timezone information
+query T
+select '2022-01-01 00:00:00-08:00'::timestamp with time zone::varchar;
+----
+2022-01-01 00:00:00-08:00
+
+statement ok
+set timezone = 'UTC';

e2e_test/streaming/dynamic_filter.slt

@@ -98,3 +98,55 @@ drop table t1;
 
 statement ok
 drop table t2;
+
+# Simple Agg with timestamp/timestamptz
+statement ok
+create table t1 (v1 timestamp);
+
+statement ok
+create table t2 (v2 timestamp with time zone);
+
+statement ok
+create materialized view mv1 as with max_v2 as (select max(v2) max from t2) select v1 from t1, max_v2 where v1 > max;
+
+statement ok
+insert into t1 values ('2001-01-01 19:00:00'), ('2001-01-01 20:00:00'), ('2001-01-01 21:00:00');
+
+# RHS is NULL means predicate is not true
+query I
+select * from mv1 order by v1;
+----
+
+statement ok
+insert into t2 values ('2001-01-01 18:00:00+00:00');
+
+query I
+select * from mv1 order by v1;
+----
+2001-01-01 19:00:00
+2001-01-01 20:00:00
+2001-01-01 21:00:00
+
+# query I
+# select * from mv2 order by v1;
+# ----
+# 1
+# 2
+# 3
+
+statement ok
+insert into t2 values ('2001-01-01 20:00:00+00:00');
+
+query I
+select * from mv1 order by v1;
+----
+2001-01-01 21:00:00
+
+statement ok
+drop materialized view mv1;
+
+statement ok
+drop table t1;
+
+statement ok
+drop table t2;

proto/expr.proto

@@ -53,6 +53,8 @@ message ExprNode {
     // Parse text to timestamp by format string.
     // e.g. `select to_timestamp('2022 08 21', 'YYYY MM DD')`
     TO_TIMESTAMP1 = 107;
+    // Performs a cast with additional timezone information.
+    CAST_WITH_TIME_ZONE = 108;
     // other functions
     CAST = 201;
     SUBSTR = 202;

src/expr/src/expr/build_expr_from_prost.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use risingwave_common::try_match_expand;
 use risingwave_common::types::DataType;
 use risingwave_common::util::value_encoding::deserialize_datum;
 use risingwave_pb::expr::expr_node::{RexNode, Type};
@@ -38,7 +39,9 @@ use crate::expr::expr_to_timestamp_const_tmpl::{
 use crate::expr::expr_unary::{
     new_length_default, new_ltrim_expr, new_rtrim_expr, new_trim_expr, new_unary_expr,
 };
-use crate::expr::{build_from_prost as expr_build_from_prost, BoxedExpression, Expression};
+use crate::expr::{
+    build_from_prost as expr_build_from_prost, BoxedExpression, Expression, LiteralExpression,
+};
 use crate::vector_op::to_char::compile_pattern_to_chrono;
 use crate::{bail, ensure, Result};
 
@@ -252,6 +255,17 @@ pub fn build_to_char_expr(prost: &ExprNode) -> Result<BoxedExpression> {
     }
 }
 
+pub fn build_now_expr(prost: &ExprNode) -> Result<BoxedExpression> {
+    let rex_node = try_match_expand!(prost.get_rex_node(), Ok)?;
+    let RexNode::FuncCall(func_call_node) = rex_node else {
+        bail!("Expected RexNode::FuncCall in Now");
+    };
+    let Some(bind_timestamp) = func_call_node.children.first() else {
+        bail!("Expected epoch timestamp bound into Now");
+    };
+    LiteralExpression::try_from(bind_timestamp).map(Expression::boxed)
+}
+
 pub fn build_to_timestamp_expr(prost: &ExprNode) -> Result<BoxedExpression> {
     let (children, ret_type) = get_children_and_return_type(prost)?;
     ensure!(children.len() == 2);

src/expr/src/expr/expr_binary_nonnull.rs

@@ -21,7 +21,7 @@ use risingwave_pb::expr::expr_node::Type;
 
 use super::Expression;
 use crate::expr::expr_binary_bytes::new_concat_op;
-use crate::expr::template::BinaryExpression;
+use crate::expr::template::{BinaryBytesExpression, BinaryExpression};
 use crate::expr::{template_fast, BoxedExpression};
 use crate::vector_op::arithmetic_op::*;
 use crate::vector_op::bitwise_op::*;
@@ -33,7 +33,9 @@ use crate::vector_op::extract::{
 use crate::vector_op::like::like_default;
 use crate::vector_op::position::position;
 use crate::vector_op::round::round_digits;
-use crate::vector_op::timestamptz::{timestamp_at_time_zone, timestamptz_at_time_zone};
+use crate::vector_op::timestamptz::{
+    str_to_timestamptz, timestamp_at_time_zone, timestamptz_at_time_zone, timestamptz_to_string,
+};
 use crate::vector_op::to_timestamp::to_timestamp;
 use crate::vector_op::tumble::{
     tumble_start_date, tumble_start_date_time, tumble_start_timestamptz,
@@ -415,6 +417,38 @@ fn build_at_time_zone_expr(
     Ok(expr)
 }
 
+fn build_cast_with_time_zone_expr(
+    ret: DataType,
+    l: BoxedExpression,
+    r: BoxedExpression,
+) -> Result<BoxedExpression> {
+    let expr: BoxedExpression = match (ret.clone(), l.return_type()) {
+        (DataType::Varchar, DataType::Timestamptz) => Box::new(BinaryBytesExpression::<
+            I64Array,
+            Utf8Array,
+            _,
+        >::new(
+            l, r, ret, timestamptz_to_string
+        )),
+        (DataType::Timestamptz, DataType::Varchar) => {
+            Box::new(BinaryExpression::<Utf8Array, Utf8Array, I64Array, _>::new(
+                l,
+                r,
+                ret,
+                str_to_timestamptz,
+            ))
+        }
+        _ => {
+            return Err(ExprError::UnsupportedFunction(format!(
+                "cannot cast at time zone (input type: {:?}, output type: {:?}",
+                l.return_type(),
+                ret,
+            )))
+        }
+    };
+    Ok(expr)
+}
+
 pub fn new_date_trunc_expr(
     ret: DataType,
     field: BoxedExpression,
@@ -630,6 +664,7 @@ pub fn new_binary_expr(
         }
         Type::Extract => build_extract_expr(ret, l, r)?,
         Type::AtTimeZone => build_at_time_zone_expr(ret, l, r)?,
+        Type::CastWithTimeZone => build_cast_with_time_zone_expr(ret, l, r)?,
         Type::RoundDigit => Box::new(template_fast::BinaryExpression::new(
             l,
             r,

src/expr/src/expr/mod.rs

@@ -49,8 +49,6 @@ pub use expr_literal::*;
 use risingwave_common::array::{ArrayRef, DataChunk};
 use risingwave_common::row::OwnedRow;
 use risingwave_common::types::{DataType, Datum};
-use risingwave_common::{bail, try_match_expand};
-use risingwave_pb::expr::expr_node::RexNode;
 use risingwave_pb::expr::ExprNode;
 
 use super::Result;
@@ -114,7 +112,7 @@ pub fn build_from_prost(prost: &ExprNode) -> Result<BoxedExpression> {
         Equal | NotEqual | LessThan | LessThanOrEqual | GreaterThan | GreaterThanOrEqual | Add
         | Subtract | Multiply | Divide | Modulus | Extract | RoundDigit | TumbleStart
         | Position | BitwiseShiftLeft | BitwiseShiftRight | BitwiseAnd | BitwiseOr | BitwiseXor
-        | ConcatOp | AtTimeZone => build_binary_expr_prost(prost),
+        | ConcatOp | AtTimeZone | CastWithTimeZone => build_binary_expr_prost(prost),
         And | Or | IsDistinctFrom | IsNotDistinctFrom | ArrayAccess => {
             build_nullable_binary_expr_prost(prost)
         }
@@ -154,16 +152,7 @@ pub fn build_from_prost(prost: &ExprNode) -> Result<BoxedExpression> {
             ArrayConcatExpression::try_from(prost).map(Expression::boxed)
         }
         Vnode => VnodeExpression::try_from(prost).map(Expression::boxed),
-        Now => {
-            let rex_node = try_match_expand!(prost.get_rex_node(), Ok)?;
-            let RexNode::FuncCall(func_call_node) = rex_node else {
-                bail!("Expected RexNode::FuncCall in Now");
-            };
-            let Option::Some(bind_timestamp) = func_call_node.children.first() else {
-                bail!("Expected epoch timestamp bound into Now");
-            };
-            LiteralExpression::try_from(bind_timestamp).map(Expression::boxed)
-        }
+        Now => build_now_expr(prost),
         Udf => UdfExpression::try_from(prost).map(Expression::boxed),
         _ => Err(ExprError::UnsupportedFunction(format!(
             "{:?}",

src/expr/src/vector_op/cast.rs

@@ -249,12 +249,6 @@ pub fn parse_bytes_traditional(s: &str) -> Result<Vec<u8>> {
     Ok(out)
 }
 
-pub fn timestamptz_to_utc_string(elem: i64, mut writer: &mut dyn Write) -> Result<()> {
-    elem.write_with_type(&DataType::Timestamptz, &mut writer)
-        .unwrap();
-    Ok(())
-}
-
 #[inline(always)]
 pub fn str_parse<T>(elem: &str) -> Result<T>
 where
@@ -342,14 +336,6 @@ pub fn interval_to_time(elem: IntervalUnit) -> NaiveTimeWrapper {
     NaiveTimeWrapper::from_num_seconds_from_midnight_uncheck(secs, nano)
 }
 
-#[inline(always)]
-pub fn timestamptz_to_timestamp(elem: i64) -> NaiveDateTimeWrapper {
-    NaiveDateTimeWrapper::from_timestamp_uncheck(
-        elem / 1_000_000,
-        (elem % 1_000_000 * 1_000) as u32,
-    )
-}
-
 #[inline(always)]
 pub fn try_cast<T1, T2>(elem: T1) -> Result<T2>
 where
@@ -426,7 +412,6 @@ macro_rules! for_all_cast_variants {
             { varchar, time, str_to_time, false },
             { varchar, interval, str_parse, false },
             { varchar, timestamp, str_to_timestamp, false },
-            { varchar, timestamptz, str_to_timestamptz, false },
             { varchar, int16, str_parse, false },
             { varchar, int32, str_parse, false },
             { varchar, int64, str_parse, false },
@@ -448,7 +433,6 @@ macro_rules! for_all_cast_variants {
             { interval, varchar, general_to_text, false },
             { date, varchar, general_to_text, false },
             { timestamp, varchar, general_to_text, false },
-            { timestamptz, varchar, timestamptz_to_utc_string, false },
             { list, varchar, |x, w| general_to_text(x, w), false },
 
             { boolean, int32, try_cast, false },
@@ -491,8 +475,7 @@ macro_rules! for_all_cast_variants {
             { time, interval, cast::<NaiveTimeWrapper, IntervalUnit>, true },
             { timestamp, date, timestamp_to_date, true },
             { timestamp, time, timestamp_to_time, true },
-            { interval, time, interval_to_time, true },
-            { timestamptz, timestamp, timestamptz_to_timestamp, true }
+            { interval, time, interval_to_time, true }
         }
     };
 }
@@ -1009,23 +992,4 @@ mod tests {
         let timestamp2 = str_to_timestamp(str2).unwrap();
         assert_eq!(timestamp2.0.timestamp_micros(), -1);
     }
-
-    #[test]
-    fn test_timestamptz() {
-        let str1 = "0001-11-15 15:35:40.999999+08:00";
-        let timestamptz1 = str_to_timestamptz(str1).unwrap();
-        assert_eq!(timestamptz1, -62108094259000001);
-
-        let mut writer = String::new();
-        timestamptz_to_utc_string(timestamptz1, &mut writer).unwrap();
-        assert_eq!(writer, "0001-11-15 07:35:40.999999+00:00");
-
-        let str2 = "1969-12-31 23:59:59.999999+00:00";
-        let timestamptz2 = str_to_timestamptz(str2).unwrap();
-        assert_eq!(timestamptz2, -1);
-
-        let mut writer = String::new();
-        timestamptz_to_utc_string(timestamptz2, &mut writer).unwrap();
-        assert_eq!(writer, str2);
-    }
 }