@@ -7,7 +7,8 @@
 use std::{fmt::Display, hash::Hash, sync::Arc};
 
 use arrow::{
-    array::{ArrayRef, BooleanArray, BooleanBufferBuilder, UInt64Array},
+    array::{ArrayRef, BooleanArray, BooleanBufferBuilder, UInt32Array, UInt64Array},
+    compute::take,
     datatypes::{DataType, Schema},
     record_batch::RecordBatch,
 };
@@ -352,8 +353,7 @@ impl HashTableLookupExpr {
     ///   for pruning only, or `None`
     ///
     /// A single non-empty map is probed directly without routing, so it may be
-    /// any [`Map`] and `router` is ignored. Otherwise `router` is required and
-    /// every map must be a [`Map::HashMap`].
+    /// any [`Map`] and `router` is ignored. Otherwise `router` is required.
     ///
     /// # Public Only for Internal Use:
     /// `datafusion-proto` tests require this constructor, but it is not part of
@@ -375,21 +375,10 @@ impl HashTableLookupExpr {
                     "a lookup over anything but one non-empty map needs a partition router"
                 );
             }
-            (Some(router), _) => {
-                if maps
-                    .iter()
-                    .flatten()
-                    .any(|map| !matches!(map.as_ref(), Map::HashMap(_)))
-                {
-                    return internal_err!(
-                        "routed lookups only support hash table partitions"
-                    );
-                }
-                Lookup::Routed {
-                    router,
-                    maps: maps.into(),
-                }
-            }
+            (Some(router), _) => Lookup::Routed {
+                router,
+                maps: maps.into(),
+            },
         };
         Ok(Self {
             on_columns,
@@ -441,16 +430,23 @@ impl HashTableLookupExpr {
             for (partition, map) in maps.iter().enumerate() {
                 // An empty partition holds no key, so its rows stay unmatched.
                 let Some(map) = map else { continue };
-                let Map::HashMap(map) = map.as_ref() else {
-                    return internal_err!(
-                        "routed lookups only support hash table partitions"
-                    );
-                };
                 let run = starts[partition]..starts[partition + 1];
                 if run.is_empty() {
                     continue;
                 }
-                let hits = map.contain_hashes(&sorted_hashes[run.clone()]);
+                let hits = match map.as_ref() {
+                    Map::HashMap(map) => map.contain_hashes(&sorted_hashes[run.clone()]),
+                    // An array map is keyed by value, not hash, so it probes
+                    // the run's own key values.
+                    Map::ArrayMap(map) => {
+                        let indices = UInt32Array::from(rows[run.clone()].to_vec());
+                        let run_keys = keys
+                            .iter()
+                            .map(|key| take(key.as_ref(), &indices, None))
+                            .collect::<std::result::Result<Vec<_>, _>>()?;
+                        map.contain_keys(&run_keys)?
+                    }
+                };
                 for hit in hits.values().set_indices() {
                     matched.set_bit(rows[run.start + hit] as usize, true);
                 }
@@ -639,7 +635,9 @@ fn evaluate_columns(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::joins::array_map::ArrayMap;
     use crate::joins::join_hash_map::JoinHashMapU32;
+    use arrow::array::AsArray;
     use datafusion_common::ScalarValue;
     use datafusion_physical_expr::expressions::Column;
     use std::collections::hash_map::DefaultHasher;
@@ -1111,7 +1109,7 @@ mod tests {
         )]));
         let batch = RecordBatch::try_new(schema, vec![Arc::clone(probe)])?;
         let matched = expr.evaluate(&batch)?.into_array(batch.num_rows())?;
-        Ok(arrow::array::AsArray::as_boolean(&matched).clone())
+        Ok(matched.as_boolean().clone())
     }
 
     /// A hash table of `keys`, hashed with `random_state` as the build side
@@ -1125,6 +1123,18 @@ mod tests {
         Ok(Arc::new(Map::HashMap(Box::new(map))))
     }
 
+    /// An array map of `keys`, as a dense build side gets.
+    fn array_map(keys: &ArrayRef) -> Result<Arc<Map>> {
+        let keys = keys.as_primitive::<arrow::datatypes::Int32Type>();
+        let key = |value: Option<i32>| {
+            ArrayMap::key_to_u64(&ScalarValue::Int32(value)).expect("Int32 key")
+        };
+        let min = key(arrow::compute::min(keys));
+        let max = key(arrow::compute::max(keys));
+        let keys: ArrayRef = Arc::new(keys.clone());
+        Ok(Arc::new(Map::ArrayMap(ArrayMap::try_new(&keys, min, max)?)))
+    }
+
     /// One randomly generated scenario, see [`routed_lookup_fuzz`].
     fn run_routed_lookup_scenario(seed: u64) -> Result<()> {
         use crate::repartition::{BatchPartitioner, REPARTITION_RANDOM_STATE};
@@ -1166,7 +1176,12 @@ mod tests {
         .partition(
             RecordBatch::try_new(schema, vec![Arc::clone(&build)])?,
             |partition, batch| {
-                maps[partition] = Some(hash_table(batch.column(0), &random_state)?);
+                // A dense partition builds an array map, others a hash table.
+                maps[partition] = Some(if rng.random_bool(0.5) {
+                    array_map(batch.column(0))?
+                } else {
+                    hash_table(batch.column(0), &random_state)?
+                });
                 Ok(())
             },
         )?;
 
@@ -721,14 +721,7 @@ impl SharedBuildAccumulator {
                 }) => {
                     real_partition_ids.push(partition_id);
                     keys_have_null |= partition_keys_have_null;
-                    let mergeable = match &pushdown {
-                        PushdownStrategy::Map(map, _) => {
-                            matches!(map.as_ref(), Map::HashMap(_))
-                        }
-                        PushdownStrategy::InList(_) => true,
-                        PushdownStrategy::Empty => false,
-                    };
-                    members = members.filter(|_| mergeable).map(|mut members| {
+                    members = members.map(|mut members| {
                         members.push((partition_id, pushdown.clone(), bounds.clone()));
                         members
                     });