@@ -216,7 +216,7 @@ impl InListExpr {
             expr,
             list,
             negated,
-            Some(instantiate_static_filter(array)?),
+            Some(instantiate_static_filter(array, &expr_data_type)?),
         ))
     }
 
@@ -243,7 +243,7 @@ impl InListExpr {
 
         // Try to create a static filter if all list expressions are constants
         let static_filter = match try_evaluate_constant_list(&list, schema)? {
-            Some(in_array) => Some(instantiate_static_filter(in_array)?),
+            Some(in_array) => Some(instantiate_static_filter(in_array, &expr_data_type)?),
             None => None, // Non-constant expressions, fall back to dynamic evaluation
         };
 
@@ -1265,6 +1265,31 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn in_list_nested_dictionary_scalar() -> Result<()> {
+        let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
+        let batch = RecordBatch::try_new(
+            Arc::new(schema.clone()),
+            vec![Arc::new(Int32Array::from(vec![0, 0, 0]))],
+        )?;
+        let needle = lit(ScalarValue::Dictionary(
+            Box::new(DataType::Int16),
+            Box::new(ScalarValue::Dictionary(
+                Box::new(DataType::Int8),
+                Box::new(ScalarValue::Int32(Some(2))),
+            )),
+        ));
+
+        let expr = in_list(needle, vec![lit(1_i32), lit(2_i32)], &false, &schema)?;
+        let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?;
+        assert_eq!(
+            as_boolean_array(&result),
+            &BooleanArray::from(vec![true, true, true])
+        );
+
+        Ok(())
+    }
+
     #[test]
     fn in_list_utf8_with_dict_types() -> Result<()> {
         fn dict_lit(key_type: DataType, value: &str) -> Arc<dyn PhysicalExpr> {
 
@@ -15,12 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use arrow::array::{
-    Array, ArrayRef, BooleanArray, downcast_array, downcast_dictionary_array,
-    make_comparator,
-};
+use arrow::array::{Array, ArrayRef, BooleanArray, make_comparator};
 use arrow::buffer::{BooleanBuffer, NullBuffer};
-use arrow::compute::{SortOptions, take};
+use arrow::compute::SortOptions;
 use arrow::datatypes::DataType;
 use arrow::util::bit_iterator::BitIndexIterator;
 use datafusion_common::Result;
@@ -141,22 +138,6 @@ impl StaticFilter for ArrayStaticFilter {
             ));
         }
 
-        // Unwrap dictionary-encoded needles when the value type matches
-        // in_array, evaluating against the dictionary values and mapping
-        // back via keys.
-        downcast_dictionary_array! {
-            v => {
-                // Only unwrap when the haystack (in_array) type matches
-                // the dictionary value type
-                if v.values().data_type() == self.in_array.data_type() {
-                    let values_contains = self.contains(v.values().as_ref(), negated)?;
-                    let result = take(&values_contains, v.keys(), None)?;
-                    return Ok(downcast_array(result.as_ref()));
-                }
-            }
-            _ => {}
-        }
-
         self.find_needles_in_haystack(v, negated)
     }
 }
@@ -77,7 +77,7 @@ use arrow::util::bit_iterator::BitIndexIterator;
 use datafusion_common::{Result, exec_datafusion_err, internal_datafusion_err};
 
 use super::result::build_result_from_contains;
-use super::static_filter::{StaticFilter, handle_dictionary};
+use super::static_filter::StaticFilter;
 
 pub(super) type BranchlessNative<T> =
     <<T as BranchlessFilterType>::CompareType as ArrowPrimitiveType>::Native;
@@ -256,8 +256,6 @@ where
     }
 
     fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
