fix: preserve the error type of a failing parquet row filter predicate

datafusion/core/tests/parquet/filter_pushdown.rs

@@ -26,21 +26,27 @@
 //! select * from data limit 10;
 //! ```
 
+use arrow::array::{ArrayRef, Int32Array, StringArray};
 use arrow::compute::concat_batches;
+use arrow::error::ArrowError;
 use arrow::record_batch::RecordBatch;
-use datafusion::physical_plan::collect;
 use datafusion::physical_plan::metrics::{MetricValue, MetricsSet};
+use datafusion::physical_plan::{collect, displayable};
 use datafusion::prelude::{
     Expr, ParquetReadOptions, SessionContext, col, lit, lit_timestamp_nano,
 };
 use datafusion::test_util::parquet::{ParquetScanOptions, TestParquetFile};
 use datafusion_expr::utils::{conjunction, disjunction, split_conjunction};
 use std::path::Path;
 
+use datafusion_common::DataFusionError;
 use datafusion_common::test_util::parquet_test_data;
 use datafusion_execution::config::SessionConfig;
 use itertools::Itertools;
+use parquet::arrow::ArrowWriter;
 use parquet::file::properties::WriterProperties;
+use std::fs::File;
+use std::sync::Arc;
 use tempfile::TempDir;
 
 /// how many rows of generated data to write to our parquet file (arbitrary)
@@ -746,3 +752,60 @@ impl PredicateCacheTest {
         Ok(())
     }
 }
+
+/// A predicate that is pushed into the parquet decoder and then fails while it
+/// is being evaluated must report the original error, so that callers can still
+/// tell a user error apart from an internal one.
+#[tokio::test]
+async fn pushed_down_predicate_reports_the_original_error() {
+    let tempdir = TempDir::new_in(Path::new(".")).unwrap();
+    let path = tempdir.path().join("cast_error.parquet");
+
+    // `v` is never referenced by the predicate, so the projection always has a
+    // column that only the scan can supply and a narrow-projection pushdown
+    // heuristic has no reason to decline this scan
+    let batch = RecordBatch::try_from_iter(vec![
+        (
+            "s",
+            Arc::new(StringArray::from(vec!["not_an_int"])) as ArrayRef,
+        ),
+        ("v", Arc::new(Int32Array::from(vec![1])) as ArrayRef),
+    ])
+    .unwrap();
+    let mut writer =
+        ArrowWriter::try_new(File::create(&path).unwrap(), batch.schema(), None).unwrap();
+    writer.write(&batch).unwrap();
+    writer.close().unwrap();
+
+    let mut config = SessionConfig::new();
+    config.options_mut().execution.parquet.pushdown_filters = true;
+    let ctx = SessionContext::new_with_config(config);
+    ctx.register_parquet("t", path.to_str().unwrap(), ParquetReadOptions::default())
+        .await
+        .unwrap();
+
+    // Casting the column in the file to `Int32` fails on this data
+    let df = ctx
+        .sql("SELECT * FROM t WHERE CAST(s AS INT) = 1")
+        .await
+        .unwrap();
+
+    let plan = df.clone().create_physical_plan().await.unwrap();
+    let plan = displayable(plan.as_ref()).indent(false).to_string();
+    assert!(
+        !plan.contains("FilterExec"),
+        "the predicate has to reach the decoder for this test to mean anything, \
+         but a FilterExec here means pushdown was declined:\n{plan}"
+    );
+
+    let err = df.collect().await.unwrap_err();
+    let root = err.find_root();
+    assert!(
+        matches!(
+            root,
+            DataFusionError::ArrowError(inner, _)
+                if matches!(inner.as_ref(), ArrowError::CastError(_))
+        ),
+        "expected the original cast error, got {root:?}"
+    );
+}

datafusion/datasource-parquet/src/row_filter.rs

@@ -161,9 +161,12 @@ impl ArrowPredicate for DatafusionArrowPredicate {
                 timer.stop();
                 Ok(bool_arr)
             })
