risingwavelabs · mergify · Dec 26, 2022 · Dec 26, 2022 · Dec 26, 2022 · Dec 26, 2022
diff --git a/dashboard/proto/gen/expr.ts b/dashboard/proto/gen/expr.ts
diff --git a/e2e_test/batch/functions/to_timestamp.slt.part b/e2e_test/batch/functions/to_timestamp.slt.part
@@ -0,0 +1,24 @@
+query T
+select to_timestamp('2022 12 25', 'YYYY MM DD');
+----
+2022-12-25 00:00:00
+
+query T
+select to_timestamp('2022/12/25 15:24:33', 'YYYY/MM/DD HH24:MI:SS');
+----
+2022-12-25 15:24:33
+
+query T
+select to_timestamp('2022/12/25 10:24:33', 'YYYY/MM/DD HH12:MI:SS');
+----
+2022-12-25 10:24:33
+
+query T
+select to_timestamp('2022/12/25 10:24:33', 'YYYY/MM/DD HH12:MI:SS');
+----
+2022-12-25 10:24:33
+
+query T
+select to_timestamp('2022/12/25 15', 'YYYY/MM/DD HH24');
+----
+2022-12-25 15:00:00
diff --git a/proto/expr.proto b/proto/expr.proto
@@ -43,9 +43,14 @@ message ExprNode {
     // date functions
     EXTRACT = 101;
     TUMBLE_START = 103;
+    // From f64 to timestamp.
+    // e.g. `select to_timestamp(1672044740.0)`
     TO_TIMESTAMP = 104;
     AT_TIME_ZONE = 105;
     DATE_TRUNC = 106;
+    // Parse text to timestamp by format string.
+    // e.g. `select to_timestamp('2022 08 21', 'YYYY MM DD')`
+    TO_TIMESTAMP1 = 107;
     // other functions
     CAST = 201;
     SUBSTR = 202;

diff --git a/src/expr/src/error.rs b/src/expr/src/error.rs
@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::borrow::Cow;
+
 pub use anyhow::anyhow;
 use regex;
 use risingwave_common::array::ArrayError;
@@ -40,7 +42,7 @@ pub enum ExprError {
     DivisionByZero,
 
     #[error("Parse error: {0}")]
-    Parse(&'static str),
+    Parse(Cow<'static, str>),
 
     #[error("Invalid parameter {name}: {reason}")]
     InvalidParam { name: &'static str, reason: String },
@@ -70,6 +72,12 @@ impl From<regex::Error> for ExprError {
     }
 }
 
+impl From<chrono::ParseError> for ExprError {
+    fn from(e: chrono::ParseError) -> Self {
+        Self::Parse(e.to_string().into())
+    }
+}
+
 impl From<ProstFieldNotFound> for ExprError {
     fn from(err: ProstFieldNotFound) -> Self {
         Self::Internal(anyhow::anyhow!(

diff --git a/src/expr/src/expr/build_expr_from_prost.rs b/src/expr/src/expr/build_expr_from_prost.rs
@@ -21,14 +21,19 @@ use crate::expr::expr_binary_bytes::{
     new_ltrim_characters, new_repeat, new_rtrim_characters, new_substr_start, new_to_char,
     new_trim_characters,
 };
-use crate::expr::expr_binary_nonnull::{new_binary_expr, new_date_trunc_expr, new_like_default};
+use crate::expr::expr_binary_nonnull::{
+    new_binary_expr, new_date_trunc_expr, new_like_default, new_to_timestamp,
+};
 use crate::expr::expr_binary_nullable::new_nullable_binary_expr;
 use crate::expr::expr_quaternary_bytes::new_overlay_for_exp;
 use crate::expr::expr_ternary_bytes::{
     new_overlay_exp, new_replace_expr, new_split_part_expr, new_substr_start_end,
     new_translate_expr,
 };
 use crate::expr::expr_to_char_const_tmpl::{ExprToCharConstTmpl, ExprToCharConstTmplContext};
+use crate::expr::expr_to_timestamp_const_tmpl::{
+    ExprToTimestampConstTmpl, ExprToTimestampConstTmplContext,
+};
 use crate::expr::expr_unary::{
     new_length_default, new_ltrim_expr, new_rtrim_expr, new_trim_expr, new_unary_expr,
 };
@@ -246,6 +251,29 @@ pub fn build_to_char_expr(prost: &ExprNode) -> Result<BoxedExpression> {
     }
 }
 
+pub fn build_to_timestamp_expr(prost: &ExprNode) -> Result<BoxedExpression> {
+    let (children, ret_type) = get_children_and_return_type(prost)?;
+    ensure!(children.len() == 2);
+    let data_expr = expr_build_from_prost(&children[0])?;
+    let tmpl_node = &children[1];
+    if let RexNode::Constant(tmpl_value) = tmpl_node.get_rex_node().unwrap()
+        && let Ok(Some(tmpl)) = deserialize_datum(tmpl_value.get_body().as_slice(), &DataType::from(tmpl_node.get_return_type().unwrap()))
+    {
+        let tmpl = tmpl.as_utf8();
+        let pattern = compile_pattern_to_chrono(tmpl);
+
+        Ok(ExprToTimestampConstTmpl {
+            ctx: ExprToTimestampConstTmplContext {
+                chrono_pattern: pattern,
+            },
+            child: data_expr,
+        }.boxed())
+    } else {
+        let tmpl_expr = expr_build_from_prost(&children[1])?;
+        Ok(new_to_timestamp(data_expr, tmpl_expr, ret_type))
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::vec;

diff --git a/src/expr/src/expr/expr_binary_nonnull.rs b/src/expr/src/expr/expr_binary_nonnull.rs
@@ -34,6 +34,7 @@ use crate::vector_op::like::like_default;
 use crate::vector_op::position::position;
 use crate::vector_op::round::round_digits;
 use crate::vector_op::timestampz::{timestamp_at_time_zone, timestampz_at_time_zone};
+use crate::vector_op::to_timestamp::to_timestamp;
 use crate::vector_op::tumble::{
     tumble_start_date, tumble_start_date_time, tumble_start_timestampz,
 };
@@ -704,6 +705,20 @@ pub fn new_like_default(
     ))
 }
 
+pub fn new_to_timestamp(
+    expr_ia1: BoxedExpression,
+    expr_ia2: BoxedExpression,
+    return_type: DataType,
+) -> BoxedExpression {
+    BinaryExpression::<Utf8Array, Utf8Array, NaiveDateTimeArray, _>::new(
+        expr_ia1,
+        expr_ia2,
+        return_type,
+        to_timestamp,
+    )
+    .boxed()
+}
+
 fn boolean_eq(l: &BoolArray, r: &BoolArray) -> BoolArray {
     let data = !(l.data() ^ r.data());
     let bitmap = l.null_bitmap() & r.null_bitmap();

diff --git a/src/expr/src/expr/expr_to_timestamp_const_tmpl.rs b/src/expr/src/expr/expr_to_timestamp_const_tmpl.rs
@@ -0,0 +1,72 @@
+// Copyright 2022 Singularity Data
+//
+// 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 std::sync::Arc;
+
+use itertools::Itertools;
+use risingwave_common::array::{Array, ArrayBuilder, NaiveDateTimeArrayBuilder, Utf8Array};
+use risingwave_common::row::OwnedRow;
+use risingwave_common::types::{DataType, Datum, ScalarImpl};
+
+use super::Expression;
+use crate::vector_op::to_char::ChronoPattern;
+use crate::vector_op::to_timestamp::to_timestamp_const_tmpl;
+
+#[derive(Debug)]
+pub(crate) struct ExprToTimestampConstTmplContext {
+    pub(crate) chrono_pattern: ChronoPattern,
+}
+
+#[derive(Debug)]
+pub(crate) struct ExprToTimestampConstTmpl {
+    pub(crate) child: Box<dyn Expression>,
+    pub(crate) ctx: ExprToTimestampConstTmplContext,
+}
+
+impl Expression for ExprToTimestampConstTmpl {
+    fn return_type(&self) -> DataType {
+        DataType::Varchar
+    }
+
+    fn eval(
+        &self,
+        input: &risingwave_common::array::DataChunk,
+    ) -> crate::Result<risingwave_common::array::ArrayRef> {
+        let data_arr = self.child.eval_checked(input)?;
+        let data_arr: &Utf8Array = data_arr.as_ref().into();
+        let mut output = NaiveDateTimeArrayBuilder::new(input.capacity());
+        for (data, vis) in data_arr.iter().zip_eq(input.vis().iter()) {
+            if !vis {
+                output.append_null();
+            } else if let Some(data) = data {
+                let res = to_timestamp_const_tmpl(data, &self.ctx.chrono_pattern)?;
+                output.append(Some(res));
+            } else {
+                output.append_null();
+            }
+        }
+
+        Ok(Arc::new(output.finish().into()))
+    }
+
+    fn eval_row(&self, input: &OwnedRow) -> crate::Result<Datum> {
+        let data = self.child.eval_row(input)?;
+        Ok(if let Some(ScalarImpl::Utf8(data)) = data {
+            let res = to_timestamp_const_tmpl(&data, &self.ctx.chrono_pattern)?;
+            Some(res.into())
+        } else {
+            None
+        })
+    }
+}
diff --git a/src/expr/src/expr/mod.rs b/src/expr/src/expr/mod.rs
@@ -32,6 +32,7 @@ mod expr_quaternary_bytes;
 mod expr_regexp;
 mod expr_ternary_bytes;
 mod expr_to_char_const_tmpl;
+mod expr_to_timestamp_const_tmpl;
 pub mod expr_unary;
 mod expr_vnode;
 mod template;
@@ -113,6 +114,7 @@ pub fn build_from_prost(prost: &ExprNode) -> Result<BoxedExpression> {
             build_nullable_binary_expr_prost(prost)
         }
         ToChar => build_to_char_expr(prost),
+        ToTimestamp1 => build_to_timestamp_expr(prost),
         Length => build_length_expr(prost),
         Replace => build_replace_expr(prost),
         Like => build_like_expr(prost),

diff --git a/src/expr/src/sig/func.rs b/src/expr/src/sig/func.rs
@@ -233,6 +233,7 @@ fn build_type_derive_map() -> FuncSigMap {
         T::Timestampz,
     );
     map.insert(E::ToTimestamp, vec![T::Float64], T::Timestampz);
+    map.insert(E::ToTimestamp1, vec![T::Varchar, T::Varchar], T::Timestamp);
     map.insert(E::AtTimeZone, vec![T::Timestamp, T::Varchar], T::Timestampz);
     map.insert(E::AtTimeZone, vec![T::Timestampz, T::Varchar], T::Timestamp);
     map.insert(E::DateTrunc, vec![T::Varchar, T::Timestamp], T::Timestamp);