refactor: ParquetExec logical expr. => phys. expr. by crepererum · Pull Request #5419 · apache/datafusion

crepererum · 2023-02-27T15:11:29Z

Which issue does this PR close?

Closes #4695.

Rationale for this change

Use Arc<dyn PhysicalExpr> instead of Expr within ParquetExec and move lowering from logical to physical expression into plan lowering (e.g. ListingTable).

This is in line w/ all other physical plan nodes (e.g. FilterExpr) and simplifies reasoning within physical optimizer but also allows correct passing of ExecutionProps into the conversion.

What changes are included in this PR?

Basically a bunch of type changes. The tests still use logical expressions because they are easier to construct and its a simple conversion from logical to phys. (at least in the test cases).

Are these changes tested?

All existing tests pass. Removed tests that no longer apply. Added one to demonstrate PruningPredicate behavior when no columns are referenced (see #5386 (comment) ).

Are there any user-facing changes?

Phys. plan printouts look slightly differently.
BREAKING: Serialization has changed.
BREAKING: ParquetExec arguments changed.

crepererum · 2023-02-27T15:13:31Z

+                &expr,
+                &table_df_schema,
+                &self.table_schema,
+                state.execution_props(),


Uses the real ExeuctionProps now :)

crepererum · 2023-02-27T15:14:24Z

    use parquet::file::reader::{FileReader, SerializedFileReader};
    use rand::prelude::*;

-    // Assume a column expression for a column not in the table schema is a projected column and ignore it


This test is no longer relevant because you cannot lower such expression from logical to physical.

crepererum · 2023-02-27T15:17:25Z

I've also fixed a few test expressions that where buggy (or at least would require type coercion which would run on the logical plan), most notably:

negate(cast(some_int to UTF8)) => cast(negate(some_int) to UTF8)
some_bool.and/or(some_int % 2) => some_bool.and/or(some_int % 2 == 0)

alamb

Thank you @crepererum -- I went through this PR's code and tsets carefully and it made sense to me. Thank you

Can you please confirm that passing a false predicate to the pruning logic on the main branch will not prune anything as well (aka that there is no behavior change?)

alamb · 2023-02-28T17:07:39Z