+            // `ExternalError` is the only `ArrowError` variant that keeps a
+            // source, and therefore the only one that leaves the original error
+            // recoverable (see `DataFusionError::find_root`)
             .map_err(|e| {
-                ArrowError::ComputeError(format!(
-                    "Error evaluating filter predicate: {e:?}"
+                ArrowError::ExternalError(Box::new(
+                    e.context("Error evaluating filter predicate"),
                 ))
             })
     }
@@ -527,13 +530,13 @@ impl<'a> RowFilterGenerator<'a> {
 mod test {
     use super::*;
     use arrow::datatypes::{DataType, Fields};
-    use datafusion_common::ScalarValue;
+    use datafusion_common::{DataFusionError, ScalarValue};
 
     use arrow::array::{
         Int32Array, ListBuilder, StringArray, StringBuilder, StructArray,
     };
     use arrow::datatypes::{Field, TimeUnit::Nanosecond};
-    use datafusion_expr::{Expr, col};
+    use datafusion_expr::{Cast, Expr, col, lit};
     use datafusion_functions::core::get_field;
     use datafusion_functions_nested::array_has::{
         array_has_all_udf, array_has_any_udf, array_has_udf,
@@ -674,6 +677,82 @@ mod test {
         assert!(matches!(filtered, Ok(a) if a == BooleanArray::from(vec![true; 8])));
     }
 
+    /// A predicate that fails while it is being evaluated must report the
+    /// original error, not an opaque string, so that callers can still tell a
+    /// user error apart from an internal one.
+    #[test]
+    fn evaluate_reports_the_original_error() {
+        let schema = Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8, false)]));
+        let batch = RecordBatch::try_new(
+            Arc::clone(&schema),
+            vec![Arc::new(StringArray::from(vec!["not_an_int"]))],
+        )
+        .expect("record batch");
+
+        let file = NamedTempFile::new().expect("temp file");
+        let mut writer =
+            ArrowWriter::try_new(file.reopen().unwrap(), Arc::clone(&schema), None)
+                .expect("writer");
+        writer.write(&batch).expect("write batch");
+        writer.close().expect("close writer");
+
+        let parquet_reader_builder =
+            ParquetRecordBatchReaderBuilder::try_new(file.reopen().expect("reopen file"))
+                .expect("reader builder");
+        let metadata = parquet_reader_builder.metadata().clone();
+        let file_schema = parquet_reader_builder.schema().clone();
+
+        // Casting the column in the file to `Int32` fails on this data
+        let expr = Expr::Cast(Cast::new(Box::new(col("s")), DataType::Int32)).eq(lit(1));
+        let expr = logical2physical(&expr, &file_schema);
+        let candidate = FilterCandidateBuilder::new(expr, Arc::clone(&file_schema))
+            .build(&metadata)
+            .expect("building candidate")
+            .expect("candidate expected");
+
+        let mut predicate = DatafusionArrowPredicate::try_new(
+            candidate,
+            Count::new(),
+            Count::new(),
+            Time::new(),
+        )
+        .expect("creating filter predicate");
+
+        let mut parquet_reader = parquet_reader_builder
+            .with_projection(predicate.projection().clone())
+            .build()
+            .expect("building reader");
+        let first_rb = parquet_reader
+            .next()
+            .expect("expected record batch")
+            .expect("expected error free record batch");
+
+        let err = predicate
+            .evaluate(first_rb)
+            .expect_err("evaluating the predicate should fail");
+
+        // The cast failure is still reachable, rather than being flattened into
+        // an untyped `ArrowError::ComputeError`
+        let err = DataFusionError::from(err);
+        let root = err.find_root();
+        assert!(
+            matches!(
+                root,
+                DataFusionError::ArrowError(inner, _)
+                    if matches!(inner.as_ref(), ArrowError::CastError(_))
+            ),
+            "expected the original cast error, got {root:?}"
+        );
+
+        // and the message still says where the failure happened
+        let message = err.to_string();
+        assert!(
+            message.contains("Error evaluating filter predicate"),
+            "{message}"
+        );
+        assert!(message.contains("Cannot cast string"), "{message}");
+    }
+
     #[test]
     fn struct_data_structures_prevent_pushdown() {
         let table_schema = Arc::new(Schema::new(vec![Field::new(