jayzhan211 commented on code in PR #27092:
URL: https://github.com/apache/datafusion/pull/27092#discussion_r4492711310
##########
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:
@adriangb here's the setup for the repro above, since it needs data to run.
`fact` is written with bloom filters, which aren't on by default when writing,
so the pruning difference only shows on files that have them. The query itself
runs with default settings.
```sql
SET datafusion.execution.parquet.bloom_filter_on_write = true;
SET datafusion.execution.parquet.max_row_group_size = 100000;
COPY (SELECT ((value * 2654435761) % 1000000)::bigint * 2 AS k, value AS p
FROM generate_series(1, 5000000)) TO 'fact.parquet' STORED AS PARQUET;
COPY (SELECT (value % 40000)::bigint AS k, CASE WHEN value % 40000 = 501
THEN 'xa' ELSE 'b' END AS s, value AS v FROM generate_series(1, 20000000)) TO
'dim.parquet' STORED AS PARQUET;
```
Could you check that the single-partition filter stays an `IN (SET)` on this
data after the change?
--
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]
