Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3198945197
> > Would anyone here like to review #17153 which enables this for left semi
joins / `IN (subquery)` / `WHERE {NOT} EXISTS`?
>
> I am not likely to have time, but if I do I will take a loo
No worries Andrew I think @kosiew is going to be working on that and I'll
help review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
alamb commented on PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3197073401
> Would anyone here like to review #17153 which enables this for left semi
joins / `IN (subquery)` / `WHERE {NOT} EXISTS`?
I am not likely to have time, but if I do I will take a loo
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
alamb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3188967321 @nuno-faria has reported a bug related to this PR: - https://github.com/apache/datafusion/issues/17188 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3184583578
Would anyone here like to review
https://github.com/apache/datafusion/pull/17153 which enables this for left
semi joins / `IN (subquery)` / `WHERE {NOT} EXISTS`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb merged PR #16445: URL: https://github.com/apache/datafusion/pull/16445 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3175316144 I plan on merging this in a couple hours if there are no objections -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3166494580 Fixed @xudong963, thanks so much for your review! I will leave this up until early next week to see if we get any more feedback. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2260615307
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -966,8 +1005,33 @@ impl ExecutionPlan for HashJoinExec {
&self.children(),
));
}
-FilterDescription::from_children(parent_filters, &self.children())
-// TODO: push down our self filters to children in the post
optimization phase
+
+// Get basic filter descriptions for both children
+let mut left_child =
crate::filter_pushdown::ChildFilterDescription::from_child(
+&parent_filters,
+self.left(),
+)?;
+let mut right_child =
crate::filter_pushdown::ChildFilterDescription::from_child(
+&parent_filters,
+self.right(),
+)?;
+
+// Add dynamic filters in Post phase if enabled
+if matches!(phase, FilterPushdownPhase::Post)
+&& config.optimizer.enable_dynamic_filter_pushdown
+{
+// Add placeholder to left side (build side)
+left_child = left_child.with_self_filter(lit(true));
Review Comment:
aa2b512
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2260557873
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -966,8 +1005,33 @@ impl ExecutionPlan for HashJoinExec {
&self.children(),
));
}
-FilterDescription::from_children(parent_filters, &self.children())
-// TODO: push down our self filters to children in the post
optimization phase
+
+// Get basic filter descriptions for both children
+let mut left_child =
crate::filter_pushdown::ChildFilterDescription::from_child(
+&parent_filters,
+self.left(),
+)?;
+let mut right_child =
crate::filter_pushdown::ChildFilterDescription::from_child(
+&parent_filters,
+self.right(),
+)?;
+
+// Add dynamic filters in Post phase if enabled
+if matches!(phase, FilterPushdownPhase::Post)
+&& config.optimizer.enable_dynamic_filter_pushdown
+{
+// Add placeholder to left side (build side)
+left_child = left_child.with_self_filter(lit(true));
Review Comment:
You're right I don't think it is, not sure why I / Claude put that in there!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
xudong963 commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2260549218
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -966,8 +1005,33 @@ impl ExecutionPlan for HashJoinExec {
&self.children(),
));
}
-FilterDescription::from_children(parent_filters, &self.children())
-// TODO: push down our self filters to children in the post
optimization phase
+
+// Get basic filter descriptions for both children
+let mut left_child =
crate::filter_pushdown::ChildFilterDescription::from_child(
+&parent_filters,
+self.left(),
+)?;
+let mut right_child =
crate::filter_pushdown::ChildFilterDescription::from_child(
+&parent_filters,
+self.right(),
+)?;
+
+// Add dynamic filters in Post phase if enabled
+if matches!(phase, FilterPushdownPhase::Post)
+&& config.optimizer.enable_dynamic_filter_pushdown
+{
+// Add placeholder to left side (build side)
+left_child = left_child.with_self_filter(lit(true));
Review Comment:
Why do we need this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
xudong963 commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3155072658 I'll have a look tomorrow -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2248281203
##
datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs:
##
@@ -734,6 +734,366 @@ async fn test_topk_dynamic_filter_pushdown() {
);
}
+#[tokio::test]
+async fn test_hashjoin_dynamic_filter_pushdown() {
+use datafusion_common::JoinType;
+use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode};
+
+// Create build side with limited values
+let build_batches = vec![record_batch!(
+("a", Utf8, ["aa", "ab"]),
+("b", Utf8, ["ba", "bb"]),
Review Comment:
cf834d315
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2248281203
##
datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs:
##
@@ -734,6 +734,366 @@ async fn test_topk_dynamic_filter_pushdown() {
);
}
+#[tokio::test]
+async fn test_hashjoin_dynamic_filter_pushdown() {
+use datafusion_common::JoinType;
+use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode};
+
+// Create build side with limited values
+let build_batches = vec![record_batch!(
+("a", Utf8, ["aa", "ab"]),
+("b", Utf8, ["ba", "bb"]),
Review Comment:
4e04665b7
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
zhuqi-lucas commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2247959415
##
datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs:
##
@@ -734,6 +734,366 @@ async fn test_topk_dynamic_filter_pushdown() {
);
}
+#[tokio::test]
+async fn test_hashjoin_dynamic_filter_pushdown() {
+use datafusion_common::JoinType;
+use datafusion_physical_plan::joins::{HashJoinExec, PartitionMode};
+
+// Create build side with limited values
+let build_batches = vec![record_batch!(
+("a", Utf8, ["aa", "ab"]),
+("b", Utf8, ["ba", "bb"]),
Review Comment:
We may add some Utf8View fields testing cases, because our default mapping
already changing to Utf8View.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
alamb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3141417070 Thanks @adriangb -- I will put this on my list of PRs to review more carefully in the next few days -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3141365277 Personally I would like to move forward with this and then make another PR to push down a reference to the entire hash table. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3141364005 Here it is in action: ```sql COPY ( with data as ( select unnest(generate_series(1, )) as id ) select id, (id / 10) as partition_id from data ) TO 'test/t1/' STORED AS parquet PARTITIONED BY (partition_id); CREATE EXTERNAL TABLE t1 ( id int ) STORED AS PARQUET PARTITIONED BY (partition_id int) LOCATION 'test/t1/'; COPY ( with data as ( select unnest(generate_series(1, 100)) as id ) select id, (id / 10) as partition_id from data ) TO 'test/t2/' STORED AS parquet PARTITIONED BY (partition_id); CREATE EXTERNAL TABLE t2 ( id int ) STORED AS PARQUET PARTITIONED BY (partition_id int) LOCATION 'test/t2/'; SET datafusion.optimizer.enable_dynamic_filter_pushdown = false; explain analyze SELECT count(*) FROM t1 JOIN t2 USING (id); SET datafusion.optimizer.enable_dynamic_filter_pushdown = true; explain analyze SELECT count(*) FROM t1 JOIN t2 USING (id); ``` | dynamic filters | time (ms) | bytes scanned | |-|---|---| | on |55 | 376,155 | | off | 401 | 296,192,099 | Explain plans ``` ❯ ./target/release/datafusion-cli DataFusion CLI v49.0.0 > COPY ( with data as ( select unnest(generate_series(1, )) as id ) select id, (id / 10) as partition_id from data ) TO 'test/t1/' STORED AS parquet PARTITIONED BY (partition_id); CREATE EXTERNAL TABLE t1 ( id int ) STORED AS PARQUET PARTITIONED BY (partition_id int) LOCATION 'test/t1/'; COPY ( with data as ( select unnest(generate_series(1, 100)) as id ) select id, (id / 10) as partition_id from data ) TO 'test/t2/' STORED AS parquet PARTITIONED BY (partition_id); CREATE EXTERNAL TABLE t2 ( id int ) STORED AS PARQUET PARTITIONED BY (partition_id int) LOCATION 'test/t2/'; +--+ | count| +--+ | | +--+ 1 row(s) fetched. Elapsed 16.986 seconds. 0 row(s) fetched. Elapsed 0.015 seconds. +---+ | count | +---+ | 100 | +---+ 1 row(s) fetched. Elapsed 0.001 seconds. 0 row(s) fetched. Elapsed 0.000 seconds. > SET datafusion.optimizer.enable_dynamic_filter_pushdown = false; 0 row(s) fetched. Elapsed 0.000 seconds. > explain analyze SELECT count(*) FROM t1 JOIN t2 USING (id); +---+-- -
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3141237953 Guess it just doesn't make a difference then. Thanks! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
alamb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3141204546 It did run > Benchmark tpch_mem_sf1.json I can run whatever benchmark you want - just let me know -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3141186574 Might be worth running TPCH or another benchmark with joins? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
alamb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3141175283 🤖: Benchmark completed Details ``` Comparing HEAD and hash-join-pushdown Benchmark clickbench_extended.json ┏━━┳━┳┳━━━┓ ┃ Query┃HEAD ┃ hash-join-pushdown ┃Change ┃ ┡━━╇━╇╇━━━┩ │ QQuery 0 │ 1936.73 ms │ 1977.69 ms │ no change │ │ QQuery 1 │ 762.69 ms │ 763.75 ms │ no change │ │ QQuery 2 │ 1481.87 ms │ 1466.68 ms │ no change │ │ QQuery 3 │ 687.14 ms │ 648.45 ms │ +1.06x faster │ │ QQuery 4 │ 1362.81 ms │ 1405.27 ms │ no change │ │ QQuery 5 │ 14887.33 ms │14948.71 ms │ no change │ │ QQuery 6 │ 2040.21 ms │ 2100.60 ms │ no change │ │ QQuery 7 │ 1859.25 ms │ 1911.89 ms │ no change │ └──┴─┴┴───┘ ┏━━━┳┓ ┃ Benchmark Summary ┃┃ ┡━━━╇┩ │ Total Time (HEAD) │ 25018.03ms │ │ Total Time (hash-join-pushdown) │ 25223.04ms │ │ Average Time (HEAD) │ 3127.25ms │ │ Average Time (hash-join-pushdown) │ 3152.88ms │ │ Queries Faster│ 1 │ │ Queries Slower│ 0 │ │ Queries with No Change│ 7 │ │ Queries with Failure │ 0 │ └───┴┘ Benchmark clickbench_partitioned.json ┏━━┳━┳┳━━━┓ ┃ Query┃HEAD ┃ hash-join-pushdown ┃Change ┃ ┡━━╇━╇╇━━━┩ │ QQuery 0 │ 2.56 ms │2.22 ms │ +1.15x faster │ │ QQuery 1 │35.52 ms │ 33.65 ms │ +1.06x faster │ │ QQuery 2 │83.47 ms │ 82.33 ms │ no change │ │ QQuery 3 │98.75 ms │ 99.65 ms │ no change │ │ QQuery 4 │ 589.45 ms │ 609.21 ms │ no change │ │ QQuery 5 │ 853.26 ms │ 893.98 ms │ no change │ │ QQuery 6 │ 2.30 ms │2.23 ms │ no change │ │ QQuery 7 │39.08 ms │ 39.01 ms │ no change │ │ QQuery 8 │ 846.57 ms │ 866.28 ms │ no change │ │ QQuery 9 │ 1192.91 ms │ 1175.10 ms │ no change │ │ QQuery 10│ 257.51 ms │ 260.38 ms │ no change │ │ QQuery 11│ 293.94 ms │ 292.24 ms │ no change │ │ QQuery 12│ 877.37 ms │ 896.65 ms │ no change │ │ QQuery 13│ 1238.00 ms │ 1291.84 ms │ no change │ │ QQuery 14│ 826.31 ms │ 833.51 ms │ no change │ │ QQuery 15│ 792.69 ms │ 800.79 ms │ no change │ │ QQuery 16│ 1597.34 ms │ 1619.89 ms │ no change │ │ QQuery 17│ 1573.03 ms │ 1627.82 ms │ no change │ │ QQuery 18│ 2835.98 ms │ 2880.59 ms │ no change │ │ QQuery 19│86.15 ms │ 88.40 ms │ no change │ │ QQuery 20│ 1118.68 ms │ 1177.86 ms │ 1.05x slower │ │ QQuery 21│ 1262.10 ms │ 1328.57 ms │ 1.05x slower │ │ QQuery 22│ 2066.92 ms │ 2236.89 ms │ 1.08x slower │ │ QQuery 23│ 7344.24 ms │ 7580.33 ms │ no change │ │ QQuery 24│ 444.04 ms │ 457.65 ms │ no change │ │ QQuery 25│ 301.35 ms │ 313.55 ms │ no change │ │ QQuery 26│ 437.08 ms │ 446.77 ms │ no change │ │ QQuery 27│ 1521.59 ms │ 1569.90 ms │ no change │ │ QQuery 28│ 11853.42 ms │12776.43 ms │ 1.08x slower │ │ QQuery 29│ 520.67 ms │ 522.14 ms │ no change │ │ QQuery 30│ 778.74 ms │ 798.51 ms │ no change │ │ QQuery 31│ 795.04 ms │ 802.20 ms │ no change │ │ QQuery 32│ 2425.30 ms │ 2434.52 ms │ no change │ │ QQuery 33│ 3152.55 ms │ 3222.42 ms │ no change │ │ QQuery 34│ 3241.80 ms │ 3218.07 ms │ no change │ │ QQuery 35│ 1285.41 ms │ 1276.66 ms │ no change │ │ QQuery 36│ 120.40 ms │ 125.16 ms │ no change │ │ QQuery 37│50.62 ms │ 51.33 ms │ no change │ │ QQuery 38│ 121.37 ms │ 120.85 ms │ no change │ │ QQuery 39│ 188.72 ms │ 200.24 ms │ 1.06x slower │ │ QQuery 40│41.17 ms │ 42.82 ms │
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2246192857
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -1039,12 +1196,50 @@ async fn collect_left_input(
let data = JoinLeftData::new(
hashmap,
single_batch,
-left_values,
+left_values.clone(),
Mutex::new(visited_indices_bitmap),
AtomicUsize::new(probe_threads_count),
reservation,
);
+// Update dynamic filter with min/max bounds if provided
+if num_rows > 0 {
Review Comment:
done!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
alamb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3141078662 🤖 `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running Linux aal-dev 6.11.0-1016-gcp #16~24.04.1-Ubuntu SMP Wed May 28 02:40:52 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux Comparing hash-join-pushdown (b135cd8614af0131d5e1484216b84499f37fe465) to d376a32f1f78d26a760c261878054bb1328800cc [diff](https://github.com/apache/datafusion/compare/d376a32f1f78d26a760c261878054bb1328800cc..b135cd8614af0131d5e1484216b84499f37fe465) using: tpch_mem clickbench_partitioned clickbench_extended Results will be posted here when complete -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3140948881 @alamb could I ask you to kick off some benchmarks? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3140961969 I think I've addressed all of the feedback and rebased on main / changes broken out into other PRs. @Dandandan @xudong963 I've tagged you both for review. @alamb would you mind kicking off benchmarks? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2246082774
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -943,10 +978,71 @@ impl ExecutionPlan for HashJoinExec {
try_embed_projection(projection, self)
}
}
+
+fn gather_filters_for_pushdown(
+&self,
+phase: FilterPushdownPhase,
+parent_filters: Vec>,
+config: &datafusion_common::config::ConfigOptions,
+) -> Result {
+// Don't allow parent filters to be pushed down for now
+// Only add our dynamic filter during the Post phase
+if !matches!(phase, FilterPushdownPhase::Post) {
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters));
+}
+
+// Only push down dynamic filters if enabled
+if config.optimizer.enable_dynamic_filter_pushdown {
+let filter = Arc::clone(&self.dynamic_filter) as Arc;
+// Push the dynamic filter to the right side (probe side) only
+// Left side (build side) gets empty vec, right side gets the
filter
+let filters_for_children = vec![vec![], vec![filter]];
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters)
+.with_self_filters_for_children(filters_for_children));
+}
+
+Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters))
+}
+}
+
+/// Compute min/max bounds for each column in the given arrays
+fn compute_bounds(arrays: &[ArrayRef]) -> Result> {
+arrays
+.iter()
+.map(|array| {
+if array.is_empty() {
+// Return NULL values for empty arrays
+return Ok((
+ScalarValue::try_from(array.data_type())?,
+ScalarValue::try_from(array.data_type())?,
+));
+}
+
+// Compute min/max using ScalarValue's utilities
+let mut min_val = ScalarValue::try_from_array(array, 0)?;
Review Comment:
67a0a1c41
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2246051561
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -943,10 +978,71 @@ impl ExecutionPlan for HashJoinExec {
try_embed_projection(projection, self)
}
}
+
+fn gather_filters_for_pushdown(
+&self,
+phase: FilterPushdownPhase,
+parent_filters: Vec>,
+config: &datafusion_common::config::ConfigOptions,
+) -> Result {
+// Don't allow parent filters to be pushed down for now
+// Only add our dynamic filter during the Post phase
+if !matches!(phase, FilterPushdownPhase::Post) {
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters));
+}
+
+// Only push down dynamic filters if enabled
+if config.optimizer.enable_dynamic_filter_pushdown {
+let filter = Arc::clone(&self.dynamic_filter) as Arc;
+// Push the dynamic filter to the right side (probe side) only
+// Left side (build side) gets empty vec, right side gets the
filter
+let filters_for_children = vec![vec![], vec![filter]];
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters)
+.with_self_filters_for_children(filters_for_children));
+}
+
+Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters))
+}
+}
+
+/// Compute min/max bounds for each column in the given arrays
+fn compute_bounds(arrays: &[ArrayRef]) -> Result> {
+arrays
+.iter()
+.map(|array| {
+if array.is_empty() {
+// Return NULL values for empty arrays
+return Ok((
+ScalarValue::try_from(array.data_type())?,
+ScalarValue::try_from(array.data_type())?,
+));
+}
+
+// Compute min/max using ScalarValue's utilities
+let mut min_val = ScalarValue::try_from_array(array, 0)?;
Review Comment:
Moved min_batch in https://github.com/apache/datafusion/pull/16593
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2245591958
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -943,10 +978,71 @@ impl ExecutionPlan for HashJoinExec {
try_embed_projection(projection, self)
}
}
+
+fn gather_filters_for_pushdown(
+&self,
+phase: FilterPushdownPhase,
+parent_filters: Vec>,
+config: &datafusion_common::config::ConfigOptions,
+) -> Result {
+// Don't allow parent filters to be pushed down for now
+// Only add our dynamic filter during the Post phase
+if !matches!(phase, FilterPushdownPhase::Post) {
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters));
+}
+
+// Only push down dynamic filters if enabled
+if config.optimizer.enable_dynamic_filter_pushdown {
+let filter = Arc::clone(&self.dynamic_filter) as Arc;
+// Push the dynamic filter to the right side (probe side) only
+// Left side (build side) gets empty vec, right side gets the
filter
+let filters_for_children = vec![vec![], vec![filter]];
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters)
+.with_self_filters_for_children(filters_for_children));
+}
+
+Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters))
+}
+}
+
+/// Compute min/max bounds for each column in the given arrays
+fn compute_bounds(arrays: &[ArrayRef]) -> Result> {
+arrays
+.iter()
+.map(|array| {
+if array.is_empty() {
+// Return NULL values for empty arrays
+return Ok((
+ScalarValue::try_from(array.data_type())?,
+ScalarValue::try_from(array.data_type())?,
+));
+}
+
+// Compute min/max using ScalarValue's utilities
+let mut min_val = ScalarValue::try_from_array(array, 0)?;
Review Comment:
still need to do this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2245195106
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -1039,12 +1196,50 @@ async fn collect_left_input(
let data = JoinLeftData::new(
hashmap,
single_batch,
-left_values,
+left_values.clone(),
Mutex::new(visited_indices_bitmap),
AtomicUsize::new(probe_threads_count),
reservation,
);
+// Update dynamic filter with min/max bounds if provided
+if num_rows > 0 {
Review Comment:
If it's easy to pipe through sounds good, but in my mind
`enable_dynamic_filter_pushdown` is more for us debugging as we've been adding
these dynamic filters / proving equivalence of query results more so than an
option users would toggle on or off -> it's not a big deal if some work is done
even if the flag is off. Our goal should be that the extra work done for
dynamic filters is always cheap enough that it does't make queries slower even
if the flag is off but the work is done anyway.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
LiaCastaneda commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2184901077
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -1039,12 +1196,50 @@ async fn collect_left_input(
let data = JoinLeftData::new(
hashmap,
single_batch,
-left_values,
+left_values.clone(),
Mutex::new(visited_indices_bitmap),
AtomicUsize::new(probe_threads_count),
reservation,
);
+// Update dynamic filter with min/max bounds if provided
+if num_rows > 0 {
Review Comment:
should we avoid computing these bounds if `enable_dynamic_filter_pushdown`
is not enabled?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3026127559 Btw here's an article that explains how DuckDB does join filter pushdown. It sounds like they _only_ push down min/max filters: https://duckdb.org/2024/09/09/announcing-duckdb-110.html#dynamic-filter-pushdown-from-joins -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3025944992 I've pulled out part of this PR, the part about pushing filters down through HashJoinExec plus some new changes to the filter pushdown APIs into https://github.com/apache/datafusion/pull/16642 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3006600650 @alamb I'd be interested to see what benchmarks say if you don't mind kicking them off? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2164715273
##
datafusion/physical-plan/src/filter_pushdown.rs:
##
@@ -353,6 +353,18 @@ impl FilterDescription {
}
}
+pub fn with_child_pushdown(
Review Comment:
More APIs 🤮. I really need to circle back to doing some whiteboard design
for these. It's complex and won't be pretty but I'm sure it can be better than
it is right now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3004662324 @Dandandan any chance you'd be willing to contribute your implementation of sharing `Arc` so we use something we know is working / I don't have to re-invent the wheel? I think you can just push it to this branch. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3001813655 I'll add that :) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
Dandandan commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3001863494 > I was originally planning on keeping this PR smaller but it's been growing so I might as well add the Arc :) Feel free to PR it however you like ;) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
Dandandan commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-3001717244 To share some experience, we recently added some similar pushdown for HashJoinExec (at Coralogix) using sharing of `Arc` / comparing column hashes and it is seems so far very effective with predicate pushdown enabled. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2164716153
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -666,10 +679,25 @@ impl DisplayAs for HashJoinExec {
.map(|(c1, c2)| format!("({c1}, {c2})"))
.collect::>()
.join(", ");
+let dynamic_filter_display =
+if let Ok(current) = self.dynamic_filter.current() {
+if !current.eq(&lit(true)) {
+format!(", filter=[{current}]")
+} else {
+"".to_string()
+}
+} else {
+"".to_string()
+};
Review Comment:
done!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2164714496
##
datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs:
##
@@ -433,6 +433,117 @@ async fn test_topk_dynamic_filter_pushdown() {
);
}
+#[tokio::test]
+async fn test_hashjoin_dynamic_filter_pushdown() {
Review Comment:
I've added a test that I think matches your suggestion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
xudong963 commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2162984236
##
datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs:
##
@@ -433,6 +433,117 @@ async fn test_topk_dynamic_filter_pushdown() {
);
}
+#[tokio::test]
+async fn test_hashjoin_dynamic_filter_pushdown() {
Review Comment:
Such test can check
1. dynamic filters are pushed down to right scan node
2. dynamic filters aren't missed during pushdown
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
xudong963 commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2162979272
##
datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs:
##
@@ -433,6 +433,117 @@ async fn test_topk_dynamic_filter_pushdown() {
);
}
+#[tokio::test]
+async fn test_hashjoin_dynamic_filter_pushdown() {
Review Comment:
Can we add some tests for multiple joins? Such as
```
Join (t1.a = t2.b)
/\
t1Join(t2.c = t3.d)
/\
t3 t2
```
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -666,10 +679,25 @@ impl DisplayAs for HashJoinExec {
.map(|(c1, c2)| format!("({c1}, {c2})"))
.collect::>()
.join(", ");
+let dynamic_filter_display =
+if let Ok(current) = self.dynamic_filter.current() {
+if !current.eq(&lit(true)) {
+format!(", filter=[{current}]")
+} else {
+"".to_string()
+}
+} else {
+"".to_string()
+};
Review Comment:
How about this:
```rust
let dynamic_filter_display = match self.dynamic_filter.current() {
Ok(current) if current != lit(true) => format!(", filter=[{current}]"),
_ => "".to_string(),
};
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-2991563630 > It's hard to say generally, but a hashtable lookup which fits into cache on a `u64` key can be really fast. I guess only benchmarks can tell. But I still think the scalar bounds are worth keeping for stats pruning reasons. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
Dandandan commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-2991531016 > > I think it makes sense to only filter on the shared hashmap and not bothering with the min/max values - creating hashes and doing a single table lookup is quite fast, so I think we want to avoid to also evaluate the min/max expression (at least for all rows) > > I'm surprised that the hash table lookup, even if O(1), has such a small constant factor that its ~ a couple of binary comparisons. That said a reason to still do both is stats and filter caching: simple filters like `col >= 123 and col <= 456` can be used for stats pruning and can easily be cached (for example for [filter caching based indexing](https://github.com/apache/datafusion/issues/15585)). So even if performance is not strictly better there is still something to be said for including a simple filter in addition to the hash table lookup. It's hard to say generally, but a hashtable lookup which fits into cache on a `u64` key can be really fast. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-2989099023 > I think it also makes sense to also thing about a heuristic we want to use to use this pushdown only when we think it might be useful - e.g. the left side is much smaller than the right side, or we know (based on column statistics) it will filter out rows Datafusion is generally _not great_ at these things: we often don't have enough stats / info to make decisions like this. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-2985988638 > I think it makes sense to only filter on the shared hashmap and not bothering with the min/max values - creating hashes and doing a single table lookup is quite fast, so I think we want to avoid to also evaluate the min/max expression (at least for all rows) I'm surprised that the hash table lookup, even if O(1), has such a small constant factor that its ~ a couple of binary comparisons. That said a reason to still do both is stats and filter caching: simple filters like `col >= 123 and col <= 456` can be used for stats pruning and can easily be cached (for example for [filter caching based indexing](https://github.com/apache/datafusion/issues/15585)). So even if performance is not strictly better there is still something to be said for including a simple filter in addition to the hash table lookup. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
Dandandan commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-2985881381 > > I think doing only the lookup is preferable above also computing / checking the bounds, I think the latter might create more overhead > > My thought was that for some cases the bounds checks are going to be quite effective at pruning and they should always be cheap to compute and cheap to apply. I'm surprised you say that they might create a lot of overhead? Maybe I should articulate it a bit more. * If we are only filtering out based on statistics, min/max might make sense to quickly filter out. * If we are filtering on values (e.g. filter pushdown) - I think it makes sense to *only* filter on the shared hashmap and not bothering with the min/max values - creating hashes and doing a single table lookup is relatively slow, so I think we want to avoid *also* evaluating the min/max expression. I think it also makes sense to also thing about a heuristic we want to use to use this pushdown only when we think it might be useful - e.g. the left side is much smaller than the right side, or we know (based on column statistics) it will fiflter out rows. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
Dandandan commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2155602331
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -943,10 +978,71 @@ impl ExecutionPlan for HashJoinExec {
try_embed_projection(projection, self)
}
}
+
+fn gather_filters_for_pushdown(
+&self,
+phase: FilterPushdownPhase,
+parent_filters: Vec>,
+config: &datafusion_common::config::ConfigOptions,
+) -> Result {
+// Don't allow parent filters to be pushed down for now
+// Only add our dynamic filter during the Post phase
+if !matches!(phase, FilterPushdownPhase::Post) {
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters));
+}
+
+// Only push down dynamic filters if enabled
+if config.optimizer.enable_dynamic_filter_pushdown {
+let filter = Arc::clone(&self.dynamic_filter) as Arc;
+// Push the dynamic filter to the right side (probe side) only
+// Left side (build side) gets empty vec, right side gets the
filter
+let filters_for_children = vec![vec![], vec![filter]];
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters)
+.with_self_filters_for_children(filters_for_children));
+}
+
+Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters))
+}
+}
+
+/// Compute min/max bounds for each column in the given arrays
+fn compute_bounds(arrays: &[ArrayRef]) -> Result> {
+arrays
+.iter()
+.map(|array| {
+if array.is_empty() {
+// Return NULL values for empty arrays
+return Ok((
+ScalarValue::try_from(array.data_type())?,
+ScalarValue::try_from(array.data_type())?,
+));
+}
+
+// Compute min/max using ScalarValue's utilities
+let mut min_val = ScalarValue::try_from_array(array, 0)?;
Review Comment:
makes sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2155025406
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -943,10 +978,71 @@ impl ExecutionPlan for HashJoinExec {
try_embed_projection(projection, self)
}
}
+
+fn gather_filters_for_pushdown(
+&self,
+phase: FilterPushdownPhase,
+parent_filters: Vec>,
+config: &datafusion_common::config::ConfigOptions,
+) -> Result {
+// Don't allow parent filters to be pushed down for now
+// Only add our dynamic filter during the Post phase
+if !matches!(phase, FilterPushdownPhase::Post) {
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters));
+}
+
+// Only push down dynamic filters if enabled
+if config.optimizer.enable_dynamic_filter_pushdown {
+let filter = Arc::clone(&self.dynamic_filter) as Arc;
+// Push the dynamic filter to the right side (probe side) only
+// Left side (build side) gets empty vec, right side gets the
filter
+let filters_for_children = vec![vec![], vec![filter]];
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters)
+.with_self_filters_for_children(filters_for_children));
+}
+
+Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters))
+}
+}
+
+/// Compute min/max bounds for each column in the given arrays
+fn compute_bounds(arrays: &[ArrayRef]) -> Result> {
+arrays
+.iter()
+.map(|array| {
+if array.is_empty() {
+// Return NULL values for empty arrays
+return Ok((
+ScalarValue::try_from(array.data_type())?,
+ScalarValue::try_from(array.data_type())?,
+));
+}
+
+// Compute min/max using ScalarValue's utilities
+let mut min_val = ScalarValue::try_from_array(array, 0)?;
Review Comment:
Could we move these functions into `functions-aggreage-common`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2155006506
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -943,10 +978,71 @@ impl ExecutionPlan for HashJoinExec {
try_embed_projection(projection, self)
}
}
+
+fn gather_filters_for_pushdown(
+&self,
+phase: FilterPushdownPhase,
+parent_filters: Vec>,
+config: &datafusion_common::config::ConfigOptions,
+) -> Result {
+// Don't allow parent filters to be pushed down for now
+// Only add our dynamic filter during the Post phase
+if !matches!(phase, FilterPushdownPhase::Post) {
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters));
+}
+
+// Only push down dynamic filters if enabled
+if config.optimizer.enable_dynamic_filter_pushdown {
+let filter = Arc::clone(&self.dynamic_filter) as Arc;
+// Push the dynamic filter to the right side (probe side) only
+// Left side (build side) gets empty vec, right side gets the
filter
+let filters_for_children = vec![vec![], vec![filter]];
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters)
+.with_self_filters_for_children(filters_for_children));
+}
+
+Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters))
+}
+}
+
+/// Compute min/max bounds for each column in the given arrays
+fn compute_bounds(arrays: &[ArrayRef]) -> Result> {
+arrays
+.iter()
+.map(|array| {
+if array.is_empty() {
+// Return NULL values for empty arrays
+return Ok((
+ScalarValue::try_from(array.data_type())?,
+ScalarValue::try_from(array.data_type())?,
+));
+}
+
+// Compute min/max using ScalarValue's utilities
+let mut min_val = ScalarValue::try_from_array(array, 0)?;
Review Comment:
Sadly
[min_batch](https://github.com/pydantic/datafusion/blob/825020d124ca12421104e1b10d32ae8c2746d9b4/datafusion/functions-aggregate/src/min_max.rs#L581)
is not public and either way `functions-aggregate` is not a dependency of
`physical-plan`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2154970369
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -943,10 +978,71 @@ impl ExecutionPlan for HashJoinExec {
try_embed_projection(projection, self)
}
}
+
+fn gather_filters_for_pushdown(
+&self,
+phase: FilterPushdownPhase,
+parent_filters: Vec>,
+config: &datafusion_common::config::ConfigOptions,
+) -> Result {
+// Don't allow parent filters to be pushed down for now
+// Only add our dynamic filter during the Post phase
+if !matches!(phase, FilterPushdownPhase::Post) {
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters));
+}
+
+// Only push down dynamic filters if enabled
+if config.optimizer.enable_dynamic_filter_pushdown {
+let filter = Arc::clone(&self.dynamic_filter) as Arc;
+// Push the dynamic filter to the right side (probe side) only
+// Left side (build side) gets empty vec, right side gets the
filter
+let filters_for_children = vec![vec![], vec![filter]];
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters)
+.with_self_filters_for_children(filters_for_children));
+}
+
+Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters))
+}
+}
+
+/// Compute min/max bounds for each column in the given arrays
+fn compute_bounds(arrays: &[ArrayRef]) -> Result> {
+arrays
+.iter()
+.map(|array| {
+if array.is_empty() {
+// Return NULL values for empty arrays
+return Ok((
+ScalarValue::try_from(array.data_type())?,
+ScalarValue::try_from(array.data_type())?,
+));
+}
+
+// Compute min/max using ScalarValue's utilities
+let mut min_val = ScalarValue::try_from_array(array, 0)?;
Review Comment:
Maybe we re-use datafusion/functions-aggregate/src/min_max.rs? Seems like
there's a lot of complexity there related the types that we wouldn't want t to
re-implement
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
adriangb commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-2984788592 > I think doing only the lookup is preferable above also computing / checking the bounds, I think the latter might create more overhead My thought was that for some cases the bounds checks are going to be super effective and they'd never be expensive. I'm surprised you say that they might create a lot of overhead? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
Dandandan commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-2984782413 Sorry, misclicked a button. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
Dandandan commented on PR #16445: URL: https://github.com/apache/datafusion/pull/16445#issuecomment-2984776797 I tink we should also consider a heuristic for not evaluating the filter if it's not useful. Also I think doing only the lookup is preferable above also computing / checking the bounds, I think the latter might create more overhead. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
Dandandan closed pull request #16445: Add dynamic filter (bounds) pushdown to HashJoinExec URL: https://github.com/apache/datafusion/pull/16445 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] Add dynamic filter (bounds) pushdown to HashJoinExec [datafusion]
Dandandan commented on code in PR #16445:
URL: https://github.com/apache/datafusion/pull/16445#discussion_r2154953337
##
datafusion/physical-plan/src/joins/hash_join.rs:
##
@@ -943,10 +978,71 @@ impl ExecutionPlan for HashJoinExec {
try_embed_projection(projection, self)
}
}
+
+fn gather_filters_for_pushdown(
+&self,
+phase: FilterPushdownPhase,
+parent_filters: Vec>,
+config: &datafusion_common::config::ConfigOptions,
+) -> Result {
+// Don't allow parent filters to be pushed down for now
+// Only add our dynamic filter during the Post phase
+if !matches!(phase, FilterPushdownPhase::Post) {
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters));
+}
+
+// Only push down dynamic filters if enabled
+if config.optimizer.enable_dynamic_filter_pushdown {
+let filter = Arc::clone(&self.dynamic_filter) as Arc;
+// Push the dynamic filter to the right side (probe side) only
+// Left side (build side) gets empty vec, right side gets the
filter
+let filters_for_children = vec![vec![], vec![filter]];
+return Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters)
+.with_self_filters_for_children(filters_for_children));
+}
+
+Ok(FilterDescription::new_with_child_count(2)
+.all_parent_filters_unsupported(parent_filters))
+}
+}
+
+/// Compute min/max bounds for each column in the given arrays
+fn compute_bounds(arrays: &[ArrayRef]) -> Result> {
+arrays
+.iter()
+.map(|array| {
+if array.is_empty() {
+// Return NULL values for empty arrays
+return Ok((
+ScalarValue::try_from(array.data_type())?,
+ScalarValue::try_from(array.data_type())?,
+));
+}
+
+// Compute min/max using ScalarValue's utilities
+let mut min_val = ScalarValue::try_from_array(array, 0)?;
Review Comment:
I think we should arrow kernel for this (this is slow).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
