Add datafusion-functions-array crate

Author: alamb

.github/workflows/rust.yml

@@ -82,6 +82,9 @@ jobs:
       - name: Check function packages (encoding_expressions)
         run: cargo check --no-default-features --features=encoding_expressions -p datafusion
 
+      - name: Check function packages (array_expressions)
+        run: cargo check --no-default-features --features=array_expressions -p datafusion
+
       - name: Check Cargo.lock for datafusion-cli
         run: |
           # If this test fails, try running `cargo update` in the `datafusion-cli` directory

Cargo.toml

@@ -17,7 +17,7 @@
 
 [workspace]
 exclude = ["datafusion-cli"]
-members = ["datafusion/common", "datafusion/core", "datafusion/expr", "datafusion/execution", "datafusion/functions", "datafusion/optimizer", "datafusion/physical-expr", "datafusion/physical-plan", "datafusion/proto", "datafusion/proto/gen", "datafusion/sql", "datafusion/sqllogictest", "datafusion/substrait", "datafusion/wasmtest", "datafusion-examples", "docs", "test-utils", "benchmarks",
+members = ["datafusion/common", "datafusion/core", "datafusion/expr", "datafusion/execution", "datafusion/functions", "datafusion/functions-array",  "datafusion/optimizer", "datafusion/physical-expr", "datafusion/physical-plan", "datafusion/proto", "datafusion/proto/gen", "datafusion/sql", "datafusion/sqllogictest", "datafusion/substrait", "datafusion/wasmtest", "datafusion-examples", "docs", "test-utils", "benchmarks",
 ]
 resolver = "2"
 
@@ -50,6 +50,7 @@ datafusion-common = { path = "datafusion/common", version = "35.0.0" }
 datafusion-execution = { path = "datafusion/execution", version = "35.0.0" }
 datafusion-expr = { path = "datafusion/expr", version = "35.0.0" }
 datafusion-functions = { path = "datafusion/functions", version = "35.0.0" }
+datafusion-functions-array = { path = "datafusion/functions-array", version = "35.0.0" }
 datafusion-optimizer = { path = "datafusion/optimizer", version = "35.0.0" }
 datafusion-physical-expr = { path = "datafusion/physical-expr", version = "35.0.0" }
 datafusion-physical-plan = { path = "datafusion/physical-plan", version = "35.0.0" }

README.md

@@ -75,6 +75,7 @@ This crate has several [features] which can be specified in your `Cargo.toml`.
 
 Default features:
 
+- `array_expressions`: functions for working with arrays such as `array_to_string`
 - `compression`: reading files compressed with `xz2`, `bzip2`, `flate2`, and `zstd`
 - `crypto_expressions`: cryptographic functions such as `md5` and `sha256`
 - `encoding_expressions`: `encode` and `decode` functions

datafusion-examples/examples/simple_udf.rs

@@ -28,7 +28,6 @@ use datafusion::error::Result;
 use datafusion::prelude::*;
 use datafusion_common::cast::as_float64_array;
 use datafusion_expr::ColumnarValue;
-use datafusion_physical_expr::functions::columnar_values_to_array;
 use std::sync::Arc;
 
 /// create local execution context with an in-memory table:
@@ -71,7 +70,7 @@ async fn main() -> Result<()> {
         // this is guaranteed by DataFusion based on the function's signature.
         assert_eq!(args.len(), 2);
 
-        let args = columnar_values_to_array(args)?;
+        let args = ColumnarValue::values_to_arrays(args)?;
 
         // 1. cast both arguments to f64. These casts MUST be aligned with the signature or this function panics!
         let base = as_float64_array(&args[0]).expect("cast failed");

datafusion/core/Cargo.toml

@@ -35,11 +35,12 @@ path = "src/lib.rs"
 
 [features]
 # Used to enable the avro format
+array_expressions = ["datafusion-functions-array"]
 avro = ["apache-avro", "num-traits", "datafusion-common/avro"]
 backtrace = ["datafusion-common/backtrace"]
 compression = ["xz2", "bzip2", "flate2", "zstd", "async-compression", "tokio-util"]
 crypto_expressions = ["datafusion-physical-expr/crypto_expressions", "datafusion-optimizer/crypto_expressions"]
-default = ["crypto_expressions", "encoding_expressions", "regex_expressions", "unicode_expressions", "compression", "parquet"]
+default = ["array_expressions", "crypto_expressions", "encoding_expressions", "regex_expressions", "unicode_expressions", "compression", "parquet"]
 encoding_expressions = ["datafusion-functions/encoding_expressions"]
 # Used for testing ONLY: causes all values to hash to the same value (test for collisions)
 force_hash_collisions = []
@@ -65,7 +66,8 @@ dashmap = { workspace = true }
 datafusion-common = { path = "../common", version = "35.0.0", features = ["object_store"], default-features = false }
 datafusion-execution = { workspace = true }
 datafusion-expr = { workspace = true }
-datafusion-functions = { path = "../functions", version = "35.0.0" }
+datafusion-functions = {workspace = true }
+datafusion-functions-array = { workspace=true, optional = true }
 datafusion-optimizer = { path = "../optimizer", version = "35.0.0", default-features = false }
 datafusion-physical-expr = { path = "../physical-expr", version = "35.0.0", default-features = false }
 datafusion-physical-plan = { workspace = true }

datafusion/core/src/execution/context/mod.rs

@@ -1348,6 +1348,11 @@ impl SessionState {
         datafusion_functions::register_all(&mut new_self)
             .expect("can not register built in functions");
 
+        // register crate of array expressions (if enabled)
+        #[cfg(feature = "array_expressions")]
+        datafusion_functions_array::register_all(&mut new_self)
+            .expect("can not register array expressions");
+
         new_self
     }
     /// Returns new [`SessionState`] using the provided

datafusion/core/src/lib.rs

@@ -521,6 +521,12 @@ pub mod functions {
     pub use datafusion_functions::*;
 }
 
+/// re-export of [`datafusion_functions_array`] crate, if "array_expressions" feature is enabled
+pub mod functions_array {
+    #[cfg(feature = "array_expressions")]
+    pub use datafusion_functions::*;
+}
+
 #[cfg(test)]
 pub mod test;
 pub mod test_util;

datafusion/core/src/prelude.rs

@@ -39,6 +39,8 @@ pub use datafusion_expr::{
     Expr,
 };
 pub use datafusion_functions::expr_fn::*;
+#[cfg(feature = "array_expressions")]
+pub use datafusion_functions_array::expr_fn::*;
 
 pub use std::ops::Not;
 pub use std::ops::{Add, Div, Mul, Neg, Rem, Sub};

datafusion/core/tests/dataframe/dataframe_functions.rs

@@ -20,6 +20,8 @@ use arrow::{
     array::{Int32Array, StringArray},
     record_batch::RecordBatch,
 };
+use arrow_array::types::Int32Type;
+use arrow_array::ListArray;
 use arrow_schema::SchemaRef;
 use std::sync::Arc;
 
@@ -40,6 +42,7 @@ fn test_schema() -> SchemaRef {
     Arc::new(Schema::new(vec![
         Field::new("a", DataType::Utf8, false),
         Field::new("b", DataType::Int32, false),
+        Field::new("l", DataType::new_list(DataType::Int32, true), true),
     ]))
 }
 
@@ -57,6 +60,12 @@ async fn create_test_table() -> Result<DataFrame> {
                 "123AbcDef",
             ])),
             Arc::new(Int32Array::from(vec![1, 10, 10, 100])),