-    fn pre_visit(&mut self, expr: &Expr) -> Result<RewriteRecursion> {
-        if let Expr::Column(column) = expr {
-            if let Ok(idx) = self.file_schema.index_of(&column.name) {
+impl<'a> TreeNodeRewriter<Arc<dyn PhysicalExpr>> for FilterCandidateBuilder<'a> {


This is cool to see that there is an equivalent logical and physical rewriter

alamb · 2023-02-28T17:09:40Z

        // test row group predicate with partially supported expression
-        // int > 1 and int % 2 => c1_max > 1 and true
-        let expr = col("c1").gt(lit(15)).and(col("c2").modulus(lit(2)));
+        // (int > 1) and ((int % 2) = 0) => c1_max > 1 and true


thank you - this is a good change

alamb · 2023-02-28T17:11:24Z

+///
+/// This works recursively.
+pub fn get_phys_expr_columns(pred: &Arc<dyn PhysicalExpr>) -> HashSet<Column> {
+    let mut rewriter = ColumnCollector::default();


👍 this is quite clever

Interesting! We had the same need (collecting columns) emerge in SHJ implementation, so we used this more lightweight recursion:

fn collect_columns_recursive(expr: &Arc<dyn PhysicalExpr>, columns: &mut Vec<Column>) { if let Some(column) = expr.as_any().downcast_ref::<Column>() { if !columns.iter().any(|c| c.eq(column)) { columns.push(column.clone()) } } expr.children() .iter() .for_each(|e| collect_columns_recursive(e, columns)) } fn collect_columns(expr: &Arc<dyn PhysicalExpr>) -> Vec<Column> { let mut columns = vec![]; collect_columns_recursive(expr, &mut columns); columns }

We used a Vec instead of a HashSet due to anticipated small sizes, but the code is essentially the same 🙂

This makes me think that doing a comprehensive code review and collecting/coalescing/documenting utilities such as this may simplify the codebase, and could be a worthy pursuit.

@ozankabak the issue w/ Vec is that you have a O(n^2) complexity in the number of used columns. In InfluxDB IOx we sometimes have schemas w/ over 200 columns and I'm somewhat worried that such a simple oversight quickly turns into a performance bug.

This makes me think that doing a comprehensive code review and collecting/coalescing/documenting utilities such as this may simplify the codebase, and could be a worthy pursuit.

This absolutely would be a worthy pursuit -- is that something you could help lead @ozankabak ?

Yes, happy to help!

alamb · 2023-02-28T17:15:08Z

        state: &SessionState,
        conf: FileScanConfig,
-        filters: &[Expr],
+        filters: Option<&Arc<dyn PhysicalExpr>>,


I think it is a good change to have this take a single PhysicalExpr rather than a slice as it is more consistent with TableProvider::scan 👍

alamb · 2023-02-28T17:16:05Z

-        let stat_schema = Schema::new(stat_fields);
-        let stat_dfschema = DFSchema::try_from(stat_schema.clone())?;
-
-        // TODO allow these properties to be passed in


👍 now it uses the real thing . Nice!

alamb · 2023-02-28T17:21:56Z

-                min_column_expr
-                    .lt_eq(expr_builder.scalar_expr().clone())
-                    .and(expr_builder.scalar_expr().clone().lt_eq(max_column_expr))
+                Arc::new(phys_expr::BinaryExpr::new(


it is unfortunate there isn't as nice "fluent" style APIs for building physical expressions. Maybe we can add a PhysicalExprExt trait that knew how to do so.

I'll file a ticket to track doing so as a follow on PR

alamb · 2023-02-28T17:26:32Z

+        // false
+        // constant literals that do NOT refer to any columns are currently not evaluated at all, hence the result is
+        // "all true"
+        let expr = lit(false);


this is unfortunate (an all false predicate should skip the entire file). However, as long as this PR doesn't make the pruning worse in this situation I think it is fine (and we can improve things in a follow on PR)

It's not making things worse. This was the state of the art even before #5386, but I was asked here #5386 (comment) for the concrete semantics and I couldn't find a test that already covered that, so I've added one.

Use `Arc<dyn PhysicalExpr>` instead of `Expr` within `ParquetExec` and move lowering from logical to physical expression into plan lowering (e.g. `ListingTable`). This is in line w/ all other physical plan nodes (e.g. `FilterExpr`) and simplifies reasoning within physical optimizer but also allows correct passing of `ExecutionProps` into the conversion. Closes apache#4695.

ursabot · 2023-03-01T16:33:37Z

Benchmark runs are scheduled for baseline = c4b8958 and contender = 03fbf9f. 03fbf9f is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ec2-t3-xlarge-us-east-2] ec2-t3-xlarge-us-east-2
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on test-mac-arm] test-mac-arm
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ursa-i9-9960x] ursa-i9-9960x
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ursa-thinkcentre-m75q] ursa-thinkcentre-m75q
Buildkite builds:
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java

alamb · 2023-03-23T19:54:36Z

This PR appears to have introduced a regression -- see #5708

jiangzhx · 2023-03-30T07:50:06Z

This feature is not implemented: Physical plan does not support logical expression EXISTS ()
#5789

version: 19.0.0 worked.

…26405) ## Which issue does this PR close? - Related to apache#21968. There is no dedicated issue for this change. ## Rationale for this change Building a pruning predicate for a Parquet file with a wide schema takes time proportional to the number of columns in the file, once for every comparison in the predicate, even when the predicate only references one column. A Parquet scan builds several pruning predicates for each file it opens (for example for row group statistics, the page index and bloom filters), so on wide tables this cost is paid many times per file. The cause is in `PruningExpressionBuilder::try_new`. For every comparison it converts the whole file schema into a `DFSchema`, which runs `DFSchema::check_names` and inserts every field name into a `BTreeSet`. The `DFSchema` is then only used through `schema.as_arrow()`, to get the data type of the expression inside a `CAST` or `TRY_CAST`. That is the Arrow schema the builder already has. In a profile of `wide_schema` Q01 (1024 columns, 256 files), `DFSchema::try_from` called from `PruningExpressionBuilder::try_new` was about 5% of CPU samples. With a file metadata cache large enough to hold every footer it was 9.8%. The `DFSchema` dates from apache#3422, when pruning rewrote logical `Expr`s and needed a `DFSchema` to resolve their types. apache#5419 moved pruning to physical expressions, which only need the Arrow schema, but kept the `DFSchema` parameter. ## What changes are included in this PR? - `rewrite_expr_to_prunable` and `rewrite_cast_child_to_prunable` take `&Schema` instead of a `DFSchema`. - `PruningExpressionBuilder::try_new` passes the schema it already has, rather than building a `DFSchema` for each comparison. - Two unit tests that built a `DFSchema` only to call `rewrite_expr_to_prunable` now pass the `Schema`. ## What is the testing strategy for this PR? The change is covered by existing tests, so this PR adds none. Both functions are private, and their behaviour is unchanged apart from the duplicate-name case described below. - `datafusion-pruning` tests, including `test_rewrite_expr_to_prunable` and `test_rewrite_expr_to_prunable_error` - `datafusion-datasource-parquet` lib tests - `datafusion` `parquet_integration` tests, which cover row group and page index pruning ## Are there any user-facing changes? There are no API changes, because both functions are private. There is one behaviour change. `DFSchema::try_from` fails when the schema has duplicate field names, so previously no comparison against such a schema could be used for pruning. Now pruning goes ahead and resolves the column by name to the first field with that name. The column lookup in the same function (`schema.column_with_name`) and the Parquet statistics lookup (`parquet_column`) already resolve names this way. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 <[email protected]>

github-actions Bot added core Core DataFusion crate physical-expr Changes to the physical-expr crates labels Feb 27, 2023

crepererum commented Feb 27, 2023

View reviewed changes

alamb approved these changes Feb 28, 2023

View reviewed changes

crepererum added 4 commits March 1, 2023 10:22

feat: get_phys_expr_columns util

0ef5906

feat: add reassign_predicate_columns util

afb03c8

feat: PhysicalExprRef type alias

bc5d03f

crepererum force-pushed the crepererum/issue4695 branch from 3dae660 to b909edf Compare March 1, 2023 09:23

alamb merged commit 03fbf9f into apache:main Mar 1, 2023

alamb mentioned this pull request Mar 23, 2023

[REGRESSION] Parquet row group pruning incorrectly prunes out row groups when columns names have . in them #5708

Closed

jiangzhx mentioned this pull request Mar 30, 2023

SQL case, This feature is not implemented: Physical plan does not support logical expression EXISTS (<subquery>) #5789

Closed

cetra3 mentioned this pull request Oct 16, 2026

perf: avoid building a DFSchema for every pruning comparison #26405

Merged

Conversation

crepererum commented Feb 27, 2023

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

crepererum commented Feb 27, 2023

Uh oh!

alamb left a comment

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!

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!

ursabot commented Mar 1, 2023

Uh oh!

alamb commented Mar 23, 2023

Uh oh!

jiangzhx commented Mar 30, 2023

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants