@@ -146,18 +146,22 @@ pub struct JoinHashMapU32 {
     map: HashTable<(u64, u32)>,
     // Stores indices in chained list data structure
     next: Vec<u32>,
+    // Whether any hash value maps to more than one row
+    chained: bool,
 }
 
 impl JoinHashMapU32 {
     #[cfg(test)]
     pub(crate) fn new(map: HashTable<(u64, u32)>, next: Vec<u32>) -> Self {
-        Self { map, next }
+        let chained = map.len() != next.len();
+        Self { map, next, chained }
     }
 
     pub fn with_capacity(cap: usize) -> Self {
         Self {
             map: HashTable::with_capacity(cap),
             next: vec![0; cap],
+            chained: false,
         }
     }
 }
@@ -176,7 +180,8 @@ impl JoinHashMapType for JoinHashMapU32 {
         iter: Box<dyn Iterator<Item = (usize, &'a u64)> + Send + 'a>,
         deleted_offset: usize,
     ) {
-        update_from_iter::<u32>(&mut self.map, &mut self.next, iter, deleted_offset);
+        self.chained |=
+            update_from_iter::<u32>(&mut self.map, &mut self.next, iter, deleted_offset);
     }
 
     fn get_matched_indices<'a>(
@@ -199,6 +204,7 @@ impl JoinHashMapType for JoinHashMapU32 {
         get_matched_indices_with_limit_offset::<u32>(
             &self.map,
             &self.next,
+            !self.chained,
             hash_values,
             valid_keys,
             limit,
@@ -226,18 +232,22 @@ pub struct JoinHashMapU64 {
     map: HashTable<(u64, u64)>,
     // Stores indices in chained list data structure
     next: Vec<u64>,
+    // Whether any hash value maps to more than one row
+    chained: bool,
 }
 
 impl JoinHashMapU64 {
     #[cfg(test)]
     pub(crate) fn new(map: HashTable<(u64, u64)>, next: Vec<u64>) -> Self {
-        Self { map, next }
+        let chained = map.len() != next.len();
+        Self { map, next, chained }
     }
 
     pub fn with_capacity(cap: usize) -> Self {
         Self {
             map: HashTable::with_capacity(cap),
             next: vec![0; cap],
+            chained: false,
         }
     }
 }
@@ -256,7 +266,8 @@ impl JoinHashMapType for JoinHashMapU64 {
         iter: Box<dyn Iterator<Item = (usize, &'a u64)> + Send + 'a>,
         deleted_offset: usize,
     ) {
-        update_from_iter::<u64>(&mut self.map, &mut self.next, iter, deleted_offset);
+        self.chained |=
+            update_from_iter::<u64>(&mut self.map, &mut self.next, iter, deleted_offset);
     }
 
     fn get_matched_indices<'a>(
@@ -279,6 +290,7 @@ impl JoinHashMapType for JoinHashMapU64 {
         get_matched_indices_with_limit_offset::<u64>(
             &self.map,
             &self.next,
+            !self.chained,
             hash_values,
             valid_keys,
             limit,
@@ -304,15 +316,19 @@ impl JoinHashMapType for JoinHashMapU64 {
 use crate::joins::MapOffset;
 use crate::joins::chain::traverse_chain;
 
+/// Inserts the rows of `iter` into `map` and `next`. Returns `true` if any row
+/// was chained onto an existing hash value.
 pub fn update_from_iter<'a, T>(
     map: &mut HashTable<(u64, T)>,
     next: &mut [T],
     iter: Box<dyn Iterator<Item = (usize, &'a u64)> + Send + 'a>,
     deleted_offset: usize,
-) where
+) -> bool
+where
     T: Copy + TryFrom<usize> + PartialOrd,
     <T as TryFrom<usize>>::Error: Debug,
 {
+    let mut chained = false;
     for (row, &hash_value) in iter {
         let entry = map.entry(
             hash_value,
@@ -329,12 +345,14 @@ pub fn update_from_iter<'a, T>(
                 *index = T::try_from(row + 1).unwrap();
                 // Update chained Vec at `row` with previous value
                 next[row - deleted_offset] = prev_index;
+                chained = true;
             }
             Vacant(vacant_entry) => {
                 vacant_entry.insert((hash_value, T::try_from(row + 1).unwrap()));
             }
         }
     }
+    chained
 }
 
 pub fn get_matched_indices<'a, T>(
@@ -385,10 +403,13 @@ where
     (input_indices, match_indices)
 }
 
+/// `unique` must be `true` only if every hash value in `map` maps to a single
+/// row; the chain in `next_chain` is then not traversed.
 #[expect(clippy::too_many_arguments)]
 pub fn get_matched_indices_with_limit_offset<T>(
     map: &HashTable<(u64, T)>,
     next_chain: &[T],
+    unique: bool,
     hash_values: &[u64],
     valid_keys: Option<&NullBuffer>,
     limit: usize,
@@ -408,7 +429,7 @@ where
 
     // Check if hashmap consists of unique values
     // If so, we can skip the chain traversal
-    if map.len() == next_chain.len() {
+    if unique {
         let start = offset.0;
         let end = (start + limit).min(hash_values.len());
         for (i, &hash) in hash_values[start..end].iter().enumerate() {
@@ -569,4 +590,109 @@ mod tests {
         assert_eq!(input_indices, vec![1, 1]);
         assert_eq!(match_indices, vec![3, 1]);
     }
+
+    #[test]
+    fn test_unique_fast_path_with_uninserted_row() {
+        // Row 1 is left out, as `update_hash` does for a NULL key under
+        // `NullEquality::NullEqualsNothing`. Every hash value still maps to a
+        // single row, so the probe takes the unique-key fast path.
+        let build_hashes = [10u64, 99, 20, 30];
+        let mut hash_map = JoinHashMapU32::with_capacity(build_hashes.len());
+        hash_map.update_from_iter(
+            Box::new(build_hashes.iter().enumerate().filter(|(row, _)| *row != 1)),
+            0,
+        );
+
+        let probe_hashes = vec![10, 20, 30];
+        let mut input_indices = vec![];
+        let mut match_indices = vec![];
+        let next_offset = hash_map.get_matched_indices_with_limit_offset(
+            &probe_hashes,
+            None,
+            2,
+            (0, None),
+            &mut input_indices,
+            &mut match_indices,
+        );
+        // The fast path limits by probe rows and resumes at probe row 2; the
+        // chain walk would stop after two matches at `(1, Some(0))`.
+        assert_eq!(next_offset, Some((2, None)));
+        assert_eq!(input_indices, vec![0, 1]);
+        assert_eq!(match_indices, vec![0, 2]);
+
+        let next_offset = hash_map.get_matched_indices_with_limit_offset(
+            &probe_hashes,
+            None,
+            2,
+            (2, None),
+            &mut input_indices,
+            &mut match_indices,
+        );
+        assert_eq!(next_offset, None);
+        assert_eq!(input_indices, vec![2]);
+        assert_eq!(match_indices, vec![3]);
+    }
+
+    #[test]
+    fn test_chain_from_earlier_update_disables_fast_path() {
+        // Rows 0 and 1 share a hash value; a later update without a chain must
+        // not re-enable the unique-key fast path.
+        let mut hash_map = JoinHashMapU32::with_capacity(3);
+        hash_map.update_from_iter(Box::new([10u64, 10u64].iter().enumerate()), 0);
+        hash_map.update_from_iter(Box::new([(2, &20u64)].into_iter()), 0);
+
+        let mut input_indices = vec![];
+        let mut match_indices = vec![];
+        let next_offset = hash_map.get_matched_indices_with_limit_offset(
+            &[10],
+            None,
+            8192,
+            (0, None),
+            &mut input_indices,
+            &mut match_indices,
+        );
+
+        assert_eq!(next_offset, None);
+        assert_eq!(input_indices, vec![0, 0]);
+        assert_eq!(match_indices, vec![1, 0]);
+    }
+
+    #[test]
+    fn test_chain_created_by_later_update_disables_fast_path() {
+        assert_chain_created_by_later_update(JoinHashMapU32::with_capacity(3));
+        assert_chain_created_by_later_update(JoinHashMapU64::with_capacity(3));
+    }
+
+    /// Rows 0 and 2 share a hash value but arrive in separate updates, and row
+    /// 1 is left out as a NULL key would be. The first update alone is unique;
+    /// the second creates the chain. Probing with `limit = 1` must resume
+    /// through the chain and return each build row exactly once.
+    fn assert_chain_created_by_later_update(mut hash_map: impl JoinHashMapType) {
+        hash_map.update_from_iter(Box::new([(0, &10u64)].into_iter()), 0);
+        hash_map.update_from_iter(Box::new([(2, &10u64)].into_iter()), 0);
+
+        let mut input_indices = vec![];
+        let mut match_indices = vec![];
+        let mut matched = vec![];
+        let mut offset = (0, None);
+        for _ in 0..3 {
+            let next_offset = hash_map.get_matched_indices_with_limit_offset(
+                &[10],
+                None,
+                1,
+                offset,
+                &mut input_indices,
+                &mut match_indices,
+            );
+            assert_eq!(input_indices, vec![0]);
+            matched.extend_from_slice(&match_indices);
+            match next_offset {
+                Some(next) => offset = next,
+                None => break,
+            }
+        }
+
+        matched.sort_unstable();
+        assert_eq!(matched, vec![0, 2]);
+    }
 }
@@ -92,6 +92,7 @@ impl JoinHashMapType for PruningJoinHashMap {
         get_matched_indices_with_limit_offset::<u64>(
             &self.map,
             &next,
+            self.map.len() == next.len(),
             hash_values,
             valid_keys,
             limit,
 
@@ -2864,6 +2864,72 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn update_hash_null_key_does_not_change_matches() -> Result<()> {
+        use crate::joins::join_hash_map::JoinHashMapU32;
+
+        let random_state = RandomState::with_seed(42);
+        let probe: ArrayRef = Arc::new(StringArray::from(vec!["c", "a", "x", "b"]));
+        let mut probe_hashes = vec![0; probe.len()];
+        create_hashes([&probe], &random_state, &mut probe_hashes)?;
+
+        // Probes a unique-key build side 2 rows at a time and returns the
+        // matched (probe row, build key) pairs.
+        let matches = |build: Vec<Option<&str>>| -> Result<Vec<(u32, String)>> {
+            let build = StringArray::from(build);
+            let schema =
+                Arc::new(Schema::new(vec![Field::new("k", DataType::Utf8, true)]));
+            let batch = RecordBatch::try_new(schema, vec![Arc::new(build.clone())])?;
+            let on: Vec<PhysicalExprRef> = vec![Arc::new(Column::new("k", 0))];
+            let mut map = JoinHashMapU32::with_capacity(batch.num_rows());
+            let mut hashes_buffer = vec![0; batch.num_rows()];
+            update_hash(
+                &on,
+                &batch,
+                &mut map,
+                0,
+                &random_state,
+                &mut hashes_buffer,
+                0,
+                true,
+                NullEquality::NullEqualsNothing,
+            )?;
+            let (mut input_indices, mut match_indices, mut pairs) =
+                (vec![], vec![], vec![]);
+            let mut offset = Some((0, None));
+            while let Some(current) = offset {
+                offset = map.get_matched_indices_with_limit_offset(
+                    &probe_hashes,
+                    None,
+                    2,
+                    current,
+                    &mut input_indices,
+                    &mut match_indices,
+                );
+                pairs.extend(
+                    input_indices
+                        .iter()
+                        .zip(&match_indices)
+                        .map(|(&p, &b)| (p, build.value(b as usize).to_string())),
+                );
+            }
+            pairs.sort();
+            Ok(pairs)
+        };
+
+        let without_null = matches(vec![Some("a"), Some("b"), Some("c")])?;
+        let with_null = matches(vec![Some("a"), None, Some("b"), Some("c")])?;
+        assert_eq!(with_null, without_null);
+        if cfg!(not(feature = "force_hash_collisions")) {
+            let expected = [(0, "c"), (1, "a"), (3, "b")];
+            let expected: Vec<_> =
+                expected.iter().map(|&(p, k)| (p, k.to_string())).collect();
+            assert_eq!(without_null, expected);
+        }
+
+        Ok(())
+    }
+
     #[test]
     fn get_anti_indices_handles_dense_matches() {
         let input = UInt32Array::from(vec![2, 3, 4, 5]);