fix: preserve the error type of a failing parquet row filter predicate (#24638)

## Which issue does this PR close?

No existing issue tracks this; happy to file one if the project prefers
that first.

## Rationale for this change

`ArrowPredicate::evaluate` must return an `ArrowError`, so
`DatafusionArrowPredicate::evaluate` built one by `Debug`-formatting the
`DataFusionError` it received:

```rust
.map_err(|e| {
    ArrowError::ComputeError(format!("Error evaluating filter predicate: {e:?}"))
})
```

Formatting the error discards its type. Every failure inside a predicate
pushed
into the parquet decoder reaches the caller as the same untyped
`ArrowError::ComputeError` carrying a `Debug` string, so a user error
such as a
failed cast is indistinguishable from an internal engine failure. Any
embedder
that classifies errors by variant, for example to decide whether a query
failed
because of the input or because of a bug, cannot do so for this path. It
also
reads badly, because the nested error is rendered with `Debug` rather
than
`Display`.

Reproduction with `datafusion-cli`:

```sql
COPY (SELECT 'not_an_int' AS s) TO 't.parquet' STORED AS PARQUET;
CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION 't.parquet';
SET datafusion.execution.parquet.pushdown_filters = true;
SELECT * FROM t WHERE CAST(s AS INT) = 1;
```

Before:

```
Error: Parquet error: External: Compute error: Error evaluating filter predicate: ArrowError(CastError("Cannot cast string 'not_an_int' to value of Int32 type"), Some(""))
```

After:

```
Error: Parquet error: External: External error: Error evaluating filter predicate
caused by
Arrow error: Cast error: Cannot cast string 'not_an_int' to value of Int32 type
```

## What changes are included in this PR?

`DatafusionArrowPredicate::evaluate` now converts the error instead of
formatting it:

```rust
.map_err(|e| e.context("Error evaluating filter predicate").into())
```

`From<DataFusionError> for ArrowError` is the conversion DataFusion
already
documents for this boundary. It leaves the original error in the
`Error::source`
chain, and the parquet decoder propagates it as
`ParquetError::External`, which
is also source preserving, so `DataFusionError::find_root` recovers the
original
variant at the top of the stack. Wrapping the error in a
`DataFusionError::Context` first keeps the description of where the
failure
happened, which the old string also carried.

One consequence worth calling out: because the context has to live
somewhere,
the returned `ArrowError` variant is `ExternalError` rather than the
original
Arrow variant. Callers that want the type use `find_root` (or walk
`Error::source`), which is the existing way to recover an error across
an
`ArrowError` boundary in DataFusion and is used the same way in
`datafusion/common/src/scalar/mod.rs` and `datafusion/physical-plan`.
Dropping
the context would yield a bare `ArrowError::CastError` here, at the cost
of no
longer saying which stage failed.

## Are these changes tested?

Yes. Two new tests, both of which fail without the change:

* `datafusion/datasource-parquet/src/row_filter.rs`:
`evaluate_reports_the_original_error` evaluates a predicate that fails
to cast
and asserts `find_root` returns the `CastError`, and that the message
still
  names the predicate. Without the change it observes
`ArrowError(ComputeError("Error evaluating filter predicate:
ArrowError(CastError(...))"))`.
* `datafusion/core/tests/parquet/filter_pushdown.rs`:
`pushed_down_predicate_reports_the_original_error` runs the same failure
through a full parquet scan. It first asserts the predicate really is
pushed
  into the scan and not left in a `FilterExec`, so the test cannot pass
  vacuously, then asserts the same about the error the query returns.

Existing suites run locally: `cargo test -p
datafusion-datasource-parquet`,
`cargo test -p datafusion --test parquet_integration`, and the full
`sqllogictest` suite, all passing. No test expectation elsewhere
depended on the
old string.

## Are there any user-facing changes?

The text of the error raised when a pushed down parquet predicate fails
changes,
as shown above. There is no public API change.

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

Author: adriangb

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

@@ -166,9 +166,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"),
                 ))
             })
     }
@@ -545,13 +548,13 @@ pub(crate) fn row_filter_from_prebuilt(
 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,
@@ -692,6 +695,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(