Revert use file schema in parquet pruning by adriangb · Pull Request #16086 · apache/datafusion

adriangb · 2025-05-19T04:31:06Z

FixesParquet predicate filters fail with "Invalid comparison operation: Utf8View <= Utf8" #15920

adriangb · 2025-05-19T04:32:00Z

+ // TODO: this is failing on main, and has been for a long time!
+ // See <comment on PR>
+ // // Predicate should prune no row groups
+ // let filter = col("c1").eq(lit(ScalarValue::UInt64(Some(1))));
+ // let read = RoundTrip::new()
+ // .with_predicate(filter)
+ // .with_schema(schema)
+ // .round_trip_to_batches(vec![batch])
+ // .await
+ // .unwrap();
+ // assert_eq!(read.len(), 1);


This has been failing at least as far back as v46.0.0 with this diff:
diff --git a/datafusion/core/src/datasource/file_format/parquet.rs b/datafusion/core/src/datasource/file_format/parquet.rs index 3b71593b3..60b403aff 100644 --- a/datafusion/core/src/datasource/file_format/parquet.rs+++ b/datafusion/core/src/datasource/file_format/parquet.rs@@ -67,13 +67,14 @@ pub(crate) mod test_util { .into_iter() .zip(tmp_files.into_iter()) .map(|(batch, mut output)| { - let builder = parquet::file::properties::WriterProperties::builder();- let props = if multi_page {+ let mut builder = parquet::file::properties::WriterProperties::builder();+ builder = if multi_page { builder.set_data_page_row_count_limit(ROWS_PER_PAGE) } else { builder - }- .build();+ };+ builder = builder.set_bloom_filter_enabled(true);+ let props = builder.build(); let mut writer = parquet::arrow::ArrowWriter::try_new( &mut output, diff --git a/datafusion/core/src/datasource/physical_plan/parquet.rs b/datafusion/core/src/datasource/physical_plan/parquet.rs index 888f3ad9e..240d84783 100644 --- a/datafusion/core/src/datasource/physical_plan/parquet.rs+++ b/datafusion/core/src/datasource/physical_plan/parquet.rs@@ -489,6 +489,37 @@ mod tests { assert_eq!(read.len(), 0); } ++ #[tokio::test]+ async fn evolved_schema_column_type_filter_ints() {+ // The table and filter have a common data type, but the file schema differs+ let c1: ArrayRef = Arc::new(Int8Array::from(vec![Some(1), Some(2)]));+ let batch = create_batch(vec![("c1", c1.clone())]);++ let schema =+ Arc::new(Schema::new(vec![Field::new("c1", DataType::UInt64, false)]));++ // // Predicate should prune all row groups+ // let filter = col("c1").eq(lit(ScalarValue::UInt64(Some(5))));+ // let read = RoundTrip::new()+ // .with_predicate(filter)+ // .with_schema(schema.clone())+ // .round_trip_to_batches(vec![batch.clone()])+ // .await+ // .unwrap();+ // assert_eq!(read.len(), 0);++ // Predicate should prune no row groups+ let filter = col("c1").eq(lit(ScalarValue::UInt64(Some(1))));+ let read = RoundTrip::new()+ .with_predicate(filter)+ .with_schema(schema)+ .round_trip_to_batches(vec![batch])+ .await+ .unwrap();+ assert_eq!(read.len(), 1);+ }+ #[tokio::test] async fn evolved_schema_disjoint_schema_filter() { let c1: ArrayRef =

xudong963 · 2025-05-19T14:32:46Z

Thank you @adriangb , I'll review the PR tomorrow carefully and try to verify if this is related to the issue we encountered during upgrading DF47

adriangb · 2025-05-19T14:32:00Z

+ } else {
+ source = source.with_pushdown_filters(false);
 }


This is just to be explicit

adriangb · 2025-05-19T14:32:59Z

+ if self.bloom_filters {
+ source = source.with_bloom_filter_on_read(true);
+ } else {
+ source = source.with_bloom_filter_on_read(false);
 }


Otherwise bloom filters will pick up the slack and do pruning even if the types mismatch and stats / page / row pruning fails (bloom filters use &str for both Utf8 and Utf8View so they don't care). Only one of the tests actually relies on bloom filters.

adriangb · 2025-05-19T14:34:00Z

Thank you @adriangb , I'll review the PR tomorrow carefully and try to verify if this is related to the issue we encountered during upgrading DF47

Thank you 🙏🏻 and sorry for any inconvenience this has caused!

alamb

Thank you @adriangb and @xudong963

alamb · 2025-05-20T13:51:13Z

 pub predicate: Option<Arc<dyn PhysicalExpr>>,
- /// Schema of the output table
- pub table_schema: SchemaRef,
+ /// Schema of the output table without partition columns.


I verified that when the code changes are reverted this test fails:
cargo test --all-features -p datafusion -- parquet ... ---- datasource::physical_plan::parquet::tests::evolved_schema_column_type_filter_ints stdout ---- thread 'datasource::physical_plan::parquet::tests::evolved_schema_column_type_filter_ints' panicked at datafusion/core/src/datasource/physical_plan/parquet.rs:927:9: assertion `left == right` failed left: 1 right: 0 note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace ---- datasource::physical_plan::parquet::tests::evolved_schema_column_type_filter_strings stdout ---- thread 'datasource::physical_plan::parquet::tests::evolved_schema_column_type_filter_strings' panicked at datafusion/core/src/datasource/physical_plan/parquet.rs:885:9: assertion `left == right` failed left: 1 right: 0 failures: datasource::physical_plan::parquet::tests::evolved_schema_column_type_filter_ints datasource::physical_plan::parquet::tests::evolved_schema_column_type_filter_strings

~~I think they're related to the PR: #16133~~
Sorry, I misread @alamb 's message

alamb · 2025-05-20T13:52:27Z

 let (pruning_predicate, page_pruning_predicate) = build_pruning_predicates(
 predicate.as_ref(),
- &physical_file_schema,
+ &logical_file_schema,


This is the actual change (from physical to logical schema) -- I am calling this out because it took me a while to spot it (at first I thought this was only a name change)

Yes sorry for also changing the names in this PR - it was just too confusing to even work on this PR without giving things a new name. Let me know if you have any suggestions for better names, happy to accept them, naming here seems to be 1/2 of the problem. You know what they say, there's 3 hard problems...

no worries -- I think the new names are much clearer to me

Sorry, coming to this too late, but I'm curious why the logical schema? To me pruning implies a physical layer kind of thing. I ask because now I'm having issues in #15821 trying to build pruning predicates for columns that don't exist in the physical schema.
This may just mean #15821 is operating at too low a level and needs to operate up here where we a) know it's parquet and b) have the physical schema.
@alamb @adriangb

I would like it to operate at the physical layer and maybe we can get back there through future work.
My immediate goal merging this PR was to fix the pretty bad regression I had introduced. A step backwards to avoid running into a wall.
But I also am not sure that we can even make it operate at the physical layer properly: there's a lot of logic that goes into determining what casts to do and such that happens at the logical layer. We'd have to re-create all of that at the physical layer because if the types of the columns change everything has to be re-evaluated.

I think the key observation is that the predicates are expressed in terms of the table schema (not the file schema).
Today there is a bunch of logic to translate table schema -> file schema for data and statistics.
So in order to evaluate predicates on the physical schema we would need to translate the expression as well, which I think is what @adriangb is saying above

Yep precisely. And that requires re evaluating cast rules, etc which is... not easy.

Ok, thanks for the explanation!

FWIW I think this ticket tracks the issue more fully:
Evaluate filter pushdown against the physical schema for performance and correctness #15780

xudong963 · 2025-05-21T10:38:19Z

+ let schema = Arc::new(Schema::new(vec![Field::new("c1", DataType::Utf8, false)]));
+
+ // Predicate should prune all row groups
+ let filter = col("c1").eq(lit(ScalarValue::Utf8(Some("aaa".to_string()))));


I think the filter should be constructed by Utf8View, it's a logical expr.
let filter = col("c1").eq(lit(ScalarValue::Utf8View(Some("aaa".to_string()))));
Then, the test will fail because https://github.com/apache/datafusion/pull/16086/files#diff-c8eebe4fb9ee7662170c6ce507ad2be92c0229b3a91a2f4431a9bfc7185a0eb9L148 uses file_schema to transfer logical expr to physical expr. And #16133 will fix it

Upon more reflection, I think the predicate needs to be created using the logical schema today
@adriangb @etseidl and myself are discussing potentially changing that here:
Evaluate filter pushdown against the physical schema for performance and correctness #15780

xudong963

Thank you @adriangb

Your test evolved_schema_column_type_filter_strings is close to my real case, my real case is that file_schema is view but table_schame is utf8.

I constructed a similar test that matches my cases:

let c1:ArrayRef =
Arc::new(StringArray::from(vec![Some("foo"),Some("bar")]));let batch = create_batch(vec![("c1", c1.clone())]);let schema = Arc::new(Schema::new(vec![Field::new("c1",DataType::Utf8View,false)]));// Predicate should prune all row groupslet filter = col("c1").eq(lit(ScalarValue::Utf8(Some("aaa".to_string()))));

With the fix: #16133 (Why with the fix? The above test I posted is using table schema to build filter), I tested the test in branch-46, branch-47, main and your branch:
branch-46: ✅
branch-47: ❌
main: ❌
your branch: ✅

alamb · 2025-05-21T12:52:33Z

Thanks again @xudong963 and @adriangb

xudong963 · 2025-06-11T02:35:50Z

Fyi, we've upgraded to DF47 with the fix successfully.

adriangb · 2025-06-11T02:37:33Z

Great to hear! Sorry for the inconvenience...

xudong963 · 2025-06-11T02:40:02Z

Great to hear! Sorry for the inconvenience...

No problem!

alamb · 2025-06-11T04:02:34Z

FWIW @phillipleblanc also hit this as well in SpiceAI. See this for more details

Fix regression in Parquet pushdown spiceai/spiceai#6178

github-actionsBot added optimizer Optimizer rules core Core DataFusion crate datasource Changes to the datasource crate labels May 19, 2025

adriangb commented May 19, 2025
View reviewed changes

Comment threaddatafusion/core/src/datasource/physical_plan/parquet.rs Outdated

xudong963 self-requested a review May 19, 2025 06:26

github-actionsBot removed the optimizer Optimizer rules label May 19, 2025

adriangb commented May 19, 2025
View reviewed changes

adriangb added 6 commits May 19, 2025 13:38

wip
3e58165

comment
7633ad5

Update datafusion/core/src/datasource/physical_plan/parquet.rs
cb5a6b2

remove prints
b52ee28

better test
8a9fa27

fmt
455582b

adriangb force-pushed the revert-use-file-schema branch from 348760b to 455582bCompare May 19, 2025 20:38

alamb approved these changes May 20, 2025
View reviewed changes

xudong963 mentioned this pull request May 21, 2025
Fix parquet test using wrong schema #16133
Closed

xudong963 reviewed May 21, 2025
View reviewed changes

xudong963 approved these changes May 21, 2025
View reviewed changes

alamb merged commit ca55f1c into apache:mainMay 21, 2025

adriangb mentioned this pull request May 21, 2025
Evaluate filter pushdown against the physical schema for performance and correctness #15780
Closed

alamb mentioned this pull request May 22, 2025
Clarify docs and names in parquet predicate pushdown tests #16155
Merged

phillipleblanc mentioned this pull request Jun 10, 2025
Fix regression in Parquet pushdown spiceai/spiceai#6178
Merged

Jeadie mentioned this pull request Sep 25, 2025
Enhancement: DataFusion v49 Upgrade spiceai/spiceai#7078
Closed
28 tasks

Uh oh!

Conversation

adriangb commented May 19, 2025• edited by alamb Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

adriangbMay 19, 2025• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

xudong963 commented May 19, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adriangbMay 19, 2025• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adriangb commented May 19, 2025

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xudong963May 21, 2025• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xudong963 left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alamb commented May 21, 2025

Uh oh!

xudong963 commented Jun 11, 2025

Uh oh!

adriangb commented Jun 11, 2025

Uh oh!

xudong963 commented Jun 11, 2025

Uh oh!

alamb commented Jun 11, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

adriangb commented May 19, 2025•
edited by alamb
Loading

adriangbMay 19, 2025•
edited
Loading

adriangbMay 19, 2025•
edited
Loading

xudong963May 21, 2025•
edited
Loading

xudong963 left a comment •
edited
Loading