perf: deduplicate hash join InList pushdown values

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<Fields> {
 ///    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<ArrayRef> {
+    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<Row> = HashSet::with_capacity(values.len());
+    let mut indices: Vec<u32> = 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::<Int32Type>().values(),
+            &[1, 2, 3]
+        );
+        assert_eq!(
+            struct_array
+                .column(1)
+                .as_string::<i32>()
+                .iter()
+                .collect::<Vec<_>>(),
+            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());
     }
 }

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;