JoinOptimization: Add build side pushdown to probe side #13054

Lordworms · 2024-10-22T02:17:17Z

Which issue does this PR close?

Closes #7955

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Lordworms · 2024-10-22T02:19:07Z

For query like

select count(*) from part join lineitem on l_partkey=p_partkey where p_partkey > 10;

with

set datafusion.execution.parquet.pushdown_filters = true;

and with data scale like

The execution time is 5 times faster

before:

after:

The script I used to generate test parquet files

Since datafusion is unlike duckdb, the default way to do join is to utilize partitions, I added an extra phase for aggregate build side information.

I tested on my M3Pro Mac, with cpu info like

Lordworms · 2024-10-22T02:20:13Z

@Dandandan Sorry for the late response, I have reinvested it and format the code. Really appreciate your suggestion. Thanks a lot.

Dandandan · 2024-10-22T06:18:37Z

Really nice @Lordworms , will have a good look later today.

simonvandel · 2024-10-22T06:49:12Z

datafusion/core/src/physical_optimizer/join_filter_pushdown.rs

+            return Ok(Transformed::yes(Arc::new(new_hash_join)));
+        }
+        Ok(Transformed::no(plan))
+    } else if let Some(parquet_exec) = plan.as_any().downcast_ref::<ParquetExec>() {


I wonder if we could allow this cool optimization on any ExecutionPlan by adding the "with_dynamic_filter" to the ExecutionPlan?
Maybe also a "supports_dynamic_filter" to know when to call "with_dynamic_filter".
So same principle as the existing static filter pushdown setup.

That makes a lot of sense

Sure, I'll. refactor this, but currently support ParquetExec first? Since I think for other PhysicalScanExec, the way to add dynamic filter is to add a FilterExec above, but in parquet, we could utilize the predicate to add filters dynamically?

Dandandan · 2024-10-22T07:29:14Z

datafusion/core/src/datasource/physical_plan/parquet/mod.rs

@@ -711,10 +724,15 @@ impl DisplayAs for ParquetExec {
                        )
                    })
                    .unwrap_or_default();
-
+                let dynamic_filter =
+                    format!("dynamic_filter: {:?}", self.dynamic_filters);


I think it's best to only write this if available. + Would be good to implement Display for DynamicFilterInfo as well

Sorry for that, this is for debug use, I'll refactor it

Dandandan · 2024-10-22T07:30:51Z

datafusion/core/src/physical_optimizer/join_filter_pushdown.rs

+    }
+}
+
+impl PhysicalOptimizerRule for JoinFilterPushdown {


This is executed during plan time, won't it be reasonable to execute this during execution of HashJoin after build side is loaded?

Dandandan · 2024-10-22T07:46:34Z

datafusion/physical-plan/src/joins/dynamic_filters.rs

+        data_types: Vec<&DataType>,
+        total_batches: usize,
+    ) -> Result<Self, DataFusionError> {
+        let (max_accumulators, min_accumulators) = data_types


I wonder if we can use a logical plan / physical plan instead of manually invoking min/max accumulators? Not sure if that makes sense :)

I don't know either, First I want to try using min/max ScalarFunction, but the ScalaFunction translation from logical_expr to physical_expr is not implemented yet... I guess using accumulator would be more Intuitive？

Dandandan · 2024-10-22T07:50:18Z

datafusion/physical-plan/src/joins/dynamic_filters.rs

