Uh oh!
There was an error while loading. Please reload this page.
Cut Parquet over to PhysicalExprAdapter, remove SchemaAdapter - #18998
Conversation
| // Test using a custom schema adapter and no explicit physical expr adapter | ||
| // This should use the custom schema adapter both for projections and predicate pushdown |
There was a problem hiding this comment.
Removing this because it's no longer true. SchemaAdapter is not used for predicate pushdown.
| #[tokio::test] | ||
| async fn test_custom_schema_adapter_no_rewriter() { | ||
| // Make a hardcoded schema adapter that adds a new column "b" with default value 0.0 | ||
| // and converts the first column "a" from Int32 to UInt64. |
There was a problem hiding this comment.
Same here I think this test is no longer relevant
| #[test] | ||
| fn test_cast_timestamp_with_timezone_to_timestamp() -> Result<()> { |
There was a problem hiding this comment.
These ended up not being necessary (the bug was elsewhere in my changes) but I think it's good coverage anyway so leaving it in here.
There was a problem hiding this comment.
These seem like they could be just as easily tested with .slt / arrow_cast calls 🤔
That would be less verbose and easier to maintain in my opinion
0704aed to
da0ecf9Compareadriangb
commented
Nov 30, 2025
run benchmarks |
alamb
commented
Nov 30, 2025
BTW your job is scheduled. My job runner got borked due to something silly (alamb/datafusion-benchmarking@213b478). The run should complete in a while |
alamb
commented
Nov 30, 2025
🤖 |
There was a problem hiding this comment.
The comment predates this PR but I agree it's wrong. I've updated it.
| .await; | ||
| assert_contains!(read.unwrap_err().to_string(), | ||
| "Cannot cast file schema field c3 of type Date64 to table schema field of type Int8"); | ||
| "Cannot cast column 'c3' from 'Date64' (physical data type) to 'Int8' (logical data type)"); |
There was a problem hiding this comment.
This looks the opposite to me - Int should be the physical type and Date is the logical one
There was a problem hiding this comment.
I checked and I think this is right.
From above:
let batch2 = create_batch(vec![("c3", c4),("c2", c2),("c1", c1)]);And:
let c4:ArrayRef = Arc::new(Date64Array::from(vec![Some(86400000),None,Some(259200000),]));But in the schema:
Field::new("c3",DataType::Int8,true)So in the physical data c3 has the data type Date64 and the logical type is Int8
alamb
commented
Nov 30, 2025
🤖: Benchmark completed Details |
| let projection_into_file_schema: Vec<usize> = projected_table_schema | ||
| .fields() | ||
| .iter() | ||
| .filter_map(|f| self.file_schema.index_of(f.name()).ok()) | ||
| .collect(); |
There was a problem hiding this comment.
This assumes predicate column indices refer to the table/logical schema (not file schema), and that we always map those logical columns into file indices by name.
That's true today because DefaultPhysicalExprAdapter rewrites expressions but preserves logical column indices and we do name-based mapping here.
Do you think adding:
a regression test for the reorder case (table: (a,b,c), file: (b,c,a)), ensuring both pruning and row-filtering work
an explicit assert and comment in [FilterCandidateBuilder::build] verifying that each required logical idx resolves to a unique file index or is intentionally missing and handled by adapter.
would prevent regression?
There was a problem hiding this comment.
I think (1) exists but I will confirm or add it. I will also add (2).
There was a problem hiding this comment.
66df4f2 are excellent projection tests for:
flipped order (c,b,a) vs (a,b,c), and
missing column (a,c) vs (a,b,c)
The tests for pruning / row filter path are still missing.
There was a problem hiding this comment.
4f00a2f
validates row-level predicate evaluation inside the scan (which goes through row_filter.rs).
I think we still need tests to validate that row-group pruning (via min/max stats) works under reordered schemas.
There was a problem hiding this comment.
By pruning you mean statistics / zone map w/o row wise predicate pushdown?
There was a problem hiding this comment.
3cd7c84 is solid regression shield for both filter and pruning paths!
091cbbd to
4f00a2fCompare73f82a5 to
c2eb2dcCompare754f54c to
3cd7c84Compare| None, | ||
| )); | ||
| Ok((left, op, right)) | ||
| } else if let Some(cast_col) = column_expr_any.downcast_ref::<CastColumnExpr>() { |
There was a problem hiding this comment.
This is a good example where the code for CastExpr and CastColumnExpr are almost identical and the duplication seems unfortunate
I took an initial shot at removing the redundancy here
| ]); | ||
| // batch2: c3(int8), c2(int64), c1(string), c4(string) | ||
| // batch2: c3(date64), c2(int64), c1(string), c4(date64) |
There was a problem hiding this comment.
🤔 batch2 only has 3 columns (c3(date64), c2(int64) and c1(string))
| // batch2: c3(date64), c2(int64), c1(string), c4(date64) | |
| // batch2: c3(date64), c2(int64), c1(string) |
| } | ||
| } | ||
| // Implement a custom PhysicalExprAdapterFactory that fills in missing columns with the default value for the field type |
There was a problem hiding this comment.
it might help to note here that "default value for the field type" really means:
1for missing Int64 values'b'for missing Utf8 columns
| ctx.deregister_table("t").unwrap(); | ||
| ctx.register_table("t", Arc::new(table)).unwrap(); | ||
| let batches = ctx | ||
| .sql("SELECT c2, c1 FROM t WHERE c1 = 2 AND c2 = 'a'") |
There was a problem hiding this comment.
it took me a while to understand why this test changed -- it is because CustomSchemaMapper used to fill in the value a but CustomPhysicalExprAdapterFactory fills in b.
Therefore this change makes sense to me 👍
| .await | ||
| .unwrap(); | ||
| let expected = [ |
There was a problem hiding this comment.
minor it would be nice to have this be formatted (maybe ignore rustfmt)?
| .zip(partitioned_file.partition_values.clone()) | ||
| .collect_vec(), | ||
| ); | ||
| let projection_expressions = projection |
There was a problem hiding this comment.
I found some of this somewhat hard to follow, maybe as a follow on some of it could move to helpers or methods on ProjectionExprs -- for example maybe a function like
implProjectionExprs{// Update this ProjectionExpr by applying a closure to all the contained expressionsfnmap<F>(self,f:F) -> Result<Self>{ ...}}| .collect::<Result<Vec<_>>>()?; | ||
| projection = ProjectionExprs::new(projection_expressions); | ||
| } | ||
| let indices = projection |
There was a problem hiding this comment.
this might be a nice function on ProjectionExprs (the inverse of ProjectionExprs::from_indices, it could be ProjectionExprs::to_indices or something)
| // Rebase column indices to match the narrowed stream schema. | ||
| // The projection expressions have indices based on physical_file_schema, | ||
| // but the stream only contains the columns selected by the ProjectionMask. | ||
| let rebased_exprs = projection |
There was a problem hiding this comment.
here is another example where ProjectionExprs::map or something like that could be used
| ) -> datafusion_common::Result<Arc<dyn FileOpener>> { | ||
| let split_projection = self.projection.clone(); | ||
| let (expr_adapter_factory, schema_adapter_factory) = match ( |
| #[test] | ||
| fn test_cast_timestamp_with_timezone_to_timestamp() -> Result<()> { |
There was a problem hiding this comment.
These seem like they could be just as easily tested with .slt / arrow_cast calls 🤔
That would be less verbose and easier to maintain in my opinion
Parquet over to PhysicalExprAdapter, remove SchemaAdapteralamb
commented
Dec 4, 2025
Will we remove |
adriangb
commented
Dec 4, 2025
I think we could but I see no rush. Will address comments tomorrow 😄 |
Uh oh!
There was an error while loading. Please reload this page.
adriangb
commented
Dec 5, 2025
Apologies, I merged with failing |
adriangb
commented
Dec 5, 2025
Thanks for the reviews @kosiew ! |
Chips away at #14993 and #16800
Changes made in this PR:
PhysicalExprAdapterinstead ofSchemaAdapterVec<usize>so we can useProjectionMask::rootsand punt the complexity of implementingProjectionExprs->ProjectionMaskuntil a later PR (there is a draft in wip struct field pushdown parquet #18966 of what that might look like).