Skip to content

Commit 542e11a

Browse files
gruuyaclaude
andcommitted
Prune hash join build side via min/max, not bloom filters
Exposes HashTableLookupExpr's single-column build values through the same compact sorted-domain rewrite ordinary large IN-lists already get (PrimitiveInListPruningExpr/StringInListPruningExpr), so row-group/file min/max stats alone can exclude containers - no bloom filter fetch, no LiteralGuarantee. On by default via hash_join_dynamic_pruning_max_distinct_values (default 100_000); set to 0 to disable. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1 parent 7b00b63 commit 542e11a

9 files changed

Lines changed: 456 additions & 21 deletions

File tree

‎datafusion/common/src/config.rs‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1844,6 +1844,26 @@ config_namespace! {
18441844
/// See: <https://trino.io/docs/current/admin/dynamic-filtering.html#dynamic-filter-collection-thresholds>
18451845
pub hash_join_inlist_pushdown_max_distinct_values: usize, default = 150
18461846

1847+
/// Maximum number of distinct build-side values to retain for row-group/file
1848+
/// min/max-based pruning once the build side is too large for `InList` pushdown
1849+
/// and falls back to an opaque hash-table-lookup filter. Set to 0 to disable.
1850+
///
1851+
/// On by default: the check is footer-only (no bloom filter or extra I/O),
1852+
/// reusing the sorted-domain rewrite an ordinary large `IN (...)` list already
1853+
/// gets, so a container is kept only if its own min/max overlaps a value.
1854+
///
1855+
/// When engaged, `EXPLAIN`'s `pruning_predicate=` gains an extra
1856+
/// `IN_SET_INTERSECTS(<col>_min, <col>_max, <n> values)` clause - the visible
1857+
/// sign this ran, distinct from the plain min/max bounds every join pushes down.
1858+
pub hash_join_dynamic_pruning_max_distinct_values: usize, default = 100_000
1859+
1860+
/// Companion size cap (bytes) for `hash_join_dynamic_pruning_max_distinct_values`,
1861+
/// mirroring `hash_join_inlist_pushdown_max_size`. Set to 0 to disable.
1862+
///
1863+
/// Checked against the *raw*, undeduplicated build-side column, so this also
1864+
/// guards against few distinct values but many duplicate rows.
1865+
pub hash_join_dynamic_pruning_max_size: usize, default = 8 * 1024 * 1024
1866+
18471867
/// The default filter selectivity used by Filter Statistics
18481868
/// when an exact selectivity cannot be determined. Valid values are
18491869
/// between 0 (no selectivity) and 100 (all rows are selected).

‎datafusion/common/src/scalar/mod.rs‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4314,6 +4314,20 @@ impl ScalarValue {
43144314
})
43154315
}
43164316

