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

I was trying to do this , but since hash function is one-way function, I can only combine the record batches to get original value... I didn't see the benefits here, I think one way to reduce overhead is just supporting max/min filter here(as what duckdb does)... If there is a better way, please tell me. I am getting a little confused here. Thanks

So the idea is just to filter out values where the hash isn't present in the hashmap, that way we can filter out most values.

It's fine to focus on min/max first, the positive side is that evaluation is fast, but it won't filter out a lot of values if the min/max range from the left side is as big as the right side (regardless of number of values being much smaller on left side).

So the idea is just to filter out values where the hash isn't present in the hashmap, that way we can filter out most values.

I got it, I was think about rebuild original array using hashmap and records 😅...... I'll add the logic

It's fine to focus on min/max first, the positive side is that evaluation is fast, but it won't filter out a lot of values if the min/max range from the left side is as big as the right side (regardless of number of values being much smaller on left side).

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

Dandandan · 2024-11-07T08:10:03Z

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

@@ -1406,12 +1403,24 @@ impl HashJoinStream {
                self.hashes_buffer.resize(batch.num_rows(), 0);
                create_hashes(&keys_values, &self.random_state, &mut self.hashes_buffer)?;

+                let (filtered_batch, filtered_hashes) =
+                    if let Some(dynamic_filter) = &self.dynamic_filter_info {
+                        dynamic_filter.filter_probe_batch(


Hmm, filter_probe_batch shouldn't be added as filter on the hash_join but rather as PhysicalExpr (returning boolean) to filter the ParquetExec

Hmm, filter_probe_batch shouldn't be added as filter on the hash_join but rather as PhysicalExpr (returning boolean) to filter the ParquetExec

Sure, I'll refactor that, other than that I found a problem with ParquetExec after rebase the main #13298. If the pushdown cost is much more than do it directly. I am not sure whether it is a 'optimization' or not....

Lordworms · 2024-11-14T02:08:26Z

I repushed with an adaptive way to generate dynamic filter(when unique value exceed a threshold, generate range filter, otherwise inlist filter), and due to #13298 I didn't pushdown the filter down to parquet scan, I pushed the filter down to file stream, also tried with random data like

import pandas as pd
import numpy as np

part = pd.DataFrame({
    'p_partkey': np.random.randint(10, 21, size=1000),
    'p_brand': np.random.choice(['Brand#1', 'Brand#2', 'Brand#3'], 1000),
    'p_container': np.random.choice(['SM BOX', 'LG BOX', 'MED BOX'], 1000)
})

lineitem = pd.DataFrame({
    'l_partkey': np.random.randint(1, 1000001, size=100000000),
    'l_quantity': np.random.uniform(1, 50, size=100000000),
    'l_extendedprice': np.random.uniform(100, 10000, size=100000000)
})

and the result is

which only get like 40% of increase, I think we could try to pushdown filter down to parquet scan if the filter for reading parquet is optimized.

Also perhaps we need to improve performance of Inlist since when I tried 20 elements, the performance is 5 times slower than using range filter.

Lordworms · 2024-11-15T02:11:48Z

Not much difference on it since I think this optimization is specialized for specific data (build side contains data in certain range, etc)

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 force-pushed the dynamic_join_pushdown branch from 1aeb19a to 2ff52e6 Compare October 25, 2024 04:24

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

mbutrovich mentioned this pull request Oct 28, 2024

feat: Runtime bloom filter in hash join operator #13147

Closed

Lordworms force-pushed the dynamic_join_pushdown branch from 45d70a8 to 85b26d8 Compare November 7, 2024 01:55

Dandandan reviewed Nov 7, 2024

View reviewed changes

Lordworms mentioned this pull request Nov 7, 2024

ParquetScan with filter takes too much time to process #13298

Open

Lordworms force-pushed the dynamic_join_pushdown branch from 85b26d8 to e5fd1eb Compare November 14, 2024 02:08

Lordworms force-pushed the dynamic_join_pushdown branch from e5fd1eb to e5a33d8 Compare November 14, 2024 02:13

init

8762266

Lordworms force-pushed the dynamic_join_pushdown branch from ca627a0 to 8762266 Compare November 15, 2024 02:14

github-actions bot added the proto Related to proto crate label Nov 15, 2024

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

Dandandan Nov 6, 2024 • edited Loading

Choose a reason for hiding this comment

Lordworms Nov 6, 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

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

Choose a reason for hiding this comment

Lordworms Nov 7, 2024 • edited Loading

Choose a reason for hiding this comment

Lordworms commented Nov 14, 2024

Lordworms commented Nov 15, 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 Nov 6, 2024 •

edited

Loading

Lordworms Nov 6, 2024 •

edited

Loading

Dandandan commented Oct 24, 2024 •

edited

Loading

Lordworms commented Oct 25, 2024 •

edited

Loading

Lordworms Nov 7, 2024 •

edited

Loading

Lordworms commented Nov 15, 2024 •

edited

Loading