adriangb commented on code in PR #27092:
URL: https://github.com/apache/datafusion/pull/27092#discussion_r4492753656
##########
datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs:
##########
@@ -692,175 +692,50 @@ impl SharedBuildAccumulator {
}
}
- /// 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.
+ /// Builds one probe-side filter from finalized partitioned build data.
+ /// Empty and canceled partitions contribute no key.
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;
- // Every non-empty partition's membership strategy and bounds, merged
into
- // one membership filter below; `None` once a partition cannot be
merged.
- let mut members = Some(Vec::with_capacity(partitions.len()));
+ let num_partitions = partitions.len();
+ let mut members = Vec::with_capacity(num_partitions);
let mut merged_bounds: Option<PartitionBounds> = None;
+ let mut keys_have_null = false;
for (partition_id, partition) in partitions.into_iter().enumerate() {
match partition {
- PartitionStatus::Reported(PartitionData {
+ // A canceled partition's stream emits no row: it is dropped
+ // before its report is delivered, and a stream probes only
+ // after delivery. So no key of its build side can matter.
+ PartitionStatus::CanceledUnknown
+ | 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 mergeable = match &pushdown {
- PushdownStrategy::Map(map, _) => {
- let map: &Map = map;
- matches!(map, Map::HashMap(_))
- }
- PushdownStrategy::InList(_) => true,
- PushdownStrategy::Empty => false,
- };
- members = members.filter(|_| mergeable).map(|mut members| {
- members.push((partition_id, pushdown.clone(),
bounds.clone()));
- members
- });
merged_bounds = Some(match merged_bounds {
None => bounds.clone(),
Some(merged) => merged.union(&bounds),
});
-
- 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);
- 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));
- // Rows routed here must pass, which a merged filter
cannot express.
- members = None;
- // A canceled partition's build content is unknown, so it
- // may hold a NULL key.
- keys_have_null = true;
+ members.push((partition_id, pushdown, bounds));
}
PartitionStatus::Pending => {
- return datafusion_common::internal_err!(
+ return internal_err!(
"attempted to finalize dynamic filter with pending
partition"
);
}
}
}
- 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 {
+ let filter_expr = if members.is_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 if let Some(members) = members {
- self.merged_membership_filter(
- &members,
- merged_bounds,
- partition_filters.len(),
- )?
} 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_with_schema(
- self.on_right.clone(),
- range_partitioning,
- &self.probe_schema,
- )?) 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.merged_membership_filter(&members, merged_bounds,
num_partitions)?
};
Review Comment:
Checked on e51f5ed83 with your setup at 1/10 scale (fact 500K rows, dim 2M
rows with `% 4000`, row groups of 10K, so 500 build rows still share one key).
The planner chose `CollectLeft` at that size, so I forced `Partitioned` with
`hash_join_single_partition_threshold` and
`hash_join_single_partition_threshold_rows` set to 0. The join is
`mode=Partitioned`, the probe filter is `k@0 >= 501 AND k@0 <= 501 AND k@0 IN
(SET) (...)`, and bloom filters prune all 50 row groups
(`row_groups_pruned_bloom_filter=50 total → 0 matched`).
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
--
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]
