adriangb commented on code in PR #27092:
URL: https://github.com/apache/datafusion/pull/27092#discussion_r4492721335
##########
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:
Applied your suggestion and added
`partitioned_one_non_empty_partition_keeps_its_own_filter` in e51f5ed83.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs:
##########
@@ -1893,10 +1597,10 @@ mod tests {
);
}
- /// A canceled partition's build content is unknown, so it may hold a NULL
key:
- /// the aggregated flag must be permissive even when no partition reported
one.
+ /// A canceled partition holds no key, so it cannot add a NULL key that
+ /// would make a null-equal join keep probe-side NULL rows.
#[test]
- fn partitioned_canceled_partition_widens_routed_filter() {
+ fn partitioned_canceled_partition_does_not_widen_filter() {
Review Comment:
Added `partitioned_canceled_partition_rejects_only_its_routed_rows` in
e51f5ed83.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs:
##########
@@ -352,8 +353,7 @@ impl HashTableLookupExpr {
/// for pruning only, or `None`
///
/// A single non-empty map is probed directly without routing, so it may be
- /// any [`Map`] and `router` is ignored. Otherwise `router` is required and
- /// every map must be a [`Map::HashMap`].
+ /// any [`Map`] and `router` is ignored. Otherwise `router` is required.
Review Comment:
Applied the suggested wording in e51f5ed83.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs:
##########
@@ -18,7 +18,7 @@ use crate::joins::hash_join::exec::HASH_JOIN_SEED;
use crate::joins::hash_join::inlist_builder::build_struct_fields;
use crate::joins::hash_join::not_in::NotIn;
use crate::joins::hash_join::partitioned_hash_eval::{
- HashExpr, HashTableLookupExpr, PartitionRouter, SeededRandomState,
+ HashTableLookupExpr, PartitionRouter, SeededRandomState,
Review Comment:
Added a doc line in e51f5ed83: `HashExpr` stays so that `datafusion-proto`
can decode older serialized plans.
_🤖 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]
