diff --git a/datafusion/core/tests/parquet/filter_pushdown.rs b/datafusion/core/tests/parquet/filter_pushdown.rs index dabb2f35b24b1..25952f8193d67 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, 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, }; @@ -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,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:?}" + ); +} diff --git a/datafusion/datasource-parquet/src/row_filter.rs b/datafusion/datasource-parquet/src/row_filter.rs index c1a47c896c170..245e00c46a07a 100644 --- a/datafusion/datasource-parquet/src/row_filter.rs +++ b/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(