support type cast failure message

Cargo.toml

@@ -31,5 +31,5 @@ members = [
 exclude = ["python"]
 
 [patch.crates-io]
-arrow2 = { git = "https://github.com/jorgecarleitao/arrow2.git", rev = "2db7f57345192c6b9ae83bd5d1f99b2c57032648" }
-arrow-flight = { git = "https://github.com/jorgecarleitao/arrow2.git", rev = "2db7f57345192c6b9ae83bd5d1f99b2c57032648" }
+arrow2 = { git = "https://github.com/jorgecarleitao/arrow2.git", rev = "b7e991366104d1647b955a828e0551256ef2e7c9" }
+arrow-flight = { git = "https://github.com/jorgecarleitao/arrow2.git", rev = "b7e991366104d1647b955a828e0551256ef2e7c9" }

datafusion/src/physical_plan/array_expressions.rs

@@ -38,18 +38,19 @@ fn array_array(arrays: &[&dyn Array]) -> Result<ArrayRef> {
                 $DATA_TYPE,
             );
             let mut array = MutableFixedSizeListArray::new(array, size);
-            // for each entry in the array
-            for index in 0..first.len() {
-                let values = array.mut_values();
-                for arg in arrays {
-                    let arg = arg.as_any().downcast_ref::<$ARRAY>().unwrap();
-                    if arg.is_null(index) {
-                        values.push(None);
-                    } else {
-                        values.push(Some(arg.value(index)));
-                    }
-                }
-            }
+            array.try_extend(
+                // for each entry in the array
+                (0..first.len()).map(|idx| {
+                    Some(arrays.iter().map(move |arg| {
+                        let arg = arg.as_any().downcast_ref::<$ARRAY>().unwrap();
+                        if arg.is_null(idx) {
+                            None
+                        } else {
+                            Some(arg.value(idx))
+                        }
+                    }))
+                }),
+            )?;
             Ok(array.as_arc())
         }};
     }
@@ -58,18 +59,20 @@ fn array_array(arrays: &[&dyn Array]) -> Result<ArrayRef> {
         ($OFFSET: ty) => {{
             let array = MutableUtf8Array::<$OFFSET>::with_capacity(first.len() * size);
             let mut array = MutableFixedSizeListArray::new(array, size);
-            // for each entry in the array
-            for index in 0..first.len() {
-                let values = array.mut_values();
-                for arg in arrays {
-                    let arg = arg.as_any().downcast_ref::<Utf8Array<$OFFSET>>().unwrap();
-                    if arg.is_null(index) {
-                        values.push::<&str>(None);
-                    } else {
-                        values.push(Some(arg.value(index)));
-                    }
-                }
-            }
+            array.try_extend(
+                // for each entry in the array
+                (0..first.len()).map(|idx| {
+                    Some(arrays.iter().map(move |arg| {
+                        let arg =
+                            arg.as_any().downcast_ref::<Utf8Array<$OFFSET>>().unwrap();
+                        if arg.is_null(idx) {
+                            None
+                        } else {
+                            Some(arg.value(idx))
+                        }
+                    }))
+                }),
+            )?;
             Ok(array.as_arc())
         }};
     }
@@ -78,18 +81,19 @@ fn array_array(arrays: &[&dyn Array]) -> Result<ArrayRef> {
         DataType::Boolean => {
             let array = MutableBooleanArray::with_capacity(first.len() * size);
             let mut array = MutableFixedSizeListArray::new(array, size);
-            // for each entry in the array
-            for index in 0..first.len() {
-                let values = array.mut_values();
-                for arg in arrays {
-                    let arg = arg.as_any().downcast_ref::<BooleanArray>().unwrap();
-                    if arg.is_null(index) {
-                        values.push(None);
-                    } else {
-                        values.push(Some(arg.value(index)));
-                    }
-                }
-            }
+            array.try_extend(
+                // for each entry in the array
+                (0..first.len()).map(|idx| {
+                    Some(arrays.iter().map(move |arg| {
+                        let arg = arg.as_any().downcast_ref::<BooleanArray>().unwrap();
+                        if arg.is_null(idx) {
+                            None
+                        } else {
+                            Some(arg.value(idx))
+                        }
+                    }))
+                }),
+            )?;
             Ok(array.as_arc())
         }
         DataType::UInt8 => array!(u8, PrimitiveArray<u8>, DataType::UInt8),

datafusion/src/physical_plan/expressions/cast.rs

@@ -23,7 +23,9 @@ use super::ColumnarValue;
 use crate::error::{DataFusionError, Result};
 use crate::physical_plan::PhysicalExpr;
 use crate::scalar::ScalarValue;
+use arrow::array::{Array, Int32Array};
 use arrow::compute::cast;
+use arrow::compute::take;
 use arrow::datatypes::{DataType, Schema};
 use arrow::record_batch::RecordBatch;
 
@@ -79,15 +81,38 @@ impl PhysicalExpr for CastExpr {
     }
 }
 