-        handle_dictionary!(self, v, negated);
-
         // Arrow compatibility ignores timestamp timezone and decimal precision/scale
         // while still requiring the same primitive representation.
         if !PrimitiveArray::<T>::is_compatible(v.data_type()) {
 
@@ -53,7 +53,7 @@ use arrow::datatypes::{
 use datafusion_common::{Result, exec_datafusion_err, internal_datafusion_err};
 
 use super::primitive_filter::instantiate_primitive_filter;
-use super::static_filter::{StaticFilter, StaticFilterRef, handle_dictionary};
+use super::static_filter::{StaticFilter, StaticFilterRef};
 
 /// Reinterpret fixed-size binary values as same-width primitive values.
 ///
@@ -99,8 +99,6 @@ where
     }
 
     fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
-        handle_dictionary!(self, v, negated);
-
         if v.data_type() != &self.data_type {
             return Err(exec_datafusion_err!(
                 "FixedSizeBinary filter: expected {} array, got {}",
@@ -165,6 +163,7 @@ mod tests {
     use arrow::buffer::{Buffer, MutableBuffer, NullBuffer};
     use arrow::datatypes::Int8Type;
 
+    use super::super::static_filter::DictionaryFilter;
     use super::*;
 
     fn value(width: i32, index: usize, miss: bool) -> Vec<u8> {
@@ -268,7 +267,7 @@ mod tests {
 
     #[test]
     fn handles_dictionary_needles() -> Result<()> {
-        let filter = make_filter(4, &[Some(value(4, 7, false))])?;
+        let filter = DictionaryFilter::new(make_filter(4, &[Some(value(4, 7, false))])?);
         let dictionary_values: ArrayRef = Arc::new(array(
             4,
             &[Some(value(4, 7, false)), Some(value(4, 8, false))],
 
@@ -19,17 +19,18 @@
 //!
 //! This module provides membership tests for Arrow primitive types.
 
+use std::hash::{Hash, Hasher};
+use std::marker::PhantomData;
+use std::sync::Arc;
+
 use arrow::array::{Array, ArrayRef, AsArray, BooleanArray};
 use arrow::datatypes::*;
 use arrow::util::bit_iterator::BitIndexIterator;
 use datafusion_common::{HashSet, Result, exec_datafusion_err};
-use std::hash::{Hash, Hasher};
-use std::marker::PhantomData;
-use std::sync::Arc;
 
 use super::branchless_filter::{BranchlessFilter, BranchlessFilterType};
 use super::result::build_in_list_result;
-use super::static_filter::{StaticFilter, StaticFilterRef, handle_dictionary};
+use super::static_filter::{StaticFilter, StaticFilterRef};
 
 /// Selects an optimized filter for a primitive representation.
 ///
@@ -311,7 +312,6 @@ where
     }
 
     fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
-        handle_dictionary!(self, v, negated);
         let v = v.as_primitive_opt::<T>().ok_or_else(|| {
             exec_datafusion_err!("BitmapFilter: expected {} array", T::DATA_TYPE)
         })?;
@@ -432,8 +432,6 @@ where
     }
 
     fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
-        handle_dictionary!(self, v, negated);
-
         let v = v.as_primitive_opt::<T>().ok_or_else(|| {
             exec_datafusion_err!(
                 "PrimitiveHashSetFilter: expected {} array",
@@ -465,6 +463,8 @@ mod tests {
     };
     use half::f16;
 
+    use super::super::static_filter::DictionaryFilter;
+
     fn uint32_array(values: Vec<Option<u32>>) -> ArrayRef {
         Arc::new(UInt32Array::from(values))
     }
@@ -553,13 +553,19 @@ mod tests {
     #[test]
     fn bitmap_filter_u8_handles_dictionary_needles() -> Result<()> {
         let haystack: ArrayRef = Arc::new(UInt8Array::from(vec![Some(1), None, Some(3)]));
-        let filter = BitmapFilter::<UInt8Type>::try_new(&haystack)?;
+        let inner: StaticFilterRef =
+            Arc::new(BitmapFilter::<UInt8Type>::try_new(&haystack)?);
+        let filter = DictionaryFilter::new(inner);
 
         let keys = Int8Array::from(vec![Some(0), Some(1), None, Some(2)]);
         let values = Arc::new(UInt8Array::from(vec![Some(1), Some(2), Some(3)]));
         let needles = DictionaryArray::try_new(keys, values)?;
 
-        assert_contains(&filter, &needles, vec![Some(true), None, None, Some(true)])
+        assert_eq!(
+            filter.contains(&needles, false)?,
+            BooleanArray::from(vec![Some(true), None, None, Some(true)])
+        );
+        Ok(())
     }
 
     #[test]
 
@@ -17,42 +17,162 @@
 
 use std::sync::Arc;
 
-use arrow::array::{Array, BooleanArray};
+use arrow::array::{Array, AsArray, BooleanArray};
+use arrow::compute::take;
+use arrow::datatypes::DataType;
 use datafusion_common::Result;
 
-pub(super) type StaticFilterRef = Arc<dyn StaticFilter + Send + Sync>;
+pub(super) type StaticFilterRef = Arc<dyn StaticFilter>;
+
+/// Adds dictionary-encoded needle support to a filter that expects plain arrays.
+/// This wrapper is only used when the input expression returns dictionaries.
+pub(super) struct DictionaryFilter {
+    inner: StaticFilterRef,
+}
+
+impl DictionaryFilter {
+    pub(super) fn new(inner: StaticFilterRef) -> Self {
+        Self { inner }
+    }
+}
+
+impl StaticFilter for DictionaryFilter {
+    fn null_count(&self) -> usize {
+        self.inner.null_count()
+    }
+
+    fn contains(&self, needles: &dyn Array, negated: bool) -> Result<BooleanArray> {
+        let Some(dictionary) = needles.as_any_dictionary_opt() else {
+            return self.inner.contains(needles, negated);
+        };
+        let values = dictionary.values();
+        let values_contains = if matches!(values.data_type(), DataType::Dictionary(_, _))
+        {
+            self.contains(values.as_ref(), negated)?
+        } else {
+            self.inner.contains(values.as_ref(), negated)?
+        };
+        let result = take(&values_contains, dictionary.keys(), None)?;
+        Ok(result.as_boolean().clone())
+    }
+}
 
 /// Trait for InList static filters.
 ///
 /// Static filters store a pre-computed set of values (the haystack) and check
 /// whether needle values are contained in that set. The haystack is always
 /// represented in its non-dictionary (value) type. Dictionary haystacks are
-/// flattened via `cast()` before construction.
+/// flattened before construction.
 ///
-/// Dictionary-encoded needles are unwrapped inside `contains()` and
-/// evaluated against the dictionary's values.
-pub(super) trait StaticFilter {
+/// Dictionary-encoded needles are handled by [`DictionaryFilter`] before the
+/// concrete filter is called.
+pub(super) trait StaticFilter: Send + Sync {
     fn null_count(&self) -> usize;
 
-    /// Checks if values in `v` (needle) are contained in this filter's
-    /// haystack. `v` may be dictionary-encoded, in which case the
-    /// implementation unwraps the dictionary and operates on its values.
-    fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray>;
+    /// Checks if non-dictionary needles are contained in this filter's haystack.
+    fn contains(&self, needles: &dyn Array, negated: bool) -> Result<BooleanArray>;
 }
 
-/// Evaluate dictionary-encoded needles by applying a filter to dictionary
-/// values and remapping the result through the keys.
-macro_rules! handle_dictionary {
-    ($self:ident, $v:ident, $negated:ident) => {
-        arrow::array::downcast_dictionary_array! {
-            $v => {
-                let values_contains = $self.contains($v.values().as_ref(), $negated)?;
-                let result = arrow::compute::take(&values_contains, $v.keys(), None)?;
-                return Ok(arrow::array::downcast_array(result.as_ref()))
-            }
-            _ => {}
-        }
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use arrow::array::{
+        ArrayRef, Decimal128Array, DictionaryArray, Int8Array, Int16Array, Int32Array,
+        Int64Array, TimestampNanosecondArray, UInt8Array, UInt16Array, UInt32Array,
+        UInt64Array,
     };
-}
 
-pub(super) use handle_dictionary;
+    use super::super::strategy::instantiate_static_filter;
+
+    #[test]
+    fn dictionary_needles_support_all_key_types() -> Result<()> {
+        let values: ArrayRef = Arc::new(Int32Array::from(vec![1, 2]));
+        let expected = BooleanArray::from(vec![Some(true), Some(false), None]);
+
+        macro_rules! check_keys {
+            ($keys:expr) => {{
+                let needles = DictionaryArray::try_new($keys, Arc::clone(&values))?;
+                let filter = instantiate_static_filter(
+                    Arc::new(Int32Array::from(vec![1, 3])),
+                    needles.data_type(),
+                )?;
+                assert_eq!(filter.contains(&needles, false)?, expected);
+            }};
+        }
+
+        check_keys!(Int8Array::from(vec![Some(0), Some(1), None]));
+        check_keys!(Int16Array::from(vec![Some(0), Some(1), None]));
+        check_keys!(Int32Array::from(vec![Some(0), Some(1), None]));
+        check_keys!(Int64Array::from(vec![Some(0), Some(1), None]));
+        check_keys!(UInt8Array::from(vec![Some(0), Some(1), None]));
+        check_keys!(UInt16Array::from(vec![Some(0), Some(1), None]));
+        check_keys!(UInt32Array::from(vec![Some(0), Some(1), None]));
+        check_keys!(UInt64Array::from(vec![Some(0), Some(1), None]));
+
+        Ok(())
+    }
+
+    #[test]
+    fn nested_dictionary_needles_preserve_nulls() -> Result<()> {
+        let values: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), Some(2), None]));
+        let inner: ArrayRef = Arc::new(DictionaryArray::try_new(
+            Int8Array::from(vec![0, 1, 2]),
+            values,
+        )?);
+        let needles = DictionaryArray::try_new(
+            Int16Array::from(vec![Some(0), Some(1), Some(2), None]),
+            inner,
+        )?;
+        let filter = instantiate_static_filter(
+            Arc::new(Int32Array::from(vec![1, 3])),
+            needles.data_type(),
+        )?;
+
+        assert_eq!(
+            filter.contains(&needles, false)?,
+            BooleanArray::from(vec![Some(true), Some(false), None, None])
+        );
+        assert_eq!(
+            filter.contains(&needles, true)?,
+            BooleanArray::from(vec![Some(false), Some(true), None, None])
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn dictionary_timestamp_needles_keep_timezone_compatibility() -> Result<()> {
+        let timestamps: ArrayRef =
+            Arc::new(TimestampNanosecondArray::from(vec![1, 3]).with_timezone("UTC"));
+        let values: ArrayRef = Arc::new(
+            TimestampNanosecondArray::from(vec![1, 2]).with_timezone("Europe/Paris"),
+        );
+        let needles = DictionaryArray::try_new(Int8Array::from(vec![0, 1]), values)?;
+        let filter = instantiate_static_filter(timestamps, needles.data_type())?;
+        assert_eq!(
+            filter.contains(&needles, false)?,
+            BooleanArray::from(vec![true, false])
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn dictionary_decimal_needles_keep_precision_scale_compatibility() -> Result<()> {
+        // Five list values use the hash-set filter.
+        let decimals: ArrayRef = Arc::new(
+            Decimal128Array::from(vec![1, 3, 5, 7, 9]).with_precision_and_scale(10, 2)?,
+        );
+        let values: ArrayRef =
+            Arc::new(Decimal128Array::from(vec![1, 2]).with_precision_and_scale(11, 3)?);
+        let needles = DictionaryArray::try_new(Int8Array::from(vec![0, 1]), values)?;
+        let filter = instantiate_static_filter(decimals, needles.data_type())?;
+        assert_eq!(
+            filter.contains(&needles, false)?,
+            BooleanArray::from(vec![true, false])
+        );
+
+        Ok(())
+    }
+}