From ddb4233c686567b5916894205a0de366da2598d9 Mon Sep 17 00:00:00 2001 From: Robert Kruszewski Date: Wed, 21 Oct 2026 23:27:49 +0100 Subject: [PATCH] Remove repeated elements from IN lists DataFusion builds the IN list of a hash join filter from the keys of the build side, which repeat. Each repeat makes the conversion and the lookup more costly. - `in_list_of` removes equal elements after it sorts them. One null keeps the SQL semantics. - The DataFusion converter converts equal literals once. - The converter reads the type and nullability of a predicate, and does not ask for its return field. DataFusion names that field by formatting the whole expression, which for a large IN list costs more than the conversion. - `contiguous_integer_bounds` checks the dtype before it views a scalar as a primitive, because a failed view builds an error. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Robert Kruszewski --- vortex-array/src/expr/exprs.rs | 8 ++++-- vortex-datafusion/src/convert/exprs/mod.rs | 16 +++++++++++- vortex-datafusion/src/convert/exprs/tests.rs | 27 ++++++++++++++++++++ 3 files changed, 48 insertions(+), 3 deletions(-) diff --git a/vortex-array/src/expr/exprs.rs b/vortex-array/src/expr/exprs.rs index d77a3472d30..92c9296ca62 100644 --- a/vortex-array/src/expr/exprs.rs +++ b/vortex-array/src/expr/exprs.rs @@ -1227,11 +1227,14 @@ pub fn in_list_of( } // Ordered elements make equal lists from the same set, so that equal filters deduplicate. + // An engine can build a list from the keys of a hash join, which repeat, and each repeat makes + // the lookup more costly, so equal elements are removed. One null keeps the SQL semantics. if elements .windows(2) .all(|pair| pair[0].partial_cmp(&pair[1]).is_some()) { elements.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)); + elements.dedup(); } let list = Scalar::list(Arc::new(element_dtype), elements, Nullability::NonNullable); Ok(in_list(needle, lit(list))) @@ -1243,10 +1246,11 @@ fn contiguous_integer_bounds(scalars: &[Scalar]) -> Option<(&Scalar, &Scalar)> { let mut values = scalars .iter() .map(|scalar| { - let primitive = scalar.as_primitive_opt()?; - if !primitive.ptype().is_int() { + // Viewing a scalar of another type as a primitive builds an error, which is costly. + if !matches!(scalar.dtype(), DType::Primitive(ptype, _) if ptype.is_int()) { return None; } + let primitive = scalar.as_primitive_opt()?; let value = if primitive.ptype().is_unsigned_int() { primitive.as_opt::()??.into() } else { diff --git a/vortex-datafusion/src/convert/exprs/mod.rs b/vortex-datafusion/src/convert/exprs/mod.rs index b3b4563db09..0607fa05c9c 100644 --- a/vortex-datafusion/src/convert/exprs/mod.rs +++ b/vortex-datafusion/src/convert/exprs/mod.rs @@ -273,7 +273,14 @@ impl DefaultExpressionConverter { ) .map_err(|_| Unconverted::Unsupported)?; let converted = self.convert_expr(expr, schema, &input_dtype)?; - let expected_dtype = self.arrow_dtype(expr.return_field(schema)?.as_ref())?; + // `return_field` names the field by formatting the whole expression. For the IN list of a + // hash join filter, with thousands of keys, that costs more than the conversion. A + // predicate has no field metadata, so its type and nullability are enough. + let expected_dtype = if expr.data_type(schema)? == DataType::Boolean { + DType::Bool(expr.nullable(schema)?.into()) + } else { + self.arrow_dtype(expr.return_field(schema)?.as_ref())? + }; if !converted_dtype(&converted, &input_dtype)?.eq_ignore_nullability(&expected_dtype) { return Err(Unconverted::Unsupported); } @@ -442,9 +449,16 @@ impl DefaultExpressionConverter { } let value_dtype = converted_dtype(&value, input_dtype)?; + // The list of a hash join filter holds every build-side key, which repeat. Each element + // converts through Arrow, so equal elements are converted once. let elements = in_list .list() .iter() + .unique_by(|element| { + element + .downcast_ref::() + .map(|literal| literal.value().clone()) + }) .map(|element| { let element = self.convert_expr(element, schema, input_dtype)?; element diff --git a/vortex-datafusion/src/convert/exprs/tests.rs b/vortex-datafusion/src/convert/exprs/tests.rs index 6e5a31b7a30..613811e282c 100644 --- a/vortex-datafusion/src/convert/exprs/tests.rs +++ b/vortex-datafusion/src/convert/exprs/tests.rs @@ -365,6 +365,33 @@ fn test_in_list_converts_to_set_lookup(#[values(false, true)] negated: bool) -> Ok(()) } +#[test] +fn test_in_list_drops_repeated_elements() -> DFResult<()> { + let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]); + let expr = Arc::new(df_expr::InListExpr::try_new( + Arc::new(df_expr::Column::new("a", 0)), + int_literals([Some(7), None, Some(1), Some(7), None, Some(1)]), + false, + &schema, + )?); + let element_dtype = Arc::new(DType::Primitive(PType::I32, Nullability::Nullable)); + let list = Scalar::list( + element_dtype, + vec![ + Scalar::null(DType::Primitive(PType::I32, Nullability::Nullable)), + Scalar::from(Some(1i32)), + Scalar::from(Some(7i32)), + ], + Nullability::NonNullable, + ); + + assert_eq!( + convert(expr, &schema)?, + vortex::expr::in_list(get_item("a", root()), lit(list)) + ); + Ok(()) +} + #[test] fn test_contiguous_in_list_converts_to_between() -> DFResult<()> { let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);