+        }
+
+        inner.batch_count = inner.batch_count.saturating_sub(1);
+        if records.num_rows() == 0 {


Couldn't the input of a join partition be empty, in which case it will never be finalized?

Haven't try that, I'll add a test for that

yes, it happens sometime, haven't got a way to solve it yet

I added a new judge "if a partition id has been recorded and it return empty batch again, we treat it as an empty partition and return true"

Dandandan · 2024-10-22T07:52:46Z

datafusion/physical-plan/src/joins/dynamic_filters.rs

+                let max_expr: Arc<dyn PhysicalExpr> = Arc::new(Literal::new(max_scalar));
+                let min_expr: Arc<dyn PhysicalExpr> = Arc::new(Literal::new(min_scalar));
+
+                let range_condition: Arc<dyn PhysicalExpr> = Arc::new(BinaryExpr::new(


Maybe it makes sense to not create a dynamic filter in certain conditions, eg min is minimum value of data type, max is maximum of datatype (expression will not filter out any rows)?

Dandandan · 2024-10-22T08:06:32Z

datafusion/physical-plan/src/joins/dynamic_filters.rs

+                let max_scalar = max_value.clone();
+                let min_scalar = min_value.clone();
+
+                let max_expr: Arc<dyn PhysicalExpr> = Arc::new(Literal::new(max_scalar));


For readability, it's better to use LogicalPlan here and convert to physical? This might also support some optimizations for free (e.g. expression simplification)...

Dandandan · 2024-10-22T13:13:46Z

datafusion/physical-plan/src/joins/dynamic_filters.rs

+
+struct DynamicFilterInfoInner {
+    max_accumulators: Vec<MaxAccumulator>,
+    min_accumulators: Vec<MinAccumulator>,


I wonder if we can just use batches: Vec<Vec<Arc<dyn Array>>> here and compute any filters from that?

This makes it pretty easy to support additional filters. It's pretty easy to add InListExpr filter based on this (I have some code locally).

I agree, I'll refactor it, after that maybe I can combine your commit.

Cool, after changes have cooled down a bit I can update and share my branch.
Adding something like that on top is actually quite easy after making the make_set helper public:

+ let col = + Arc::<datafusion_physical_expr::expressions::Column>::clone(column); + let batches: Vec<&dyn Array> = inner.batches[i] + .iter() + .map(|x| x as &dyn Array) + .collect(); + + // use inlist rather than min/max, use some threshold to avoid big inputs + // TODO: big inputs could be using bloom filter + if batches.iter().map(|x|x.len()).sum::<usize>() < 100 { + let batches = concat(batches.as_slice())?; + let set = make_set(&batches)?; + + let unique_condition = InListExpr::new(col, vec![], false, Some(set)); + condition = Arc::new(BinaryExpr::new( + Arc::new(unique_condition), Operator::And, - range_condition, - )) - as Arc<dyn PhysicalExpr>)), - None => Ok(Some(range_condition)), + condition, + )); } + Ok(Some(condition))

Something that would be fun to try is creating a PhysicalExpr from the JoinHashMap - that way we can build a good filter without any overhead.

I can do that

Lordworms · 2024-10-22T23:40:24Z

datafusion/physical-plan/src/joins/dynamic_filters.rs

+
+    let data_type = arrays[0].data_type();
+    match data_type {
+        DataType::Int8 => process_min_max!(arrays, Int8Array, Int8, i8),


Currently support numeric types columns(which adds a range filter), would support string types in next PR(add IN (xx, xx) filter)

String types could perform range as well, no?

I'll add it.

berkaysynnada · 2024-10-24T06:59:01Z

Could you update the display of sources to include these pushed-down filters? I believe it would be more clear if we could see them clearly in the output (I'm also curious to see how this impacts the test plans). Additionally, are there any benchmark results you could share with us?

Dandandan · 2024-10-24T08:18:21Z

Could you update the display of sources to include these pushed-down filters? I believe it would be more clear if we could see them clearly in the output (I'm also curious to see how this impacts the test plans). Additionally, are there any benchmark results you could share with us?

How would you display them in sources? The dynamic filter will only be added during execution, so it will only be available through e.g. ParquetExec after loading the build side.

berkaysynnada · 2024-10-24T10:24:43Z

How would you display them in sources? The dynamic filter will only be added during execution, so it will only be available through e.g. ParquetExec after loading the build side.

Doesn’t JoinFilterPushdown: PhysicalOptimizerRule modify the dynamic_filters in ParquetExec during optimizations? My intention was to ensure the rule is functioning correctly and to prevent this feature from being silently broken in the future. Perhaps we could add some unit tests to verify this.

Dandandan · 2024-10-24T10:34:06Z

How would you display them in sources? The dynamic filter will only be added during execution, so it will only be available through e.g. ParquetExec after loading the build side.

Doesn’t JoinFilterPushdown: PhysicalOptimizerRule modify the dynamic_filters in ParquetExec during optimizations? My intention was to ensure the rule is functioning correctly and to prevent this feature from being silently broken in the future. Perhaps we could add some unit tests to verify this.

Yes that's correct, just wanted to stress the actual filter isn't added during the optimization phase, only a "placeholder" that might be filled with some filter expressions during execution for particular columns. So it only shows ParquetExec might be filtered dynamically from the join values on certain columns, but the range or set of values will not be added at that point.

Lordworms · 2024-10-25T03:17:04Z

How would you display them in sources? The dynamic filter will only be added during execution, so it will only be available through e.g. ParquetExec after loading the build side.

Doesn’t JoinFilterPushdown: PhysicalOptimizerRule modify the dynamic_filters in ParquetExec during optimizations? My intention was to ensure the rule is functioning correctly and to prevent this feature from being silently broken in the future. Perhaps we could add some unit tests to verify this.

Maybe I can show them in 'explain analyze'

Something like

github-actions bot added logical-expr Logical plan and expressions physical-expr Physical Expressions optimizer Optimizer rules core Core DataFusion crate common Related to common crate functions labels Oct 22, 2024

simonvandel reviewed Oct 22, 2024

View reviewed changes

Dandandan reviewed Oct 22, 2024

View reviewed changes

Lordworms commented Oct 22, 2024

View reviewed changes

github-actions bot added the sqllogictest SQL Logic Tests (.slt) label Oct 25, 2024

Lordworms added 6 commits October 24, 2024 21:22

init

a06fd59

add feature

60d41d6

make with_dynamic_filters part of execution_plan trait

677536f

use batches instead of accumulator

cb03278

deal with empty partition

a7e7406

refactor

ec477e0

Lordworms added 3 commits October 24, 2024 21:22

supporting string

a7d383c

fix failed tests and refactor code

e9cc865

fix failed test, add analyze info, fix compile error

2ff52e6

Lordworms force-pushed the dynamic_join_pushdown branch from 1aeb19a to 2ff52e6 Compare October 25, 2024 04:24

Lordworms added 9 commits October 24, 2024 21:28

fix format

a8fb818

fix format

0309de6

remove print

d1f92cc

remove print

c4ac9af

fix ci

f6418b0

fix ci

610c7bd

fix ci

aa10fe7

fix ci

21b3cab

fix clippy

1993ea7

github-actions bot added the documentation Improvements or additions to documentation label Oct 25, 2024

fix clippy

45d70a8

mbutrovich mentioned this pull request Oct 28, 2024

feat: Runtime bloom filter in hash join operator #13147

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

JoinOptimization: Add build side pushdown to probe side #13054

JoinOptimization: Add build side pushdown to probe side #13054

Lordworms commented Oct 22, 2024

Lordworms commented Oct 22, 2024 •

edited

Loading

Lordworms commented Oct 22, 2024

Dandandan commented Oct 22, 2024

simonvandel Oct 22, 2024

Dandandan Oct 22, 2024

Lordworms Oct 22, 2024 •

edited

Loading

Dandandan Oct 22, 2024

Lordworms Oct 22, 2024

Dandandan Oct 22, 2024

Dandandan Oct 22, 2024 •

edited

Loading

Lordworms Oct 22, 2024

Dandandan Oct 22, 2024

Lordworms Oct 22, 2024

Lordworms Oct 23, 2024 •

edited

Loading

Lordworms Oct 23, 2024

Dandandan Oct 22, 2024 •

edited

Loading

Dandandan Oct 22, 2024 •

edited

Loading

Lordworms Oct 22, 2024

Dandandan Oct 22, 2024

Dandandan Oct 22, 2024

Lordworms Oct 22, 2024

Dandandan Oct 23, 2024

Dandandan Oct 31, 2024

Lordworms Oct 31, 2024

Lordworms Oct 22, 2024

Dandandan Oct 23, 2024

Lordworms Oct 23, 2024

berkaysynnada commented Oct 24, 2024

Dandandan commented Oct 24, 2024

berkaysynnada commented Oct 24, 2024

Dandandan commented Oct 24, 2024 •

edited

Loading

Lordworms commented Oct 25, 2024 •

edited

Loading

JoinOptimization: Add build side pushdown to probe side #13054

Are you sure you want to change the base?

JoinOptimization: Add build side pushdown to probe side #13054

Conversation

Lordworms commented Oct 22, 2024

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?

Lordworms commented Oct 22, 2024 • edited Loading

Lordworms commented Oct 22, 2024

Dandandan commented Oct 22, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Lordworms Oct 22, 2024 • edited Loading

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

Dandandan Oct 22, 2024 • edited Loading

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

Lordworms Oct 23, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Dandandan Oct 22, 2024 • edited Loading

Choose a reason for hiding this comment

Dandandan Oct 22, 2024 • edited Loading

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

berkaysynnada commented Oct 24, 2024

Dandandan commented Oct 24, 2024

berkaysynnada commented Oct 24, 2024

Dandandan commented Oct 24, 2024 • edited Loading

Lordworms commented Oct 25, 2024 • edited Loading

Lordworms commented Oct 22, 2024 •

edited

Loading

Lordworms Oct 22, 2024 •

edited

Loading

Dandandan Oct 22, 2024 •

edited

Loading

Lordworms Oct 23, 2024 •

edited

Loading

Dandandan Oct 22, 2024 •

edited

Loading

Dandandan Oct 22, 2024 •

edited

Loading

Dandandan commented Oct 24, 2024 •

edited

Loading

Lordworms commented Oct 25, 2024 •

edited

Loading