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 <noreply@anthropic.com>
Signed-off-by: Robert Kruszewski <github@robertk.io>

Author: robert3005

vortex-array/src/expr/exprs.rs

@@ -1208,11 +1208,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)))
@@ -1224,10 +1227,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::<u64>()??.into()
             } else {

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::<df_expr::Literal>()
+                    .map(|literal| literal.value().clone())
+            })
             .map(|element| {
                 let element = self.convert_expr(element, schema, input_dtype)?;
                 element

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)]);