@@ -6,6 +6,8 @@
 use std::sync::Arc;
 
 use vortex_error::VortexExpect;
+use vortex_error::VortexResult;
+use vortex_error::vortex_ensure;
 use vortex_error::vortex_panic;
 use vortex_utils::iter::ReduceBalancedIterExt;
 
@@ -1148,31 +1150,63 @@ pub fn in_list(needle: Expression, list: Expression) -> Expression {
     )
 }
 
-/// Creates `needle IN (elements)` for a list of constant elements of `element_dtype`.
+/// Creates `needle IN (elements)` with SQL null semantics for a list of constant elements of
+/// `element_dtype`, ignoring nullability. An element can also be an untyped null.
+///
+/// The nullability of `element_dtype` is not used: the list elements are nullable only when a
+/// null element is present.
 ///
 /// Non-null integers that hold every value between their smallest and largest become a range
 /// check, which is cheaper to evaluate and prunes with tighter bounds than a set lookup, as for
 /// the keys of a small dimension table that a join pushes down.
 ///
+/// # Errors
+///
+/// Returns an error if an element is not a null and not of `element_dtype`.
+///
 /// ```rust
 /// # use vortex_array::dtype::{DType, Nullability, PType};
 /// # use vortex_array::expr::{in_list_of, root};
 /// let dtype = DType::Primitive(PType::I32, Nullability::NonNullable);
-/// let expr = in_list_of(root(), dtype, vec![3i32.into(), 1i32.into(), 2i32.into()]);
+/// let expr = in_list_of(root(), dtype, vec![3i32.into(), 1i32.into(), 2i32.into()])?;
 /// assert_eq!(expr.to_string(), "(1i32 <= $ <= 3i32)");
+/// # Ok::<(), vortex_error::VortexError>(())
 /// ```
 pub fn in_list_of(
     needle: Expression,
     element_dtype: DType,
-    mut elements: Vec<Scalar>,
-) -> Expression {
+    elements: Vec<Scalar>,
+) -> VortexResult<Expression> {
+    let element_dtype = if elements.iter().any(Scalar::is_null) {
+        element_dtype.as_nullable()
+    } else {
+        element_dtype.as_nonnullable()
+    };
+    let mut elements = elements
+        .iter()
+        .map(|element| {
+            vortex_ensure!(
+                element.dtype() == &DType::Null
+                    || element.dtype().eq_ignore_nullability(&element_dtype),
+                "IN list element {element} is not of type {element_dtype}"
+            );
+            element.cast(&element_dtype)
+        })
+        .collect::<VortexResult<Vec<_>>>()?;
+
     if let Some((low, high)) = contiguous_integer_bounds(&elements) {
         let options = BetweenOptions {
             lower_strict: StrictComparison::NonStrict,
             upper_strict: StrictComparison::NonStrict,
         };
-        return between(needle, lit(low.clone()), lit(high.clone()), options);
+        return Ok(between(
+            needle,
+            lit(low.clone()),
+            lit(high.clone()),
+            options,
+        ));
     }
+
     // Ordered elements make equal lists from the same set, so that equal filters deduplicate.
     if elements
         .windows(2)
@@ -1181,7 +1215,7 @@ pub fn in_list_of(
         elements.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
     }
     let list = Scalar::list(Arc::new(element_dtype), elements, Nullability::NonNullable);
-    in_list(needle, lit(list))
+    Ok(in_list(needle, lit(list)))
 }
 
 /// The smallest and largest of non-null integer `scalars` when they hold every integer between
 
@@ -26,9 +26,7 @@ use vortex::VortexSessionDefault;
 use vortex::array::VortexSessionExecute;
 use vortex::dtype::DType;
 use vortex::dtype::Nullability;
-use vortex::error::VortexResult;
 use vortex::expr::Expression;
-use vortex::expr::and_collect;
 use vortex::expr::byte_length;
 use vortex::expr::cast;
 use vortex::expr::get_item;
@@ -39,10 +37,8 @@ use vortex::expr::list_length;
 use vortex::expr::lit;
 use vortex::expr::nested_case_when;
 use vortex::expr::not;
-use vortex::expr::or_collect;
 use vortex::expr::pack;
 use vortex::expr::root;
-use vortex::scalar::Scalar;
 use vortex::scalar_fn::ScalarFnVTableExt;
 use vortex::scalar_fn::fns::binary::Binary;
 use vortex::scalar_fn::fns::like::Like;
@@ -440,64 +436,27 @@ impl DefaultExpressionConverter {
             return Err(Unconverted::Unsupported);
         }
         let value_dtype = converted_dtype(&value, input_dtype)?;
-        let operator = if in_list.negated() {
-            Operator::NotEq
-        } else {
-            Operator::Eq
-        };
-        // A positive list of literals becomes one set lookup. An OR of equalities would
-        // scan the needle once per element, which dynamic join filters make up to 150 times.
-        // Floats keep the comparisons, whose -0.0 and NaN semantics differ from set membership.
-        let is_float = matches!(&value_dtype, DType::Primitive(ptype, _) if ptype.is_float());
-        if !in_list.negated() && !is_float {
-            let elements = in_list
-                .list()
-                .iter()
-                .map(|element| self.convert_expr(element, schema, input_dtype))
-                .collect::<Conversion<Vec<_>>>()?;
-            let scalars = elements
-                .iter()
-                .map(|element| element.as_opt::<Literal>().cloned())
-                .collect::<Option<Vec<_>>>();
-            // Null elements make a nullable list, whose SQL semantics `in_list` preserves.
-            let element_dtype = if scalars.iter().flatten().any(Scalar::is_null) {
-                value_dtype.as_nullable()
-            } else {
-                value_dtype.as_nonnullable()
-            };
-            if let Some(scalars) = scalars
-                && !scalars.is_empty()
-                && scalars
-                    .iter()
-                    .all(|scalar| scalar.dtype().eq_ignore_nullability(&element_dtype))
-            {
-                let scalars = scalars
-                    .iter()
-                    .map(|scalar| scalar.cast(&element_dtype))
-                    .collect::<VortexResult<Vec<_>>>()
-                    .map_err(|_| Unconverted::Unsupported)?;
-                return Ok(in_list_of(value, element_dtype, scalars));
-            }
-        }
-        let mut comparisons = Vec::with_capacity(in_list.len());
-        for element in in_list.list() {
-            let element = self.convert_expr(element, schema, input_dtype)?;
-            let element_dtype = converted_dtype(&element, input_dtype)?;
-            if element_dtype == DType::Null {
-                comparisons.push(lit(None::<bool>));
-            } else if value_dtype.eq_ignore_nullability(&element_dtype) {
-                comparisons.push(Binary.new_expr(operator, [value.clone(), element]));
-            } else {
-                return Err(Unconverted::Unsupported);
-            }
-        }
-        // Kleene AND/OR preserve SQL IN/NOT IN nulls; list_contains does not.
-        let membership = if in_list.negated() {
-            and_collect(comparisons)
+
+        let elements = in_list
+            .list()
+            .iter()
+            .map(|element| {
+                let element = self.convert_expr(element, schema, input_dtype)?;
+                element
+                    .as_opt::<Literal>()
+                    .cloned()
+                    .ok_or(Unconverted::Unsupported)
+            })
+            .collect::<Conversion<Vec<_>>>()?;
+
+        // An element of another type than the value declines the pushdown: no implicit casts.
+        let membership =
+            in_list_of(value, value_dtype, elements).map_err(|_| Unconverted::Unsupported)?;
+        Ok(if in_list.negated() {
+            not(membership)
         } else {
-            or_collect(comparisons)
-        };
-        membership.ok_or(Unconverted::Unsupported)
+            membership
+        })
     }
 
     fn convert_scalar_function(
 
@@ -329,13 +329,13 @@ fn test_native_in_list_large(#[values(false, true)] negated: bool) -> anyhow::Re
     )
 }
 
-#[test]
-fn test_in_list_converts_to_set_lookup() -> DFResult<()> {
+#[rstest]
+fn test_in_list_converts_to_set_lookup(#[values(false, true)] negated: bool) -> 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(1), Some(3), Some(7)]),
-        false,
+        negated,
         &schema,
     )?);
     let element_dtype = Arc::new(DType::Primitive(PType::I32, Nullability::NonNullable));
@@ -344,10 +344,10 @@ fn test_in_list_converts_to_set_lookup() -> DFResult<()> {
         vec![1i32.into(), 3i32.into(), 7i32.into()],
         Nullability::NonNullable,
     );
-    assert_eq!(
-        convert(expr, &schema)?,
-        vortex::expr::in_list(get_item("a", root()), lit(list))
-    );
+    let membership = vortex::expr::in_list(get_item("a", root()), lit(list));
+    let expected = if negated { not(membership) } else { membership };
+
+    assert_eq!(convert(expr, &schema)?, expected);
     Ok(())
 }
 
 
@@ -751,9 +751,8 @@ fn try_from_compare_in(
     else {
         return Ok(None);
     };
-    // The IN list itself is never null. Null values are elements, and in_list handles them.
     let element_dtype = list_elements[0].dtype().clone();
-    let expr = in_list_of(element, element_dtype, list_elements);
+    let expr = in_list_of(element, element_dtype, list_elements)?;
     Ok(Some(if not_in { not(expr) } else { expr }))
 }
 
 
@@ -89,9 +89,7 @@ pub fn try_from_table_filter(
                 "IN filter must have at least one value"
             );
             let dtype = scalars[0].dtype().clone();
-            // The IN list itself is never null. Null values are elements, and in_list handles
-            // them.
-            in_list_of(col.clone(), dtype, scalars)
+            in_list_of(col.clone(), dtype, scalars)?
         }
         TableFilterClass::Dynamic(dynamic) => {
             dynamic_filter_expr(dynamic, col.return_dtype(scope_dtype)?, col.clone())?