You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/08/20 00:49:56 UTC
(datafusion) branch gh-readonly-queue/main/pr-24417-47e47656eb4bb2f74ccf4504263657ebb3dda6fe created (now d7f65a9613)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-24417-47e47656eb4bb2f74ccf4504263657ebb3dda6fe
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at d7f65a9613 Apply same sparse routing for range partition dynamic filter as hash partition and restructure `build_filter` (#24417)
This branch includes the following new commits:
new d7f65a9613 Apply same sparse routing for range partition dynamic filter as hash partition and restructure `build_filter` (#24417)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: Apply same sparse routing for range partition dynamic filter as hash partition and restructure `build_filter` (#24417)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24417-47e47656eb4bb2f74ccf4504263657ebb3dda6fe
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit d7f65a9613f9eddd19867f519bb61ccc3f702d5b
Author: Peter Lee <pe...@gmail.com>
AuthorDate: Thu Aug 20 00:49:45 2026 +0000
Apply same sparse routing for range partition dynamic filter as hash partition and restructure `build_filter` (#24417)
## Which issue does this PR close?
per
https://github.com/apache/datafusion/pull/23854#discussion_r3758046872
> as a follow on it might be nice to break this function into smaller
functions (mostly so they can be documented more clearly) -- the
techniques here are quite clever
## Rationale for this change
1. Code in `SharedBuildAccumulator.build_filter` is too complicated and
too clever to understand, need to extract it into smaller pieces of
logic block.
2. While trying to refactor the code, I realized my previous pr
https://github.com/apache/datafusion/pull/23854 didn't follow the sparse
routing branches build that hash partitioning is using, because I think
I didnt notice that's a thing atm. so I decided to deliever the fix for
range partition along with the refactor, because then both range and
hash partition dynamic filter expression would have almost similar
building structure.
## What changes are included in this PR?
- Apply same sparse routing for range partition dynamic filter that hash
partition, so we can share more code among them, now there difference is
literally only at the top physical expression to decide which partition
should the `key` go.
- Extract `build_filter` in `shared_bounds` to
`build_collect_left_filter` and `build_partitioned_filter`, and add
comment explanation throughout the code. The most complicated part are
filter building fast-path(`all_partitions_canceled`,
`all_partitions_empty` and `one_non_empty_partition`) and sparse
branches filter building. btw, I skipped documenting the state
management for `PartitionStatus` because I think its code is already
self-explanatory.
## Are these changes tested?
- test mod share_bounds
- filter_pushdown test
- range_partition.slt
## Are there any user-facing changes?
no
---
.../tests/physical_optimizer/filter_pushdown.rs | 2 +-
.../src/joins/hash_join/shared_bounds.rs | 349 +++++++++++----------
.../sqllogictest/test_files/range_partitioning.slt | 6 +-
3 files changed, 184 insertions(+), 173 deletions(-)
diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
index 57d7dba295..3c6f554a94 100644
--- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
+++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
@@ -1378,7 +1378,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_range_partitioned() {
- RepartitionExec: partitioning=Range([a@0 ASC, b@1 ASC], [(aa, bb)], 2), input_partitions=1
- DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
- RepartitionExec: partitioning=Range([a@0 ASC, b@1 ASC], [(aa, bb)], 2), input_partitions=1
- - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ CASE range_partition WHEN 0 THEN a@0 >= aa AND a@0 <= aa AND b@1 >= ba AND b@1 <= ba AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}]) ELSE a@0 >= ab AND a@0 <= ab AND b@1 >= bb AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:ab,c1:bb}]) END ]
+ - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ CASE range_partition WHEN 0 THEN a@0 >= aa AND a@0 <= aa AND b@1 >= ba AND b@1 <= ba AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}]) WHEN 1 THEN a@0 >= ab AND a@0 <= ab AND b@1 >= bb AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:ab,c1:bb}]) ELSE false END ]
"
);
diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
index 77b327accf..1564ca115e 100644
--- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
@@ -611,188 +611,190 @@ impl SharedBuildAccumulator {
fn build_filter(&self, finalize_input: FinalizeInput) -> Result<()> {
match finalize_input {
- FinalizeInput::CollectLeft(partition) => match partition {
- PartitionStatus::Reported(partition_data) => {
+ FinalizeInput::CollectLeft(partition) => {
+ self.build_collect_left_filter(partition)
+ }
+ FinalizeInput::Partitioned(partitions) => {
+ self.build_partitioned_filter(partitions)
+ }
+ }
+ }
+
+ /// Builds the single global filter used by a collect-left join.
+ fn build_collect_left_filter(&self, partition: PartitionStatus) -> Result<()> {
+ match partition {
+ PartitionStatus::Reported(PartitionData {
+ bounds,
+ pushdown,
+ keys_have_null,
+ }) => {
+ let membership_expr = create_membership_predicate(
+ &self.on_right,
+ pushdown,
+ &HASH_JOIN_SEED,
+ self.probe_schema.as_ref(),
+ )?;
+ let bounds_expr = create_bounds_predicate(&self.on_right, &bounds);
+
+ if let Some(filter_expr) =
+ combine_membership_and_bounds(membership_expr, bounds_expr)
+ {
+ self.dynamic_filter.update(
+ self.preserve_probe_nulls(filter_expr, keys_have_null)?,
+ )?;
+ }
+ Ok(())
+ }
+ PartitionStatus::Pending => datafusion_common::internal_err!(
+ "attempted to finalize collect-left dynamic filter without reported build data"
+ ),
+ PartitionStatus::CanceledUnknown => datafusion_common::internal_err!(
+ "collect-left dynamic filter cannot finalize with canceled build data"
+ ),
+ }
+ }
+
+ /// Builds one routed probe-side filter from finalized partitioned build data.
+ /// Empty partitions reject their routed rows, while canceled partitions stay
+ /// permissive because their build contents are unknown.
+ fn build_partitioned_filter(&self, partitions: Vec<PartitionStatus>) -> Result<()> {
+ let mut partition_filters = Vec::with_capacity(partitions.len());
+ let mut real_partition_ids = Vec::new();
+ let mut empty_partition_ids = Vec::new();
+ let mut has_canceled_unknown = false;
+ let mut keys_have_null = false;
+
+ for (partition_id, partition) in partitions.into_iter().enumerate() {
+ match partition {
+ PartitionStatus::Reported(PartitionData {
+ pushdown: PushdownStrategy::Empty,
+ ..
+ }) => {
+ empty_partition_ids.push(partition_id);
+ partition_filters.push(lit(false));
+ }
+ PartitionStatus::Reported(PartitionData {
+ bounds,
+ pushdown,
+ keys_have_null: partition_keys_have_null,
+ }) => {
+ real_partition_ids.push(partition_id);
+ keys_have_null |= partition_keys_have_null;
let membership_expr = create_membership_predicate(
&self.on_right,
- partition_data.pushdown.clone(),
+ pushdown,
&HASH_JOIN_SEED,
self.probe_schema.as_ref(),
)?;
- let bounds_expr =
- create_bounds_predicate(&self.on_right, &partition_data.bounds);
-
- if let Some(filter_expr) =
+ let bounds_expr = create_bounds_predicate(&self.on_right, &bounds);
+ let then_expr =
combine_membership_and_bounds(membership_expr, bounds_expr)
- {
- self.dynamic_filter.update(self.preserve_probe_nulls(
- filter_expr,
- partition_data.keys_have_null,
- )?)?;
- }
- }
- PartitionStatus::Pending => {
- return datafusion_common::internal_err!(
- "attempted to finalize collect-left dynamic filter without reported build data"
- );
+ .unwrap_or_else(|| lit(true));
+ partition_filters.push(then_expr);
}
PartitionStatus::CanceledUnknown => {
+ has_canceled_unknown = true;
+ partition_filters.push(lit(true));
+ // A canceled partition's build content is unknown, so it
+ // may hold a NULL key.
+ keys_have_null = true;
+ }
+ PartitionStatus::Pending => {
return datafusion_common::internal_err!(
- "collect-left dynamic filter cannot finalize with canceled build data"
+ "attempted to finalize dynamic filter with pending partition"
);
}
- },
- FinalizeInput::Partitioned(partitions) => {
- let num_partitions = partitions.len();
- let mut partition_filters = Vec::with_capacity(num_partitions);
- let mut real_partition_ids = Vec::new();
- let mut empty_partition_ids = Vec::new();
- let mut has_canceled_unknown = false;
- let mut keys_have_null = false;
-
- for (partition_id, partition) in partitions.iter().enumerate() {
- match partition {
- PartitionStatus::Reported(partition)
- if matches!(partition.pushdown, PushdownStrategy::Empty) =>
- {
- empty_partition_ids.push(partition_id);
- partition_filters.push(lit(false));
- }
- PartitionStatus::Reported(partition) => {
- real_partition_ids.push(partition_id);
- keys_have_null |= partition.keys_have_null;
- let membership_expr = create_membership_predicate(
- &self.on_right,
- partition.pushdown.clone(),
- &HASH_JOIN_SEED,
- self.probe_schema.as_ref(),
- )?;
- let bounds_expr = create_bounds_predicate(
- &self.on_right,
- &partition.bounds,
- );
- let then_expr = combine_membership_and_bounds(
- membership_expr,
- bounds_expr,
- )
- .unwrap_or_else(|| lit(true));
- partition_filters.push(then_expr);
- }
- PartitionStatus::CanceledUnknown => {
- has_canceled_unknown = true;
- partition_filters.push(lit(true));
- // A canceled partition's build content is unknown, so it
- // may hold a NULL key.
- keys_have_null = true;
- }
- PartitionStatus::Pending => {
- return datafusion_common::internal_err!(
- "attempted to finalize dynamic filter with pending partition"
- );
- }
- }
- }
-
- let filter_expr = if has_canceled_unknown
- && real_partition_ids.is_empty()
- && empty_partition_ids.is_empty()
- {
- lit(true)
- } else if !has_canceled_unknown && real_partition_ids.is_empty() {
- lit(false)
- } else if !has_canceled_unknown
- && real_partition_ids.len() == 1
- && empty_partition_ids.len() + 1 == num_partitions
- {
- Arc::clone(&partition_filters[real_partition_ids[0]])
- } else if let Some(range_partitioning) = &self.probe_range_partitioning {
- // Range partitioning
- assert_or_internal_err!(
- partition_filters.len() == range_partitioning.partition_count(),
- "Dynamic filter partition count {} does not match Range partition count {}",
- partition_filters.len(),
- range_partitioning.partition_count()
- );
- let routing_range_expr = Arc::new(RangeExpr::try_new(
- self.on_right.clone(),
- range_partitioning,
- )?)
- as Arc<dyn PhysicalExpr>;
- let else_expr = partition_filters
- .pop()
- .expect("Range partitioning always has at least one partition");
-
- // CASE range_partition(key)
- // WHEN 0 THEN F0
- // WHEN 1 THEN F1
- // ...
- // ELSE Fn
- // END
- let when_then_expr = partition_filters
- .into_iter()
- .enumerate()
- .map(|(partition_id, then_expr)| {
- (
- lit(ScalarValue::UInt64(Some(partition_id as u64))),
- then_expr,
- )
- })
- .collect();
-
- Arc::new(CaseExpr::try_new(
- Some(routing_range_expr),
- when_then_expr,
- Some(else_expr),
- )?) as Arc<dyn PhysicalExpr>
- } else {
- // Hash partitioning
- let routing_hash_expr = Arc::new(HashExpr::new(
- self.on_right.clone(),
- self.repartition_random_state.clone(),
- "hash_repartition".to_string(),
- ))
- as Arc<dyn PhysicalExpr>;
- let modulo_expr = Arc::new(BinaryExpr::new(
- routing_hash_expr,
- Operator::Modulo,
- lit(ScalarValue::UInt64(Some(num_partitions as u64))),
- )) as Arc<dyn PhysicalExpr>;
-
- let mut when_then_branches = if has_canceled_unknown {
- empty_partition_ids
- .into_iter()
- .map(|partition_id| {
- (
- lit(ScalarValue::UInt64(Some(partition_id as u64))),
- lit(false),
- )
- })
- .collect::<Vec<_>>()
- } else {
- vec![]
- };
- when_then_branches.extend(real_partition_ids.into_iter().map(
- |partition_id| {
- (
- lit(ScalarValue::UInt64(Some(partition_id as u64))),
- Arc::clone(&partition_filters[partition_id]),
- )
- },
- ));
-
- Arc::new(CaseExpr::try_new(
- Some(modulo_expr),
- when_then_branches,
- Some(lit(has_canceled_unknown)),
- )?) as Arc<dyn PhysicalExpr>
- };
-
- self.dynamic_filter
- .update(self.preserve_probe_nulls(filter_expr, keys_have_null)?)?;
}
}
- Ok(())
+ let all_partitions_canceled = has_canceled_unknown
+ && real_partition_ids.is_empty()
+ && empty_partition_ids.is_empty();
+ let all_partitions_empty = !has_canceled_unknown && real_partition_ids.is_empty();
+ let one_non_empty_partition =
+ !has_canceled_unknown && real_partition_ids.len() == 1;
+
+ let filter_expr = if all_partitions_canceled {
+ // No build data is known, so filtering any probe row could discard a match.
+ lit(true)
+ } else if all_partitions_empty {
+ // No build row exists, so no probe row can match.
+ lit(false)
+ } else if one_non_empty_partition {
+ // Only one build partition contains rows, so its filter covers every
+ // possible probe match without routing.
+ Arc::clone(&partition_filters[real_partition_ids[0]])
+ } else {
+ // Builds the shared sparse `CASE` for partition filter routing.
+ // Without cancellation, omitted branches are known empty and safely fall
+ // through to `ELSE false`. With cancellation, omitted canceled partitions
+ // have unknown contents and must fall through to `ELSE true`, so known-empty
+ // partitions are emitted explicitly as false branches.
+ let mut branches = if has_canceled_unknown {
+ empty_partition_ids
+ .iter()
+ .map(|&partition_id| (lit(partition_id as u64), lit(false)))
+ .collect::<Vec<_>>()
+ } else {
+ vec![]
+ };
+ branches.extend(real_partition_ids.iter().map(|&partition_id| {
+ (
+ lit(partition_id as u64),
+ Arc::clone(&partition_filters[partition_id]),
+ )
+ }));
+
+ let routing_expr = if let Some(range_partitioning) =
+ &self.probe_range_partitioning
+ {
+ // Routes probe rows with the partition id selected by [`RangeExpr`].
+ // CASE range_partition(keys)
+ // WHEN empty_partition_id THEN false -- only when cancellation exists
+ // WHEN real_partition_id THEN F(real_partition_id)
+ // ...
+ // ELSE has_canceled_unknown
+ // END
+ assert_or_internal_err!(
+ partition_filters.len() == range_partitioning.partition_count(),
+ "Dynamic filter partition count {} does not match Range partition count {}",
+ partition_filters.len(),
+ range_partitioning.partition_count()
+ );
+ Arc::new(RangeExpr::try_new(
+ self.on_right.clone(),
+ range_partitioning,
+ )?) as Arc<dyn PhysicalExpr>
+ } else {
+ // Routes probe rows with the same `hash(keys) % partition_count` expression used
+ // by Hash repartitioning.
+ // CASE hash(keys) % partition_count
+ // WHEN empty_partition_id THEN false -- only when cancellation exists
+ // WHEN real_partition_id THEN F(real_partition_id)
+ // ...
+ // ELSE has_canceled_unknown
+ // END
+ let routing_hash_expr = Arc::new(HashExpr::new(
+ self.on_right.clone(),
+ self.repartition_random_state.clone(),
+ "hash_repartition".to_string(),
+ )) as Arc<dyn PhysicalExpr>;
+ Arc::new(BinaryExpr::new(
+ routing_hash_expr,
+ Operator::Modulo,
+ lit(partition_filters.len() as u64),
+ )) as Arc<dyn PhysicalExpr>
+ };
+
+ Arc::new(CaseExpr::try_new(
+ Some(routing_expr),
+ branches,
+ Some(lit(has_canceled_unknown)),
+ )?)
+ };
+
+ self.dynamic_filter
+ .update(self.preserve_probe_nulls(filter_expr, keys_have_null)?)
}
/// Keeps probe rows with a NULL key when the join semantics need them.
@@ -1180,6 +1182,11 @@ mod tests {
"Range routing must use RangeExpr"
);
assert_eq!(case.when_then_expr().len(), 3);
+ assert_literal_bool(&case.when_then_expr()[0].1, false);
+ assert_literal_bool(
+ case.else_expr().expect("expected permissive fallback"),
+ true,
+ );
let batch = RecordBatch::try_new(
test_probe_schema(),
@@ -1253,7 +1260,11 @@ mod tests {
let expr = current_expr(&acc);
let case = case_expr(&expr);
assert!(case.expr().is_some());
- assert_eq!(case.when_then_expr().len(), 3);
+ assert_eq!(case.when_then_expr().len(), 2);
+ assert_literal_bool(
+ case.else_expr().expect("expected permissive fallback"),
+ true,
+ );
let batch = RecordBatch::try_new(
probe_schema,
diff --git a/datafusion/sqllogictest/test_files/range_partitioning.slt b/datafusion/sqllogictest/test_files/range_partitioning.slt
index 326856a352..dd80fded93 100644
--- a/datafusion/sqllogictest/test_files/range_partitioning.slt
+++ b/datafusion/sqllogictest/test_files/range_partitioning.slt
@@ -1840,8 +1840,8 @@ SELECT range_key, SUM(value) FROM (
# TEST 48: Hash Join Dynamic Filter Pushdown on Compatible Range Inputs
# The Parquet-backed probe accepts the partition-routed dynamic filter. Matching
# Range split points keep build filter i aligned with probe partition i. The
-# build has rows only in partitions 0 and 2, so the runtime filter must route
-# with a four-way CASE rather than collapse to a single filter.
+# build has rows only in partitions 0 and 2, so the sparse runtime CASE includes
+# only those branches and lets empty partitions fall through to false.
##########
statement ok
@@ -1882,7 +1882,7 @@ JOIN range_partitioned p ON b.range_key = p.range_key;
Plan with Metrics
01)HashJoinExec: mode=Partitioned<slt:ignore>metrics=[output_rows=2,<slt:ignore>]
02)--DataSourceExec: <slt:ignore>file_type=parquet, predicate=range_key@0 = 5 OR range_key@0 = 20<slt:ignore>metrics=[output_rows=2,<slt:ignore>]
-03)--DataSourceExec: <slt:ignore>file_type=parquet, predicate=DynamicFilter [ CASE range_partition WHEN 0 THEN range_key@0 >= 5 AND range_key@0 <= 5 AND range_key@0 IN (SET) ([5]) WHEN 1 THEN false WHEN 2 THEN range_key@0 >= 20 AND range_key@0 <= 20 AND range_key@0 IN (SET) ([20]) ELSE false END ]<slt:ignore>metrics=[output_rows=2,<slt:ignore>]
+03)--DataSourceExec: <slt:ignore>file_type=parquet, predicate=DynamicFilter [ CASE range_partition WHEN 0 THEN range_key@0 >= 5 AND range_key@0 <= 5 AND range_key@0 IN (SET) ([5]) WHEN 2 THEN range_key@0 >= 20 AND range_key@0 <= 20 AND range_key@0 IN (SET) ([20]) ELSE false END ]<slt:ignore>metrics=[output_rows=2,<slt:ignore>]
query III
SELECT b.range_key, b.value, p.value
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org