feat(expr): add rw_epoch_to_ts (#19473)

Signed-off-by: xxchan <[email protected]>

proto/expr.proto

@@ -325,6 +325,7 @@ message ExprNode {
     HAS_SCHEMA_PRIVILEGE = 2409;
     PG_IS_IN_RECOVERY = 2411;
     RW_RECOVERY_STATUS = 2412;
+    RW_EPOCH_TO_TS = 2413;
 
     // EXTERNAL
     ICEBERG_TRANSFORM = 2201;

src/frontend/src/binder/expr/function/builtin_scalar.rs

@@ -661,6 +661,7 @@ impl Binder {
                 ("shobj_description", raw_literal(ExprImpl::literal_varchar("".to_string()))),
                 ("pg_is_in_recovery", raw_call(ExprType::PgIsInRecovery)),
                 ("rw_recovery_status", raw_call(ExprType::RwRecoveryStatus)),
+                ("rw_epoch_to_ts", raw_call(ExprType::RwEpochToTs)),
                 // internal
                 ("rw_vnode", raw_call(ExprType::VnodeUser)),
                 ("rw_test_paid_tier", raw_call(ExprType::TestPaidTier)), // for testing purposes

src/frontend/src/expr/function_impl/mod.rs

@@ -22,4 +22,5 @@ mod pg_get_viewdef;
 mod pg_index_column_has_property;
 mod pg_indexes_size;
 mod pg_relation_size;
+mod rw_epoch_to_ts;
 mod rw_recovery_status;

src/frontend/src/expr/function_impl/rw_epoch_to_ts.rs

@@ -0,0 +1,22 @@
+// Copyright 2024 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use risingwave_common::types::Timestamptz;
+use risingwave_common::util::epoch::Epoch;
+use risingwave_expr::{function, Result};
+
+#[function("rw_epoch_to_ts(int8) -> timestamptz")]
+fn rw_epoch_to_ts(epoch: i64) -> Result<Timestamptz> {
+    Ok(Epoch(epoch as u64).as_timestamptz())
+}

src/frontend/src/expr/pure.rs

@@ -262,7 +262,8 @@ impl ExprVisitor for ImpureAnalyzer {
             | Type::MapDelete
             | Type::MapInsert
             | Type::MapLength
-            | Type::VnodeUser =>
+            | Type::VnodeUser
+            |Type::RwEpochToTs =>
             // expression output is deterministic(same result for the same input)
             {
                 func_call

src/frontend/src/optimizer/plan_expr_visitor/strong.rs

@@ -326,7 +326,8 @@ impl Strong {
             | ExprType::HasAnyColumnPrivilege
             | ExprType::HasSchemaPrivilege
             | ExprType::InetAton
-            | ExprType::InetNtoa => false,
+            | ExprType::InetNtoa
+            | ExprType::RwEpochToTs => false,
             ExprType::Unspecified => unreachable!(),
         }
     }