+fn cast_with_error(array: &dyn Array, cast_type: &DataType) -> Result<Box<dyn Array>> {
+    let result = cast::cast(array, cast_type)?;
+    if result.null_count() != array.null_count() {
+        let casted_valids = result.validity().unwrap();
+        let failed_casts = match array.validity() {
+            Some(valids) => valids ^ casted_valids,
+            None => !casted_valids,
+        };
+        let invalid_indices = failed_casts
+            .iter()
+            .enumerate()
+            .filter(|(_, failed)| *failed)
+            .map(|(idx, _)| Some(idx as i32))
+            .collect::<Vec<Option<i32>>>();
+        let invalid_values = take::take(array, &Int32Array::from(&invalid_indices))?;
+        return Err(DataFusionError::Execution(format!(
+            "Could not cast {} to value of type {}",
+            invalid_values, cast_type
+        )));
+    }
+    Ok(result)
+}
+
 /// Internal cast function for casting ColumnarValue -> ColumnarValue for cast_type
 pub fn cast_column(value: &ColumnarValue, cast_type: &DataType) -> Result<ColumnarValue> {
     match value {
         ColumnarValue::Array(array) => Ok(ColumnarValue::Array(
-            cast::cast(array.as_ref(), cast_type)?.into(),
+            cast_with_error(array.as_ref(), cast_type)?.into(),
         )),
         ColumnarValue::Scalar(scalar) => {
             let scalar_array = scalar.to_array();
-            let cast_array = cast::cast(scalar_array.as_ref(), cast_type)?.into();
+            let cast_array = cast_with_error(scalar_array.as_ref(), cast_type)?.into();
             let cast_scalar = ScalarValue::try_from_array(&cast_array, 0)?;
             Ok(ColumnarValue::Scalar(cast_scalar))
         }
@@ -243,4 +268,14 @@ mod tests {
         let result = cast(col("a", &schema).unwrap(), &schema, DataType::LargeBinary);
         result.expect_err("expected Invalid CAST");
     }
+
+    #[test]
+    fn invalid_str_cast() {
+        let arr = Utf8Array::<i32>::from_slice(&["a", "b", "123", "!", "456"]);
+        let err = cast_with_error(&arr, &DataType::Int64).unwrap_err();
+        assert_eq!(
+            err.to_string(),
+            "Execution error: Could not cast Utf8[a, b, !] to value of type Int64"
+        );
+    }
 }

datafusion/src/physical_plan/regex_expressions.rs

@@ -241,13 +241,7 @@ pub fn regexp_matches<O: Offset>(
     });
     let mut array = MutableListArray::<O, MutableUtf8Array<O>>::new();
     for items in iter {
-        if let Some(items) = items? {
-            let values = array.mut_values();
-            values.try_extend(items)?;
-            array.try_push_valid()?;
-        } else {
-            array.push_null();
-        }
+        array.try_push(items?)?;
     }
 
     Ok(array.into())

datafusion/tests/sql.rs

@@ -4326,9 +4326,20 @@ async fn test_cast_expressions_error() -> Result<()> {
     let mut ctx = create_ctx()?;
     register_aggregate_csv(&mut ctx)?;
     let sql = "SELECT CAST(c1 AS INT) FROM aggregate_test_100";
-    let actual = execute(&mut ctx, sql).await;
-    let expected = vec![vec![""]; 100];
-    assert_eq!(expected, actual);
+    let plan = ctx.create_logical_plan(sql).unwrap();
+    let plan = ctx.optimize(&plan).unwrap();
+    let plan = ctx.create_physical_plan(&plan).unwrap();
+    let result = collect(plan).await;
+
+    match result {
+        Ok(_) => panic!("expected cast error"),
+        Err(e) => {
+            assert_contains!(
+                e.to_string(),
+                "Execution error: Could not cast Utf8[c, d, b, a, b, b, e, a, d, a, d, a, e, d, b, c, e, d, d, e, e, d, a, e, c, a, c, a, a, b, e, c, e, b, a, c, d, c, c, c, b, d, d, a, e, b, b, c, a, d, b, c, d, d, b, d, e, b, a, b, c, b, c, e, e, d, e, c, d, e, e, a, a, e, a, b, e, c, e, c, a, c, b, a, a, c, a, c, c, c, b, a, a, b, d, e, e, d, b, e] to value of type Int32"
+            );
+        }
+    }
 
     Ok(())
 }