From a5314c465afcad2622e2336fc408936171070dec Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Mon, 24 Aug 2026 14:57:11 -0500 Subject: [PATCH 1/4] fix: preserve the error type of a failing parquet row filter predicate `DatafusionArrowPredicate::evaluate` must return an `ArrowError`, and it built one by `Debug`-formatting the `DataFusionError` into an `ArrowError::ComputeError`. Every failure inside a predicate pushed into the parquet decoder therefore arrived as the same untyped string, so a caller could no longer tell a user error such as a failed cast from an internal engine failure. Convert the error instead of formatting it. `From for ArrowError` leaves the original error in the source chain, so `DataFusionError::find_root` recovers the original variant, and wrapping it in a `DataFusionError::Context` first keeps the description of where the failure happened. --- .../core/tests/parquet/filter_pushdown.rs | 56 ++++++++++- .../datasource-parquet/src/row_filter.rs | 94 +++++++++++++++++-- 2 files changed, 141 insertions(+), 9 deletions(-) diff --git a/datafusion/core/tests/parquet/filter_pushdown.rs b/datafusion/core/tests/parquet/filter_pushdown.rs index dabb2f35b24b1..e526997b5d859 100644 --- a/datafusion/core/tests/parquet/filter_pushdown.rs +++ b/datafusion/core/tests/parquet/filter_pushdown.rs @@ -26,10 +26,12 @@ //! select * from data limit 10; //! ``` +use arrow::array::{ArrayRef, 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, }; @@ -37,10 +39,14 @@ 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,51 @@ 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"); + + let batch = RecordBatch::try_from_iter(vec![( + "s", + Arc::new(StringArray::from(vec!["not_an_int"])) 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(); + + // The predicate has to reach the decoder for this test to mean anything + let plan = df.clone().create_physical_plan().await.unwrap(); + let plan = displayable(plan.as_ref()).indent(false).to_string(); + assert!(!plan.contains("FilterExec"), "{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:?}" + ); +} diff --git a/datafusion/datasource-parquet/src/row_filter.rs b/datafusion/datasource-parquet/src/row_filter.rs index c1a47c896c170..6c75e1fa76636 100644 --- a/datafusion/datasource-parquet/src/row_filter.rs +++ b/datafusion/datasource-parquet/src/row_filter.rs @@ -69,7 +69,7 @@ use std::sync::Arc; use arrow::array::BooleanArray; use arrow::datatypes::{Schema, SchemaRef}; -use arrow::error::{ArrowError, Result as ArrowResult}; +use arrow::error::Result as ArrowResult; use arrow::record_batch::RecordBatch; use parquet::arrow::ProjectionMask; use parquet::arrow::arrow_reader::{ArrowPredicate, RowFilter}; @@ -161,11 +161,12 @@ impl ArrowPredicate for DatafusionArrowPredicate { timer.stop(); Ok(bool_arr) }) - .map_err(|e| { - ArrowError::ComputeError(format!( - "Error evaluating filter predicate: {e:?}" - )) - }) + // Convert rather than format the error: the decoder requires an + // `ArrowError`, and formatting collapses every failure into one + // untyped `ComputeError`. `From for ArrowError` + // leaves the original error in the source chain, so callers can + // still recover it (for example with `DataFusionError::find_root`). + .map_err(|e| e.context("Error evaluating filter predicate").into()) } } @@ -527,13 +528,14 @@ impl<'a> RowFilterGenerator<'a> { mod test { use super::*; use arrow::datatypes::{DataType, Fields}; - use datafusion_common::ScalarValue; + use arrow::error::ArrowError; + 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 +676,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( From 7f66b65fccf30a731895bb76d53be1066f529f2e Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Mon, 24 Aug 2026 19:19:49 -0500 Subject: [PATCH 2/4] review: trim the comment on the row filter error conversion Keep the part that is not obvious from the code (converting preserves the source chain) and drop the restatement of what the old code did. Co-Authored-By: Claude Opus 5 --- datafusion/datasource-parquet/src/row_filter.rs | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/datafusion/datasource-parquet/src/row_filter.rs b/datafusion/datasource-parquet/src/row_filter.rs index 6c75e1fa76636..552648b5a0c31 100644 --- a/datafusion/datasource-parquet/src/row_filter.rs +++ b/datafusion/datasource-parquet/src/row_filter.rs @@ -161,11 +161,9 @@ impl ArrowPredicate for DatafusionArrowPredicate { timer.stop(); Ok(bool_arr) }) - // Convert rather than format the error: the decoder requires an - // `ArrowError`, and formatting collapses every failure into one - // untyped `ComputeError`. `From for ArrowError` - // leaves the original error in the source chain, so callers can - // still recover it (for example with `DataFusionError::find_root`). + // Convert rather than format: converting leaves the original error + // in the source chain, so callers can still recover it (for example + // with `DataFusionError::find_root`) .map_err(|e| e.context("Error evaluating filter predicate").into()) } } From 7d751390066a72826f77d3692ff6b5bb53d62b9a Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Tue, 25 Aug 2026 08:24:43 -0500 Subject: [PATCH 3/4] review: address feedback on the row filter error test and comment Give the end-to-end test a column the predicate does not reference, so the projection always has a non-filter column and a narrow-projection pushdown heuristic has no reason to decline the scan, and say in the assertion why a FilterExec would invalidate the test. Also spell out on the `map_err` why the direct conversion is not equivalent: a bare `ArrowError` has no source, so after the decoder re-wraps it as `ParquetError::External` there is no `DataFusionError` left in the chain for `find_root` to recover. Co-Authored-By: Claude Opus 5 --- .../core/tests/parquet/filter_pushdown.rs | 23 +++++++++++++------ .../datasource-parquet/src/row_filter.rs | 10 +++++--- 2 files changed, 23 insertions(+), 10 deletions(-) diff --git a/datafusion/core/tests/parquet/filter_pushdown.rs b/datafusion/core/tests/parquet/filter_pushdown.rs index e526997b5d859..25952f8193d67 100644 --- a/datafusion/core/tests/parquet/filter_pushdown.rs +++ b/datafusion/core/tests/parquet/filter_pushdown.rs @@ -26,7 +26,7 @@ //! select * from data limit 10; //! ``` -use arrow::array::{ArrayRef, StringArray}; +use arrow::array::{ArrayRef, Int32Array, StringArray}; use arrow::compute::concat_batches; use arrow::error::ArrowError; use arrow::record_batch::RecordBatch; @@ -761,10 +761,16 @@ 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"); - let batch = RecordBatch::try_from_iter(vec![( - "s", - Arc::new(StringArray::from(vec!["not_an_int"])) as ArrayRef, - )]) + // `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(); @@ -784,10 +790,13 @@ async fn pushed_down_predicate_reports_the_original_error() { .await .unwrap(); - // The predicate has to reach the decoder for this test to mean anything let plan = df.clone().create_physical_plan().await.unwrap(); let plan = displayable(plan.as_ref()).indent(false).to_string(); - assert!(!plan.contains("FilterExec"), "{plan}"); + 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(); diff --git a/datafusion/datasource-parquet/src/row_filter.rs b/datafusion/datasource-parquet/src/row_filter.rs index 552648b5a0c31..782914b7bbdc0 100644 --- a/datafusion/datasource-parquet/src/row_filter.rs +++ b/datafusion/datasource-parquet/src/row_filter.rs @@ -161,9 +161,13 @@ impl ArrowPredicate for DatafusionArrowPredicate { timer.stop(); Ok(bool_arr) }) - // Convert rather than format: converting leaves the original error - // in the source chain, so callers can still recover it (for example - // with `DataFusionError::find_root`) + // Convert rather than format, and keep the context: a plain + // conversion of a `DataFusionError::ArrowError` yields a bare + // `ArrowError`, which carries only a `String` and no source, so once + // the decoder re-wraps it as `ParquetError::External` nothing in the + // chain is a `DataFusionError` any more. Wrapping in a context first + // routes it to `ArrowError::ExternalError`, which keeps the original + // error recoverable (for example with `DataFusionError::find_root`). .map_err(|e| e.context("Error evaluating filter predicate").into()) } } From ff38dfbb7fcc3aa0b1f6a34b78cd1592befd5024 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Tue, 25 Aug 2026 08:34:29 -0500 Subject: [PATCH 4/4] review: construct the ArrowError explicitly instead of via a conversion Relying on `.context()` to route past the `DataFusionError::ArrowError` arm of `From for ArrowError` produced the shape we want as a side effect, which took a paragraph of comment to justify. Name the variant instead: `ExternalError` is the only one that keeps a source. Co-Authored-By: Claude Opus 5 --- .../datasource-parquet/src/row_filter.rs | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/datafusion/datasource-parquet/src/row_filter.rs b/datafusion/datasource-parquet/src/row_filter.rs index 782914b7bbdc0..245e00c46a07a 100644 --- a/datafusion/datasource-parquet/src/row_filter.rs +++ b/datafusion/datasource-parquet/src/row_filter.rs @@ -69,7 +69,7 @@ use std::sync::Arc; use arrow::array::BooleanArray; use arrow::datatypes::{Schema, SchemaRef}; -use arrow::error::Result as ArrowResult; +use arrow::error::{ArrowError, Result as ArrowResult}; use arrow::record_batch::RecordBatch; use parquet::arrow::ProjectionMask; use parquet::arrow::arrow_reader::{ArrowPredicate, RowFilter}; @@ -161,14 +161,14 @@ impl ArrowPredicate for DatafusionArrowPredicate { timer.stop(); Ok(bool_arr) }) - // Convert rather than format, and keep the context: a plain - // conversion of a `DataFusionError::ArrowError` yields a bare - // `ArrowError`, which carries only a `String` and no source, so once - // the decoder re-wraps it as `ParquetError::External` nothing in the - // chain is a `DataFusionError` any more. Wrapping in a context first - // routes it to `ArrowError::ExternalError`, which keeps the original - // error recoverable (for example with `DataFusionError::find_root`). - .map_err(|e| e.context("Error evaluating filter predicate").into()) + // `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::ExternalError(Box::new( + e.context("Error evaluating filter predicate"), + )) + }) } } @@ -530,7 +530,6 @@ impl<'a> RowFilterGenerator<'a> { mod test { use super::*; use arrow::datatypes::{DataType, Fields}; - use arrow::error::ArrowError; use datafusion_common::{DataFusionError, ScalarValue}; use arrow::array::{