jayzhan211 commented on code in PR #27092:
URL: https://github.com/apache/datafusion/pull/27092#discussion_r4492670718
##########
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:
Optional cleanup: this was the last place that built a `HashExpr`, so only
proto decoding and tests use it now. Could you either open a follow-up to
remove it, or add a comment saying it stays so older serialized plans still
decode?
##########
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:
Removing the single-partition branch changes behavior, and it costs bloom
filter pruning. `[None, Some(map), None]` doesn't match `(_, [Some(map)])`, so
it gets routed, and the merged IN list limit counts duplicate rows, so one key
with more than 150 build rows becomes `hash_lookup`.
```sql
-- default config, plans Partitioned: 500 dim rows pass, all with one key,
which is absent from fact
SELECT count(*) FROM dim d JOIN fact ON d.k = fact.k WHERE d.s LIKE 'x%';
```
| | main | this PR | with this change |
|---|---|---|---|
| Filter | `IN (SET)` | `hash_lookup` | `IN (SET)` |
| Row groups kept by bloom | 5 of 50 | no bloom pruning | 5 of 50 |
| Rows scanned | 500K | 5M | 500K |
| Bytes scanned | 7.9 MB | 16.2 MB | 7.9 MB |
Routing also costs more when the probes land in that partition: filter
evaluation was about 3.5x slower with a hash table and 7-9x with an `ArrayMap`,
under a noisy load. I ran this change with the touched tests:
````suggestion
let filter_expr = match members.as_slice() {
// No build row exists, so no probe row can match.
[] => lit(false),
// Only one build partition holds rows, so its own filter covers
every
// probe match without routing, and an IN list keeps its pruning.
[(_, pushdown, bounds)] => combine_membership_and_bounds(
create_membership_predicate(
&self.on_right,
pushdown.clone(),
&HASH_JOIN_SEED,
self.probe_schema.as_ref(),
)?,
create_bounds_predicate(&self.on_right, bounds),
)
.expect("membership predicate is always present"),
_ => {
self.merged_membership_filter(&members, merged_bounds,
num_partitions)?
}
};
````
Could you also add this test? It fails without the change, because both
shapes get routed:
```rs
/// One non-empty partition keeps its own filter: no routing, and an IN list
/// stays an IN list even when its duplicate rows exceed the merged list
limit.
#[test]
fn partitioned_one_non_empty_partition_keeps_its_own_filter() {
let empty = || reported(PushdownStrategy::Empty, no_bounds());
let mut acc = make_partitioned_expr_accumulator_for_test(3);
acc.in_list_max_values = 150;
acc.build_filter(FinalizeInput::Partitioned(vec![
empty(),
reported(in_list(&[2; 1000]), no_bounds()),
PartitionStatus::CanceledUnknown,
]))
.unwrap();
in_list_expr(¤t_expr(&acc));
let acc = make_partitioned_expr_accumulator_for_test(3);
acc.build_filter(FinalizeInput::Partitioned(vec![
empty(),
reported(hash_map(), bounds(1, 7)),
PartitionStatus::CanceledUnknown,
]))
.unwrap();
let expr = format!("{:?}", current_expr(&acc));
assert!(!expr.contains("routed by"), "{expr}");
}
```
##########
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:
Optional cleanup: this reads as if one non-empty map skips routing, but only
a one-entry `maps` does.
```suggestion
/// When `maps` has exactly one entry, that map is probed directly
without
/// routing and `router` is ignored. Otherwise `router` is required,
even if
/// only one entry is non-empty.
```
##########
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:
Nothing here covers a canceled partition next to live ones anymore, since
the range test now has only live and empty partitions. Could you add this one?
It passes on this branch and fails on main, where those rows are kept:
```rs
/// A canceled partition rejects the rows routed to it, and only those: the
/// live partitions keep their matches.
#[cfg(not(feature = "force_hash_collisions"))]
#[test]
fn partitioned_canceled_partition_rejects_only_its_routed_rows() {
let acc = make_partitioned_expr_accumulator_for_test(4);
let keys: Vec<i32> = (0..40).collect();
let mut partitions =
routed_partitions(&acc, 4, &keys, |_, held| hash_map_of(held));
partitions[0] = PartitionStatus::CanceledUnknown;
acc.build_filter(FinalizeInput::Partitioned(partitions))
.unwrap();
let array: ArrayRef = Arc::new(Int32Array::from(keys.clone()));
let mut hashes = vec![0; keys.len()];
create_hashes(
[&array],
acc.repartition_random_state.random_state(),
&mut hashes,
)
.unwrap();
let expected: Vec<bool> = hashes.iter().map(|hash| *hash % 4 !=
0).collect();
assert!(expected.contains(&true) && expected.contains(&false));
assert_matches(&acc, &keys, &expected);
}
```
--
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]