4317+
/// Converts `array`'s values into non-null [`ScalarValue`]s, preserving duplicates,
4318+
/// returning `None` if any value fails to convert (e.g. an unsupported type).
4319+
pub fn nonnull_scalars(array: &dyn Array) -> Option<Vec<ScalarValue>> {
4320+
let mut values = Vec::with_capacity(array.len());
4321+
for i in 0..array.len() {
4322+
match ScalarValue::try_from_array(array, i) {
4323+
Ok(v) if !v.is_null() => values.push(v),
4324+
Ok(_) => {} // NULL never satisfies `=`; harmless to drop
4325+
Err(_) => return None,
4326+
}
4327+
}
4328+
Some(values)
4329+
}
4330+
43174331
/// Try to parse `value` into a ScalarValue of type `target_type`
43184332
pub fn try_from_string(value: String, target_type: &DataType) -> Result<Self> {
43194333
ScalarValue::from(value).cast_to(target_type)
@@ -6051,6 +6065,28 @@ mod tests {
60516065
use insta::assert_snapshot;
60526066
use rand::Rng;
60536067

6068+
#[test]
6069+
fn test_nonnull_scalars() {
6070+
let array = Int32Array::from(vec![Some(1), Some(2), Some(1), None, Some(2)]);
6071+
let values = ScalarValue::nonnull_scalars(&array).expect("convertible");
6072+
assert_eq!(
6073+
values,
6074+
vec![
6075+
ScalarValue::Int32(Some(1)),
6076+
ScalarValue::Int32(Some(2)),
6077+
ScalarValue::Int32(Some(1)),
6078+
ScalarValue::Int32(Some(2)),
6079+
]
6080+
);
6081+
}
6082+
6083+
#[test]
6084+
fn test_nonnull_scalars_all_null() {
6085+
let array = Int32Array::from(vec![None, None]);
6086+
let values = ScalarValue::nonnull_scalars(&array).expect("convertible");
6087+
assert!(values.is_empty());
6088+
}
6089+
60546090
#[test]
60556091
fn test_scalar_value_from_for_map() {
60566092
let string_builder = StringBuilder::new();

‎datafusion/physical-plan/src/joins/hash_join/exec.rs‎

Lines changed: 27 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -2961,19 +2961,37 @@ async fn collect_left_input(
29612961
.iter()
29622962
.map(|arr| arr.get_array_memory_size())
29632963
.sum::<usize>();
2964-
if left_values.is_empty()
2965-
|| left_values[0].is_empty()
2966-
|| estimated_size > config.optimizer.hash_join_inlist_pushdown_max_size
2967-
|| map.num_of_distinct_key()
2968-
> config
2964+
2965+
let pushdown_inlist = !left_values.is_empty()
2966+
&& !left_values[0].is_empty()
2967+
&& estimated_size <= config.optimizer.hash_join_inlist_pushdown_max_size
2968+
&& map.num_of_distinct_key()
2969+
<= config
29692970
.optimizer
2970-
.hash_join_inlist_pushdown_max_distinct_values
2971+
.hash_join_inlist_pushdown_max_distinct_values;
2972+
2973+
if pushdown_inlist
2974+
&& let Some(in_list_values) = build_struct_inlist_values(&left_values)?
29712975
{
2972-
PushdownStrategy::Map(Arc::clone(&map))
2973-
} else if let Some(in_list_values) = build_struct_inlist_values(&left_values)? {
29742976
PushdownStrategy::InList(in_list_values)
29752977
} else {
2976-
PushdownStrategy::Map(Arc::clone(&map))
2978+
// Past the InList threshold, retain raw values for pruning only (not row
2979+
// filtering) up to a separate, more generous cap; dedup happens lazily
2980+
// inside `HashTableLookupExpr` on first actual use, not eagerly here.
2981+
let pushdown_values = !left_values.is_empty()
2982+
&& !left_values[0].is_empty()
2983+
&& estimated_size <= config.optimizer.hash_join_dynamic_pruning_max_size
2984+
&& map.num_of_distinct_key()
2985+
<= config
2986+
.optimizer
2987+
.hash_join_dynamic_pruning_max_distinct_values;
2988+
2989+
if pushdown_values {
2990+
let pruning_literals = build_struct_inlist_values(&left_values)?;
2991+
PushdownStrategy::Map(Arc::clone(&map), pruning_literals)
2992+
} else {
2993+
PushdownStrategy::Map(Arc::clone(&map), None)
2994+
}
29772995
}
29782996
};
29792997

‎datafusion/physical-plan/src/joins/hash_join/inlist_builder.rs‎

Lines changed: 43 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,9 @@
1919
2020
use std::sync::Arc;
2121

22-
use arrow::array::{ArrayRef, StructArray};
23-
use arrow::datatypes::{Field, FieldRef, Fields};
22+
use arrow::array::{ArrayRef, AsArray, StructArray};
23+
use arrow::compute::cast;
24+
use arrow::datatypes::{Field, FieldRef, Fields, Int32Type};
2425
use arrow_schema::DataType;
2526
use datafusion_common::Result;
2627

@@ -77,6 +78,25 @@ pub(super) fn build_struct_inlist_values(
7778
Ok(Some(source_array))
7879
}
7980

81+
/// Deduplicates `array`'s values (dropping nulls) via Arrow's dictionary-encoding cast,
82+
/// cheaper than converting every row to a `ScalarValue` to test uniqueness.
83+
///
84+
/// Returns `None` if the type isn't dictionary-encodable or every value is null -
85+
/// both safe to treat as "no pruning literals available".
86+
pub(super) fn dedupe_array_values(array: &ArrayRef) -> Option<ArrayRef> {
87+
let dict_type = DataType::Dictionary(
88+
Box::new(DataType::Int32),
89+
Box::new(array.data_type().clone()),
90+
);
91+
let dict = cast(array.as_ref(), &dict_type).ok()?;
92+
let values = Arc::clone(dict.as_dictionary::<Int32Type>().values());
93+
if values.is_empty() {
94+
None
95+
} else {
96+
Some(values)
97+
}
98+
}
99+
80100
#[cfg(test)]
81101
mod tests {
82102
use super::*;
@@ -142,6 +162,27 @@ mod tests {
142162
);
143163
}
144164

165+
#[test]
166+
fn test_dedupe_array_values() {
167+
let array = Arc::new(Int32Array::from(vec![
168+
Some(1),
169+
Some(2),
170+
Some(1),
171+
None,
172+
Some(2),
173+
])) as ArrayRef;
174+
let deduped = dedupe_array_values(&array).expect("dictionary-encodable");
175+
assert_eq!(deduped.len(), 2);
176+
let deduped = deduped.as_ref().as_primitive::<Int32Type>();
177+
assert_eq!(deduped.values(), &[1, 2]);
178+
}
179+
180+
#[test]
181+
fn test_dedupe_array_values_all_null() {
182+
let array = Arc::new(Int32Array::from(vec![None, None])) as ArrayRef;
183+
assert!(dedupe_array_values(&array).is_none());
184+
}
185+
145186
#[test]
146187
fn test_build_single_column_dictionary_inlist() {
147188
let keys = Int8Array::from(vec![0i8, 0, 0]);

0 commit comments

Comments
 (0)