refactor: ParquetExec logical expr. => phys. expr. #5419

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

datafusion/core/src/datasource/listing/table.rs

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


Uses the real ExeuctionProps now :)

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

datafusion/core/src/physical_plan/file_format/parquet/row_filter.rs

    use parquet::arrow::parquet_to_arrow_schema;
    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

datafusion/core/src/physical_plan/file_format/parquet/row_filter.rs

-    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

datafusion/core/src/physical_plan/file_format/parquet/row_groups.rs

@@ -324,12 +329,15 @@ mod tests {
    fn row_group_pruning_predicate_partial_expr() {
        use datafusion_expr::{col, lit};
        // 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

datafusion/physical-expr/src/utils.rs

+///
+/// 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

datafusion/core/src/datasource/file_format/mod.rs

@@ -84,7 +84,7 @@ pub trait FileFormat: Send + Sync + fmt::Debug {
        &self,
        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

datafusion/core/src/physical_optimizer/pruning.rs

-        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!

datafusion/core/src/physical_optimizer/pruning.rs

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

datafusion/core/src/physical_optimizer/pruning.rs

-                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

datafusion/core/src/physical_optimizer/pruning.rs

+        // 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.

github-actions bot added core Core datafusion crate physical-expr Physical Expressions 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

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor: ParquetExec logical expr. => phys. expr. #5419

refactor: ParquetExec logical expr. => phys. expr. #5419

crepererum commented Feb 27, 2023

crepererum Feb 27, 2023

crepererum Feb 27, 2023

crepererum commented Feb 27, 2023

alamb left a comment

alamb Feb 28, 2023

alamb Feb 28, 2023

alamb Feb 28, 2023

ozankabak Mar 1, 2023

crepererum Mar 1, 2023

alamb Mar 1, 2023

ozankabak Mar 1, 2023

alamb Feb 28, 2023

alamb Feb 28, 2023

alamb Feb 28, 2023

alamb Feb 28, 2023

crepererum Feb 28, 2023

ursabot commented Mar 1, 2023

alamb commented Mar 23, 2023

jiangzhx commented Mar 30, 2023

refactor: ParquetExec logical expr. => phys. expr. #5419

refactor: ParquetExec logical expr. => phys. expr. #5419

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?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

crepererum commented Feb 27, 2023

alamb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ursabot commented Mar 1, 2023

alamb commented Mar 23, 2023

jiangzhx commented Mar 30, 2023