adriangb commented on code in PR #27091:
URL: https://github.com/apache/datafusion/pull/27091#discussion_r4513482487
##########
datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs:
##########
@@ -1195,7 +1206,12 @@ mod tests {
.partition(
RecordBatch::try_new(schema, vec![Arc::clone(&build)])?,
|partition, batch| {
- maps[partition] = Some(hash_table(batch.column(0),
&random_state)?);
+ // A dense partition builds an array map, others a hash table.
+ maps[partition] = Some(if rng.random_bool(0.5) {
Review Comment:
Applied in 3e1673d18: the fuzz test keeps hash tables under
`force_hash_collisions`; it passes with and without the feature.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs:
##########
@@ -722,15 +722,7 @@ impl SharedBuildAccumulator {
}) => {
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 = members.map(|mut members| {
Review Comment:
Added `array_map_of`, `partitioned_array_map_and_hash_map_partitions_merge`
and `partitioned_range_array_map_partitions_merge` in 3e1673d18.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs:
##########
@@ -464,17 +454,24 @@ impl HashTableLookupExpr {
for (partition, map) in maps.iter().enumerate() {
// An empty partition holds no key, so its rows stay unmatched.
let Some(map) = map else { continue };
- let map: &Map = map;
- let Map::HashMap(map) = map else {
- return internal_err!(
- "routed lookups only support hash table partitions"
- );
- };
let run = starts[partition]..starts[partition + 1];
if run.is_empty() {
continue;
}
- let hits = map.contain_hashes(&sorted_hashes[run.clone()]);
+ let map: &Map = map;
+ let hits = match map {
+ Map::HashMap(map) =>
map.contain_hashes(&sorted_hashes[run.clone()]),
+ // An array map is keyed by value, not hash, so it probes
+ // the run's own key values.
+ Map::ArrayMap(map) => {
+ let indices =
UInt32Array::from(rows[run.clone()].to_vec());
Review Comment:
Left for a follow-up, as you suggested: this PR keeps hashing every row.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs:
##########
@@ -464,17 +454,24 @@ impl HashTableLookupExpr {
for (partition, map) in maps.iter().enumerate() {
// An empty partition holds no key, so its rows stay unmatched.
let Some(map) = map else { continue };
- let map: &Map = map;
- let Map::HashMap(map) = map else {
- return internal_err!(
- "routed lookups only support hash table partitions"
- );
- };
let run = starts[partition]..starts[partition + 1];
if run.is_empty() {
continue;
}
- let hits = map.contain_hashes(&sorted_hashes[run.clone()]);
+ let map: &Map = map;
+ let hits = match map {
+ Map::HashMap(map) =>
map.contain_hashes(&sorted_hashes[run.clone()]),
+ // An array map is keyed by value, not hash, so it probes
+ // the run's own key values.
+ Map::ArrayMap(map) => {
+ let indices =
UInt32Array::from(rows[run.clone()].to_vec());
+ let run_keys = keys
+ .iter()
+ .map(|key| take(key.as_ref(), &indices, None))
+ .collect::<std::result::Result<Vec<_>, _>>()?;
+ map.contain_keys(&run_keys)?
Review Comment:
Switched to `take_arrays` in 3e1673d18.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs:
##########
@@ -1154,6 +1153,18 @@ mod tests {
Ok(Arc::new(Map::HashMap(Box::new(map))))
}
+ /// An array map of `keys`, as a dense build side gets.
+ fn array_map(keys: &ArrayRef) -> Result<Arc<Map>> {
+ let keys = keys.as_primitive::<arrow::datatypes::Int32Type>();
+ let key = |value: Option<i32>| {
+ ArrayMap::key_to_u64(&ScalarValue::Int32(value)).expect("Int32
key")
+ };
+ let min = key(arrow::compute::min(keys));
+ let max = key(arrow::compute::max(keys));
+ let keys: ArrayRef = Arc::new(keys.clone());
+ Ok(Arc::new(Map::ArrayMap(ArrayMap::try_new(&keys, min, max)?)))
Review Comment:
Applied in 3e1673d18.
_🤖 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]
