diff --git a/datafusion/physical-plan/src/joins/hash_join/inlist_builder.rs b/datafusion/physical-plan/src/joins/hash_join/inlist_builder.rs index 2fc3201c6363..f1fb49002096 100644 --- a/datafusion/physical-plan/src/joins/hash_join/inlist_builder.rs +++ b/datafusion/physical-plan/src/joins/hash_join/inlist_builder.rs @@ -17,10 +17,13 @@ //! Utilities for building InList expressions from hash join build side data +use std::collections::HashSet; use std::sync::Arc; -use arrow::array::{ArrayRef, StructArray}; +use arrow::array::{Array, ArrayRef, StructArray, UInt32Array}; +use arrow::compute::take; use arrow::datatypes::{Field, FieldRef, Fields}; +use arrow::row::{Row, RowConverter, SortField}; use arrow_schema::DataType; use datafusion_common::Result; @@ -43,8 +46,15 @@ pub(super) fn build_struct_fields(data_types: &[DataType]) -> Result { /// that is: this will produce `IN LIST ((1, "a"), (2, "b"))` expected to be used as `(2, "b") IN LIST ((1, "a"), (2, "b"))`. /// The field names of the struct are auto-generated as "c0", "c1", ... and should match the struct expression used in the join keys. /// -/// Note that this function does not deduplicate values - deduplication will happen later -/// when building an InList expression from this array via `InListExpr::try_new_from_array`. +/// The returned array is deduplicated (see [`dedup_inlist_values`]): the build side is +/// gated on its *distinct* key count, not its row count, so the raw key arrays routinely +/// contain far more rows than distinct values. An `IN` list is a set, so dropping the +/// duplicates cannot change any result, but it does shrink everything downstream that is +/// sized by the list length: the per-value [`ScalarValue`] literals built by +/// `InListExpr::try_new_from_array`, and — wherever the pushed-down filter reaches a +/// `PruningPredicate` — the `LiteralGuarantee` it materializes per row group. +/// +/// [`ScalarValue`]: datafusion_common::ScalarValue /// /// Returns `None` if the estimated size exceeds `max_size_bytes` or if the number of rows /// exceeds `max_distinct_values`. @@ -74,15 +84,57 @@ pub(super) fn build_struct_inlist_values( Arc::new(StructArray::from(arrays_with_fields)) }; - Ok(Some(source_array)) + Ok(Some(dedup_inlist_values(source_array)?)) +} + +/// Removes duplicate entries from an `IN` list value array, preserving first-occurrence order. +/// +/// Equality is the arrow row-format byte equality produced by [`RowConverter`], which is +/// exactly the equality used elsewhere in DataFusion for grouping. In particular NULLs +/// compare equal to each other, so a list with many NULLs collapses to a single NULL. That +/// is semantics-preserving for `IN`: one NULL in the haystack already yields the same +/// three-valued result as a thousand. +/// +/// The input is returned unchanged when it holds fewer than two rows, when it contains no +/// duplicates, or when its type cannot be row-encoded (dedup is an optimization, never a +/// requirement). +fn dedup_inlist_values(values: ArrayRef) -> Result { + if values.len() < 2 { + return Ok(values); + } + + let sort_field = SortField::new(values.data_type().clone()); + if !RowConverter::supports_fields(std::slice::from_ref(&sort_field)) { + return Ok(values); + } + + let converter = RowConverter::new(vec![sort_field])?; + let rows = converter.convert_columns(std::slice::from_ref(&values))?; + + let mut seen: HashSet = HashSet::with_capacity(values.len()); + let mut indices: Vec = Vec::new(); + for (idx, row) in rows.iter().enumerate() { + if seen.insert(row) { + indices.push(idx as u32); + } + } + + if indices.len() == values.len() { + // Nothing to remove: skip the copy. + return Ok(values); + } + + Ok(take(values.as_ref(), &UInt32Array::from(indices), None)?) } #[cfg(test)] mod tests { use super::*; use arrow::array::{ - DictionaryArray, Int8Array, Int32Array, StringArray, StringDictionaryBuilder, + AsArray, DictionaryArray, Int8Array, Int32Array, StringArray, + StringDictionaryBuilder, }; + use arrow::datatypes::Int32Type; #[test] fn test_build_single_column_inlist_array() { @@ -91,9 +143,41 @@ mod tests { .unwrap() .unwrap(); + // Duplicates are dropped, first-occurrence order is preserved. + let expected = Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef; + assert!(expected.eq(&result)); + } + + #[test] + fn test_build_single_column_inlist_array_without_duplicates() { + let array = Arc::new(Int32Array::from(vec![3, 1, 2])) as ArrayRef; + let result = build_struct_inlist_values(std::slice::from_ref(&array)) + .unwrap() + .unwrap(); + assert!(array.eq(&result)); } + #[test] + fn test_build_single_column_inlist_array_with_nulls() { + let array = Arc::new(Int32Array::from(vec![ + Some(1), + None, + Some(2), + None, + Some(1), + ])) as ArrayRef; + let result = build_struct_inlist_values(std::slice::from_ref(&array)) + .unwrap() + .unwrap(); + + // A single NULL is kept: `IN` is a set, and one NULL in the haystack already + // produces the same three-valued result as many. + let expected = + Arc::new(Int32Array::from(vec![Some(1), None, Some(2)])) as ArrayRef; + assert!(expected.eq(&result)); + } + #[test] fn test_build_multi_column_inlist() { let array1 = Arc::new(Int32Array::from(vec![1, 2, 3, 2, 1])) as ArrayRef; @@ -110,6 +194,34 @@ mod tests { build_struct_fields(&[DataType::Int32, DataType::Utf8]).unwrap() ) ); + // Deduplication is on the whole tuple, not per column. + assert_eq!(result.len(), 3); + let struct_array = result.as_struct(); + assert_eq!( + struct_array.column(0).as_primitive::().values(), + &[1, 2, 3] + ); + assert_eq!( + struct_array + .column(1) + .as_string::() + .iter() + .collect::>(), + vec![Some("a"), Some("b"), Some("c")] + ); + } + + #[test] + fn test_build_multi_column_inlist_distinct_tuples_from_duplicate_columns() { + // Each column on its own has duplicates, but every (a, b) tuple is distinct. + let array1 = Arc::new(Int32Array::from(vec![1, 1, 2, 2])) as ArrayRef; + let array2 = Arc::new(StringArray::from(vec!["a", "b", "a", "b"])) as ArrayRef; + + let result = build_struct_inlist_values(&[array1, array2]) + .unwrap() + .unwrap(); + + assert_eq!(result.len(), 4); } #[test] @@ -152,7 +264,9 @@ mod tests { .unwrap() .unwrap(); - assert_eq!(result.len(), 3); + // All three rows decode to "foo", so the list collapses to one entry while + // keeping the dictionary encoding. + assert_eq!(result.len(), 1); assert_eq!(result.data_type(), dict_array.data_type()); } } diff --git a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt index 72d034067663..db2ef24622ae 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt @@ -1336,3 +1336,57 @@ RESET datafusion.execution.parquet.pushdown_filters; statement ok drop table t; + + +######## +# The InList pushdown is gated on the build side's *distinct* key count, but the +# values it ships come from the raw build-side key arrays (one entry per build +# row). Duplicated build keys must not inflate the pushed-down `IN (SET)` list or +# the `required_guarantees` derived from it. +######## + +statement ok +COPY (SELECT * FROM (VALUES (11), (11), (11), (22), (22), (22)) v(id)) TO 'test_files/scratch/push_down_filter_parquet/dedup_build.parquet' STORED AS PARQUET; + +statement ok +COPY (SELECT * FROM (VALUES (11), (22), (33), (44), (55), (66), (77), (88), (99), (110), (121), (132)) v(id)) TO 'test_files/scratch/push_down_filter_parquet/dedup_probe.parquet' STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE dedup_build STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/dedup_build.parquet'; + +statement ok +CREATE EXTERNAL TABLE dedup_probe STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/dedup_probe.parquet'; + +# Data-correctness: deduplicating the pushed-down set must not drop join output +# rows, which still repeat once per matching build row. +query II rowsort +SELECT dedup_build.id, dedup_probe.id FROM dedup_build JOIN dedup_probe ON dedup_build.id = dedup_probe.id +---- +11 11 +11 11 +11 11 +22 22 +22 22 +22 22 + +statement ok +set datafusion.explain.analyze_categories = 'rows'; + +# The six build rows carry only two distinct keys, so the pushed-down filter is +# `IN (SET) ([11, 22])`, not `([11, 11, 11, 22, 22, 22])`. +query TT +EXPLAIN ANALYZE SELECT dedup_build.id, dedup_probe.id FROM dedup_build JOIN dedup_probe ON dedup_build.id = dedup_probe.id +---- +Plan with Metrics +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], metrics=[output_rows=6, output_batches=1, array_map_created_count=1, build_input_batches=1, build_input_rows=6, input_batches=1, input_rows=12, avg_fanout=300% (6/2), probe_hit_rate=16.67% (2/12)] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/dedup_build.parquet]]}, projection=[id], file_type=parquet, metrics=[output_rows=6, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, predicate_cache_inner_records=0, predicate_cache_records=0, scan_efficiency_ratio=13.71% (68/496)] +03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/dedup_probe.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ id@0 >= 11 AND id@0 <= 22 AND id@0 IN (SET) ([11, 22]) ], dynamic_rg_pruning=eligible, pruning_predicate=id_null_count@1 != row_count@2 AND id_max@0 >= 11 AND id_null_count@1 != row_count@2 AND id_min@3 <= 22 AND (id_null_count@1 != row_count@2 AND id_min@3 <= 11 AND 11 <= id_max@0 OR id_null_count@1 != row_count@2 AND id_min@3 <= 22 AND 22 <= id_max@0), required_guarantees=[id in (11, 22)], metrics=[output_rows=12, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=1 total → 1 matched, page_index_rows_pruned=12 total → 12 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, predicate_cache_inner_records=0, predicate_cache_records=0, scan_efficiency_ratio=18.34% (97/529)] + +statement ok +reset datafusion.explain.analyze_categories; + +statement ok +drop table dedup_build; + +statement ok +drop table dedup_probe;