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,48 @@ 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
+
+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!(
             "{:?}",