+            Arc::new(ListArray::from_iter_primitive::<Int32Type, _, _>(vec![
+                Some(vec![Some(0), Some(1), Some(2)]),
+                None,
+                Some(vec![Some(3), None, Some(5)]),
+                Some(vec![Some(6), Some(7)]),
+            ])),
         ],
     )?;
 
@@ -67,7 +76,7 @@ async fn create_test_table() -> Result<DataFrame> {
     ctx.table("test").await
 }
 
-/// Excutes an expression on the test dataframe as a select.
+/// Executes an expression on the test dataframe as a select.
 /// Compares formatted output of a record batch with an expected
 /// vector of strings, using the assert_batch_eq! macro
 macro_rules! assert_fn_batches {
@@ -841,3 +850,22 @@ async fn test_fn_decode() -> Result<()> {
 
     Ok(())
 }
+
+#[tokio::test]
+async fn test_fn_array_to_string() -> Result<()> {
+    let expr = array_to_string(col("l"), lit("***"));
+
+    let expected = [
+        "+-------------------------------------+",
+        "| array_to_string(test.l,Utf8(\"***\")) |",
+        "+-------------------------------------+",
+        "| 0***1***2                           |",
+        "|                                     |",
+        "| 3***5                               |",
+        "| 6***7                               |",
+        "+-------------------------------------+",
+    ];
+    assert_fn_batches!(expr, expected);
+
+    Ok(())
+}

datafusion/expr/src/built_in_function.rs

@@ -175,8 +175,6 @@ pub enum BuiltinScalarFunction {
     ArrayReverse,
     /// array_slice
     ArraySlice,
-    /// array_to_string
-    ArrayToString,
     /// array_intersect
     ArrayIntersect,
     /// array_union
@@ -432,7 +430,6 @@ impl BuiltinScalarFunction {
             BuiltinScalarFunction::ArrayReverse => Volatility::Immutable,
             BuiltinScalarFunction::Flatten => Volatility::Immutable,
             BuiltinScalarFunction::ArraySlice => Volatility::Immutable,
-            BuiltinScalarFunction::ArrayToString => Volatility::Immutable,
             BuiltinScalarFunction::ArrayIntersect => Volatility::Immutable,
             BuiltinScalarFunction::ArrayUnion => Volatility::Immutable,
             BuiltinScalarFunction::ArrayResize => Volatility::Immutable,
@@ -628,7 +625,6 @@ impl BuiltinScalarFunction {
             BuiltinScalarFunction::ArrayReverse => Ok(input_expr_types[0].clone()),
             BuiltinScalarFunction::ArraySlice => Ok(input_expr_types[0].clone()),
             BuiltinScalarFunction::ArrayResize => Ok(input_expr_types[0].clone()),
-            BuiltinScalarFunction::ArrayToString => Ok(Utf8),
             BuiltinScalarFunction::ArrayIntersect => {
                 match (input_expr_types[0].clone(), input_expr_types[1].clone()) {
                     (DataType::Null, DataType::Null) | (DataType::Null, _) => {
@@ -970,9 +966,6 @@ impl BuiltinScalarFunction {
                 Signature::variadic_any(self.volatility())
             }
 
-            BuiltinScalarFunction::ArrayToString => {
-                Signature::variadic_any(self.volatility())
-            }
             BuiltinScalarFunction::ArrayIntersect => Signature::any(2, self.volatility()),
             BuiltinScalarFunction::ArrayUnion => Signature::any(2, self.volatility()),
             BuiltinScalarFunction::Cardinality => Signature::any(1, self.volatility()),
@@ -1574,12 +1567,6 @@ impl BuiltinScalarFunction {
             }
             BuiltinScalarFunction::ArrayReverse => &["array_reverse", "list_reverse"],
             BuiltinScalarFunction::ArraySlice => &["array_slice", "list_slice"],
-            BuiltinScalarFunction::ArrayToString => &[
-                "array_to_string",
-                "list_to_string",
-                "array_join",
-                "list_join",
-            ],
             BuiltinScalarFunction::ArrayUnion => &["array_union", "list_union"],
             BuiltinScalarFunction::Cardinality => &["cardinality"],
             BuiltinScalarFunction::ArrayResize => &["array_resize", "list_resize"],

datafusion/expr/src/columnar_value.rs

@@ -20,7 +20,7 @@
 use arrow::array::ArrayRef;
 use arrow::array::NullArray;
 use arrow::datatypes::DataType;
-use datafusion_common::{Result, ScalarValue};
+use datafusion_common::{internal_err, DataFusionError, Result, ScalarValue};
 use std::sync::Arc;
 
 /// Represents the result of evaluating an expression: either a single
@@ -75,4 +75,51 @@ impl ColumnarValue {
     pub fn create_null_array(num_rows: usize) -> Self {
         ColumnarValue::Array(Arc::new(NullArray::new(num_rows)))
     }
+
+    /// A helper function used to infer the length of arguments of Scalar
+    /// functions and convert [`ColumnarValue`]s to [`ArrayRef`]s with the
+    /// inferred length.
+    ///
+    /// # Note on Performance
+    ///
+    /// This function expands any `ScalalarValues` to arrays of the same length.
+    /// This allows a single function implementaiton in terms of arrays, but it
+    /// can be inefficient. It is recommended to provide specialized
+    /// implementations for scalar values if performance is a concern.
+    ///
+    /// # Errors
+    ///
+    /// If there are multiple array arguments that have different lengths
+    pub fn values_to_arrays(args: &[ColumnarValue]) -> Result<Vec<ArrayRef>> {
+        if args.is_empty() {
+            return Ok(vec![]);
+        }
+
+        let mut array_len = None;
+        for arg in args {
+            array_len = match (arg, array_len) {
+                (ColumnarValue::Array(a), None) => Some(a.len()),
+                (ColumnarValue::Array(a), Some(array_len)) => {
+                    if array_len == a.len() {
+                        Some(array_len)
+                    } else {
+                        return internal_err!(
+                            "Arguments has mixed length. Expected length: {array_len}, found length: {}",a.len()
+                        );
+                    }
+                }
+                (ColumnarValue::Scalar(_), array_len) => array_len,
+            }
+        }
+
+        // If array_len is none, it means there are only scalars, so make a 1 element array
+        let inferred_length = array_len.unwrap_or(1);
+
+        let args = args
+            .iter()
+            .map(|arg| arg.clone().into_array(inferred_length))
+            .collect::<Result<Vec<_>>>()?;
+
+        Ok(args)
+    }
 }

datafusion/expr/src/expr_fn.rs

@@ -740,12 +740,6 @@ scalar_expr!(
     array begin end stride,
     "returns a slice of the array."
 );
-scalar_expr!(
-    ArrayToString,
-    array_to_string,
-    array delimiter,
-    "converts each element to its text representation."
-);
 scalar_expr!(ArrayUnion, array_union, array1 array2, "returns an array of the elements in the union of array1 and array2 without duplicates.");
 
 scalar_expr!(
@@ -1447,7 +1441,6 @@ mod test {
         test_scalar_expr!(ArrayReplace, array_replace, array, from, to);
         test_scalar_expr!(ArrayReplaceN, array_replace_n, array, from, to, max);
         test_scalar_expr!(ArrayReplaceAll, array_replace_all, array, from, to);
-        test_scalar_expr!(ArrayToString, array_to_string, array, delimiter);
         test_unary_scalar_expr!(Cardinality, cardinality);
         test_nary_scalar_expr!(MakeArray, array, input);
 

datafusion/functions-array/Cargo.toml

@@ -0,0 +1,45 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+
+[package]
+name = "datafusion-functions-array"
+description = "Array Function packages for the DataFusion query engine"
+keywords = ["datafusion", "logical", "plan", "expressions"]
+readme = "README.md"
+version = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+repository = { workspace = true }
+license = { workspace = true }
+authors = { workspace = true }
+rust-version = { workspace = true }
+
+[features]
+
+[lib]
+name = "datafusion_functions_array"
+path = "src/lib.rs"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+arrow = { workspace = true }
+datafusion-common = { workspace = true }
+datafusion-execution = { workspace = true }
+datafusion-expr = { workspace = true }
+log = "0.4.20"
+paste = "1.0.14"

datafusion/functions-array/README.md

@@ -0,0 +1,27 @@
+<!---
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you 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.
+-->
+
+# DataFusion Function Library
+
+[DataFusion][df] is an extensible query execution framework, written in Rust, that uses Apache Arrow as its in-memory format.
+
+This crate contains functions for working with arrays, such as `array_append` that work with
+`ListArray`, `LargeListArray` and `FixedListArray` types from the `arrow` crate.
+
+[df]: https://crates.io/crates/datafusion