refactor(func): optimize function array_aggregate (#19005)

* refine

* fix

* update

* update

Author: forsaken628

Cargo.lock

@@ -0,0 +1,92 @@
+// 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 chrono_tz::UTC;
+use databend_common_exception::ErrorCode;
+use databend_common_exception::Result;
+use databend_common_expression::types::DataType;
+use databend_common_expression::Column;
+use databend_common_expression::ColumnBuilder;
+use databend_common_io::GeometryDataType;
+use jiff::tz::TimeZone;
+use serde_json::Value;
+
+use crate::field_decoder::FieldJsonAstDecoder;
+use crate::FileFormatOptionsExt;
+
+fn default_json_options() -> FileFormatOptionsExt {
+    FileFormatOptionsExt {
+        ident_case_sensitive: false,
+        headers: 0,
+        json_compact: false,
+        json_strings: false,
+        disable_variant_check: false,
+        timezone: UTC,
+        jiff_timezone: TimeZone::UTC,
+        is_select: false,
+        is_clickhouse: false,
+        is_rounding_mode: true,
+        geometry_format: GeometryDataType::default(),
+        enable_dst_hour_fix: false,
+    }
+}
+
+pub fn column_from_json_value(data_type: &DataType, json: Value) -> Result<Column> {
+    let rows = match json {
+        Value::Array(values) => values,
+        other => {
+            return Err(ErrorCode::BadArguments(format!(
+                "from_json! expects a json array to describe column values, got {other:?}"
+            )))
+        }
+    };
+
+    let options = default_json_options();
+    let decoder = FieldJsonAstDecoder::create(&options);
+    let mut builder = ColumnBuilder::with_capacity(data_type, rows.len());
+    for value in rows {
+        decoder.read_field(&mut builder, &value)?;
+    }
+    Ok(builder.build())
+}
+
+#[macro_export]
+macro_rules! column_from_json {
+    ($data_type:expr, $($json:tt)+) => {{
+        $crate::column_from_json_value(&$data_type, ::serde_json::json!($($json)+))
+            .expect("from_json! expects a valid json literal for the provided type")
+    }};
+}
+
+#[cfg(test)]
+mod tests {
+    use databend_common_expression::types::*;
+    use databend_common_expression::FromData;
+
+    #[test]
+    fn test_from_json_macro_strings() {
+        let column = column_from_json!(DataType::String, ["a", "b", "c"]);
+        assert_eq!(column, StringType::from_data(vec!["a", "b", "c"]));
+    }
+
+    #[test]
+    fn test_from_json_nullable_booleans() {
+        let data_type = DataType::Nullable(Box::new(DataType::Boolean));
+        let column = column_from_json!(data_type, [true, null, false]);
+        assert_eq!(
+            column,
+            BooleanType::from_data_with_validity(vec![true, false, false], vec![true, false, true])
+        );
+    }
+}

src/query/formats/src/column_from_json.rs

@@ -113,9 +113,20 @@ impl FieldJsonAstDecoder {
             ColumnBuilder::Geography(c) => self.read_geography(c, value),
             ColumnBuilder::Interval(c) => self.read_interval(c, value),
             ColumnBuilder::Vector(c) => self.read_vector(c, value),
-            ColumnBuilder::EmptyArray { .. } | ColumnBuilder::EmptyMap { .. } => {
-                Err(ErrorCode::Unimplemented("empty array/map literal"))
-            }
+            ColumnBuilder::EmptyArray { len } => match value.as_array() {
+                Some(array) if array.is_empty() => {
+                    *len += 1;
+                    Ok(())
+                }
+                _ => Err(ErrorCode::BadBytes("Incorrect empty array value")),
+            },
+            ColumnBuilder::EmptyMap { len } => match value.as_object() {
+                Some(array) if array.is_empty() => {
+                    *len += 1;
+                    Ok(())
+                }
+                _ => Err(ErrorCode::BadBytes("Incorrect empty map value")),
+            },
             ColumnBuilder::Opaque(_) => Err(ErrorCode::Unimplemented(
                 "Opaque type not supported in json_ast",
             )),

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

@@ -12,14 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(clippy::uninlined_format_args)]
 #![feature(box_patterns)]
 #![feature(cursor_split)]
 
-extern crate core;
-
 mod binary;
 mod clickhouse;
+pub mod column_from_json;
 mod common_settings;
 mod delimiter;
 mod field_decoder;
@@ -28,6 +26,7 @@ mod file_format_type;
 pub mod output_format;
 
 pub use clickhouse::ClickhouseFormatType;
+pub use column_from_json::column_from_json_value;
 pub use delimiter::RecordDelimiter;
 pub use field_decoder::*;
 pub use file_format_type::parse_timezone;

src/query/formats/src/lib.rs

@@ -70,6 +70,7 @@ unicase = { workspace = true }
 [dev-dependencies]
 comfy-table = { workspace = true }
 databend-common-ast = { workspace = true }
+databend-common-formats = { workspace = true }
 divan = { workspace = true }
 goldenfile = { workspace = true }
 

src/query/functions/Cargo.toml

@@ -173,7 +173,7 @@ pub fn eval_aggr(
     rows: usize,
     sort_descs: Vec<AggregateFunctionSortDesc>,
 ) -> Result<(Column, DataType)> {
-    eval_aggr_for_test(name, params, entries, rows, false, sort_descs)
+    eval_aggr_inner(name, params, entries, rows, false, sort_descs)
 }
 
 pub fn eval_aggr_for_test(
@@ -183,6 +183,18 @@ pub fn eval_aggr_for_test(
     rows: usize,
     with_serialize: bool,
     sort_descs: Vec<AggregateFunctionSortDesc>,
+) -> Result<(Column, DataType)> {
+    eval_aggr_inner(name, params, entries, rows, with_serialize, sort_descs)
+}
+
+#[inline]
+fn eval_aggr_inner(
+    name: &str,
+    params: Vec<Scalar>,
+    entries: &[BlockEntry],
+    rows: usize,
+    with_serialize: bool,
+    sort_descs: Vec<AggregateFunctionSortDesc>,
 ) -> Result<(Column, DataType)> {
     let factory = AggregateFunctionFactory::instance();
     let arguments = entries.iter().map(BlockEntry::data_type).collect();