feat(query): load ndjson support ts with diff units. (#17239)

* chore: Split functions crate into multi crates (#17133)

* chore: Split functions crate into multi crates

* split new crate databend-functions-scalar-arithmetic databend-functions-scalar-arithmetic-modulo databend-functions-scalar-datetime databend-functions-scalar-decimal-utils

* split numeric basic arithmetic and other bit arithmetic

* fix ut

* restore int/double arithmetic

* fix

* feat(query): load ndjson support ts with diff units. (#17203)

* feat(query): load ndjson/csv/tsv support ts with diff units.

* fix

* refactor

* refactor

* fix clippy

---------

Co-authored-by: TCeason <[email protected]>

Author: youngsofun

Cargo.lock

@@ -36,6 +36,12 @@ members = [
     "src/query/expression",
     "src/query/formats",
     "src/query/functions",
+    "src/query/functions",
+    "src/query/functions/src/scalars/mathematics",
+    "src/query/functions/src/scalars/geographic",
+    "src/query/functions/src/scalars/timestamp",
+    "src/query/functions/src/scalars/numeric_basic_arithmetic",
+    "src/query/functions/src/scalars/arithmetic",
     "src/query/management",
     "src/query/pipeline/core",
     "src/query/pipeline/sinks",
@@ -184,6 +190,12 @@ databend-enterprise-storage-quota = { path = "src/query/ee_features/storage_quot
 databend-enterprise-stream-handler = { path = "src/query/ee_features/stream_handler" }
 databend-enterprise-vacuum-handler = { path = "src/query/ee_features/vacuum_handler" }
 databend-enterprise-virtual-column = { path = "src/query/ee_features/virtual_column" }
+databend-functions-scalar-arithmetic = { path = "src/query/functions/src/scalars/arithmetic" }
+databend-functions-scalar-datetime = { path = "src/query/functions/src/scalars/timestamp" }
+databend-functions-scalar-decimal = { path = "src/query/functions/src/scalars/decimal" }
+databend-functions-scalar-geo = { path = "src/query/functions/src/scalars/geographic" }
+databend-functions-scalar-math = { path = "src/query/functions/src/scalars/mathematics" }
+databend-functions-scalar-numeric-basic-arithmetic = { path = "src/query/functions/src/scalars/numeric_basic_arithmetic" }
 databend-meta = { path = "src/meta/service" }
 databend-query = { path = "src/query/service" }
 databend-sqllogictests = { path = "tests/sqllogictests" }

Cargo.toml

@@ -19,6 +19,7 @@ databend-common-expression = { workspace = true }
 databend-common-io = { workspace = true }
 databend-common-meta-app = { workspace = true }
 databend-common-settings = { workspace = true }
+databend-functions-scalar-datetime = { workspace = true }
 databend-storages-common-blocks = { workspace = true }
 databend-storages-common-table-meta = { workspace = true }
 

src/query/formats/Cargo.toml

@@ -0,0 +1,59 @@
+// Copyright 2021 Datafuse 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 std::io::Cursor;
+
+use bstr::ByteSlice;
+use databend_common_exception::ErrorCode;
+use databend_common_exception::Result;
+use databend_common_expression::types::timestamp::clamp_timestamp;
+use databend_common_io::cursor_ext::read_num_text_exact;
+use databend_common_io::cursor_ext::BufferReadDateTimeExt;
+use databend_common_io::cursor_ext::DateTimeResType;
+use databend_common_io::cursor_ext::ReadBytesExt;
+use databend_functions_scalar_datetime::datetime::int64_to_timestamp;
+
+use crate::InputCommonSettings;
+
+pub(crate) fn read_timestamp(
+    column: &mut Vec<i64>,
+    data: &[u8],
+    settings: &InputCommonSettings,
+) -> Result<()> {
+    let ts = if !data.contains(&b'-') {
+        int64_to_timestamp(read_num_text_exact(data)?)
+    } else {
+        let mut buffer_readr = Cursor::new(&data);
+        let t = buffer_readr.read_timestamp_text(&settings.jiff_timezone)?;
+        match t {
+            DateTimeResType::Datetime(t) => {
+                if !buffer_readr.eof() {
+                    let data = data.to_str().unwrap_or("not utf8");
+                    let msg = format!(
+                        "fail to deserialize timestamp, unexpected end at pos {} of {}",
+                        buffer_readr.position(),
+                        data
+                    );
+                    return Err(ErrorCode::BadBytes(msg));
+                }
+                let mut ts = t.timestamp().as_microsecond();
+                clamp_timestamp(&mut ts);
+                ts
+            }
+            _ => unreachable!(),
+        }
+    };
+    column.push(ts);
+    Ok(())
+}

src/query/formats/src/field_decoder/common.rs

@@ -37,7 +37,6 @@ use databend_common_expression::types::decimal::DecimalSize;
 use databend_common_expression::types::nullable::NullableColumnBuilder;
 use databend_common_expression::types::number::Number;
 use databend_common_expression::types::string::StringColumnBuilder;
-use databend_common_expression::types::timestamp::clamp_timestamp;
 use databend_common_expression::types::AnyType;
 use databend_common_expression::types::MutableBitmap;
 use databend_common_expression::types::NumberColumnBuilder;
@@ -51,7 +50,6 @@ use databend_common_io::constants::NULL_BYTES_UPPER;
 use databend_common_io::constants::TRUE_BYTES_LOWER;
 use databend_common_io::cursor_ext::BufferReadDateTimeExt;
 use databend_common_io::cursor_ext::BufferReadStringExt;
-use databend_common_io::cursor_ext::DateTimeResType;
 use databend_common_io::cursor_ext::ReadBytesExt;
 use databend_common_io::cursor_ext::ReadCheckPointExt;
 use databend_common_io::cursor_ext::ReadNumberExt;
@@ -62,9 +60,9 @@ use databend_common_io::prelude::FormatSettings;
 use databend_common_io::Interval;
 use jsonb::parse_value;
 use lexical_core::FromLexical;
-use num::cast::AsPrimitive;
 use num_traits::NumCast;
 
+use crate::field_decoder::common::read_timestamp;
 use crate::FieldDecoder;
 use crate::InputCommonSettings;
 
@@ -323,26 +321,7 @@ impl FastFieldDecoderValues {
     ) -> Result<()> {
         let mut buf = Vec::new();
         self.read_string_inner(reader, &mut buf, positions)?;
-        let mut buffer_readr = Cursor::new(&buf);
-        let ts = buffer_readr.read_timestamp_text(&self.common_settings().jiff_timezone)?;
-        match ts {
-            DateTimeResType::Datetime(ts) => {
-                if !buffer_readr.eof() {
-                    let data = buf.to_str().unwrap_or("not utf8");
-                    let msg = format!(
-                        "fail to deserialize timestamp, unexpected end at pos {} of {}",
-                        buffer_readr.position(),
-                        data
-                    );
-                    return Err(ErrorCode::BadBytes(msg));
-                }
-                let mut micros = ts.timestamp().as_microsecond();
-                clamp_timestamp(&mut micros);
-                column.push(micros.as_());
-            }
-            _ => unreachable!(),
-        }
-        Ok(())
+        read_timestamp(column, &buf, self.common_settings())
     }
 
     fn read_array<R: AsRef<[u8]>>(

src/query/formats/src/field_decoder/fast_values.rs

@@ -40,6 +40,7 @@ use databend_common_io::cursor_ext::DateTimeResType;
 use databend_common_io::geography::geography_from_ewkt;
 use databend_common_io::geometry_from_ewkt;
 use databend_common_io::parse_bitmap;
+use databend_functions_scalar_datetime::datetime::int64_to_timestamp;
 use jiff::tz::TimeZone;
 use lexical_core::FromLexical;
 use num::cast::AsPrimitive;
@@ -296,8 +297,8 @@ impl FieldJsonAstDecoder {
                 Ok(())
             }
             Value::Number(number) => match number.as_i64() {
-                Some(mut n) => {
-                    clamp_timestamp(&mut n);
+                Some(n) => {
+                    let n = int64_to_timestamp(n);
                     column.push(n);
                     Ok(())
                 }

src/query/formats/src/field_decoder/json_ast.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+mod common;
 mod fast_values;
 mod json_ast;
 mod nested;

src/query/formats/src/field_decoder/mod.rs

@@ -32,7 +32,6 @@ use databend_common_expression::types::decimal::DecimalSize;
 use databend_common_expression::types::nullable::NullableColumnBuilder;
 use databend_common_expression::types::number::Number;
 use databend_common_expression::types::string::StringColumnBuilder;
-use databend_common_expression::types::timestamp::clamp_timestamp;
 use databend_common_expression::types::AnyType;
 use databend_common_expression::types::MutableBitmap;
 use databend_common_expression::types::NumberColumnBuilder;
@@ -45,7 +44,6 @@ use databend_common_io::constants::NULL_BYTES_UPPER;
 use databend_common_io::constants::TRUE_BYTES_LOWER;
 use databend_common_io::cursor_ext::BufferReadDateTimeExt;
 use databend_common_io::cursor_ext::BufferReadStringExt;
-use databend_common_io::cursor_ext::DateTimeResType;
 use databend_common_io::cursor_ext::ReadBytesExt;
 use databend_common_io::cursor_ext::ReadCheckPointExt;
 use databend_common_io::cursor_ext::ReadNumberExt;
@@ -57,6 +55,7 @@ use jsonb::parse_value;
 use lexical_core::FromLexical;
 
 use crate::binary::decode_binary;
+use crate::field_decoder::common::read_timestamp;
 use crate::FileFormatOptionsExt;
 use crate::InputCommonSettings;
 
@@ -67,9 +66,9 @@ pub struct NestedValues {
 
 impl NestedValues {
     /// Consider map/tuple/array as a private object format like JSON.
-    /// Currently we assume it as a fixed format, embed it in "strings" of other formats.
-    /// So we can used the same code to encode/decode in clients.
-    /// It maybe need to be configurable in future,
+    /// Currently, we assume it as a fixed format, embed it in "strings" of other formats.
+    /// So we can use the same code to encode/decode in clients.
+    /// It maybe needs to be configurable in the future,
     /// to read data from other DB which also support map/tuple/array.
     pub fn create(options_ext: &FileFormatOptionsExt) -> Self {
         NestedValues {
@@ -281,30 +280,7 @@ impl NestedValues {
     ) -> Result<()> {
         let mut buf = Vec::new();
         self.read_string_inner(reader, &mut buf)?;
-        let mut buffer_readr = Cursor::new(&buf);
-        let mut ts = if !buf.contains(&b'-') {
-            buffer_readr.read_num_text_exact()?
-        } else {
-            let t = buffer_readr.read_timestamp_text(&self.common_settings().jiff_timezone)?;
-            match t {
-                DateTimeResType::Datetime(t) => {
-                    if !buffer_readr.eof() {
-                        let data = buf.to_str().unwrap_or("not utf8");
-                        let msg = format!(
-                            "fail to deserialize timestamp, unexpected end at pos {} of {}",
-                            buffer_readr.position(),
-                            data
-                        );
-                        return Err(ErrorCode::BadBytes(msg));
-                    }
-                    t.timestamp().as_microsecond()
-                }
-                _ => unreachable!(),
-            }
-        };
-        clamp_timestamp(&mut ts);
-        column.push(ts);
-        Ok(())
+        read_timestamp(column, &buf, self.common_settings())
     }
 
     fn read_bitmap<R: AsRef<[u8]>>(