@@ -54,8 +54,8 @@ use datafusion_physical_expr_common::datum::apply_cmp;
 use crate::joins::PartitionMode;
 use crate::joins::SharedBitmapBuilder;
 use crate::joins::hash_join::exec::{JoinLeftData, new_join_hashmap};
-use crate::joins::join_hash_map::JoinHashMapType;
-use crate::joins::join_hash_map::ProbeOffset::AtRow;
+use crate::joins::join_hash_map::ProbeOffset::{AtRow, MidChain};
+use crate::joins::join_hash_map::{JoinHashMapType, ProbeOffset};
 use crate::joins::probe_completion::ProbeSideSummary;
 use crate::joins::utils::{
     BuildProbeJoinMetrics, JoinFilter, apply_join_filter_to_indices,
@@ -991,7 +991,8 @@ pub(super) struct LookupBuffers {
 /// chunks of at most `batch_size` pairs.
 ///
 /// `f` marks rows (see [`UnknownMarks`]), and a marked row cannot become
-/// unmarked. So a cross product drops the marked rows again after each chunk.
+/// unmarked. So the pairing drops the pairs that decide a marked row before
+/// it sends them, and drops the marked rows again after each chunk.
 struct UnmarkedPairs<'a, F> {
     marks: UnknownMarks<'a>,
     batch_size: usize,
@@ -1000,6 +1001,9 @@ struct UnmarkedPairs<'a, F> {
     /// Whether `f` was called since the rows that the pairs decide were last
     /// refreshed.
     marks_since_refresh: bool,
+    /// The matches that [`Self::scope_matches`] dropped because the row they
+    /// decide was marked.
+    skipped_matches: usize,
     f: F,
 }
 
@@ -1014,6 +1018,7 @@ where
             build_chunk: Vec::with_capacity(batch_size),
             probe_chunk: Vec::with_capacity(batch_size),
             marks_since_refresh: false,
+            skipped_matches: 0,
             f,
         }
     }
@@ -1044,6 +1049,14 @@ where
     /// hashes can have different scope keys, so `f` must still check that the
     /// scope keys are equal.
     ///
+    /// The pairing starts with a lookup per probe row, like the join itself.
+    /// When the marks are on the build rows, each probe row still finds all
+    /// the build rows in its scope, marked or not. So a marked build row costs
+    /// a little for every later probe row in its scope. When those costs add
+    /// up to more than a sort of the probe rows, the rest of the probe rows
+    /// are grouped by hash. Each hash is then looked up once, and its group
+    /// is paired like a cross product that drops the build rows as they are
+    /// marked.
     fn scope_matches(
         &mut self,
         scope_map: &dyn JoinHashMapType,
@@ -1057,20 +1070,82 @@ where
             .iter()
             .map(|row| probe_hashes[*row as usize])
             .collect();
+
+        // A sort costs about as much as this many skipped matches per probe
+        // row. A probe row is looked up once, so the marks on probe rows
+        // never make grouping worth it.
+        const SKIPPED_MATCHES_PER_SORTED_ROW: usize = 8;
+        let max_skipped_matches = match self.marks {
+            UnknownMarks::Build(_) => SKIPPED_MATCHES_PER_SORTED_ROW * probe_rows.len(),
+            UnknownMarks::Probe(_) => usize::MAX,
+        };
+        self.skipped_matches = 0;
         let mut offset = AtRow { row: 0 };
-        loop {
-            let next_offset = scope_map.get_matched_indices_with_limit_offset(
+        let group_from = loop {
+            let next_offset = self.lookup_unmarked(
+                scope_map,
                 &hashes,
-                None,
-                self.batch_size,
+                |position| probe_rows[position as usize],
+                offset,
+                &to_build_row,
+                lookup,
+            );
+            for (probe_row, build_row) in
+                lookup.probe_indices.iter().zip(lookup.build_indices.iter())
+            {
+                self.push(*build_row, *probe_row)?;
+            }
+            let Some(next_offset) = next_offset else {
+                return Ok(());
+            };
+            offset = next_offset;
+            if self.skipped_matches > max_skipped_matches {
+                break match next_offset {
+                    AtRow { row } | MidChain { row, .. } => row,
+                };
+            }
+        };
+
+        // Send the pairs that are left, so the lookups below see their marks.
+        self.send()?;
+
+        // Group the probe rows from the one being looked up. Its build rows
+        // that were already paired are paired again, which changes no mark.
+        let mut rest: Vec<(u64, u32)> = hashes[group_from..]
+            .iter()
+            .copied()
+            .zip(probe_rows[group_from..].iter().copied())
+            .collect();
+        rest.sort_unstable_by_key(|(hash, _)| *hash);
+        let groups: Vec<_> = rest.chunk_by(|a, b| a.0 == b.0).collect();
+        let group_hashes: Vec<u64> = groups.iter().map(|group| group[0].0).collect();
+
+        let mut build_rows = vec![];
+        let mut offset = AtRow { row: 0 };
+        loop {
+            // A lookup chunk holds the build rows of one or more groups, in
+            // group order.
+            let next_offset = self.lookup_unmarked(
+                scope_map,
+                &group_hashes,
+                |group| group,
                 offset,
-                &mut lookup.probe_indices,
-                &mut lookup.build_indices,
+                &to_build_row,
+                lookup,
             );
-            for i in 0..lookup.build_indices.len() {
-                self.push(
-                    to_build_row(lookup.build_indices[i]),
-                    probe_rows[lookup.probe_indices[i] as usize],
+            let mut matches = lookup.build_indices.as_slice();
+            for group in lookup.probe_indices.chunk_by(|a, b| a == b) {
+                let group_matches;
+                (group_matches, matches) = matches.split_at(group.len());
+                build_rows.clear();
+                build_rows.extend_from_slice(group_matches);
+                // A build row belongs to one group only, so the marks made
+                // for the earlier groups do not concern these build rows.
+                self.marks_since_refresh = false;
+                self.pair_unmarked(
+                    &mut build_rows,
+                    groups[group[0] as usize].iter().map(|(_, row)| *row),
+                    |b, p| (b, p),
                 )?;
             }
             let Some(next_offset) = next_offset else {
@@ -1080,6 +1155,49 @@ where
         }
     }
 
+    /// Looks up the next chunk of matches of `hashes` in `scope_map`, from
+    /// `offset`, into the buffers: the position in `hashes` translated with
+    /// `to_probe_row`, and the build row translated with `to_build_row`. Drops the
+    /// matches that decide a marked row, and adds their number to
+    /// `skipped_matches`.
+    fn lookup_unmarked(
+        &mut self,
+        scope_map: &dyn JoinHashMapType,
+        hashes: &[u64],
+        to_probe_row: impl Fn(u32) -> u32,
+        offset: ProbeOffset,
+        to_build_row: impl Fn(u64) -> u64,
+        lookup: &mut LookupBuffers,
+    ) -> Option<ProbeOffset> {
+        let LookupBuffers {
+            probe_indices,
+            build_indices,
+        } = lookup;
+        let next_offset = scope_map.get_matched_indices_with_limit_offset(
+            hashes,
+            None,
+            self.batch_size,
+            offset,
+            probe_indices,
+            build_indices,
+        );
+        let bitmap = self.marks.bitmap().lock();
+        let mut kept = 0;
+        for i in 0..build_indices.len() {
+            let build = to_build_row(build_indices[i]);
+            let probe = to_probe_row(probe_indices[i]);
+            if !bitmap.get_bit(self.marks.unknown_row(build, probe)) {
+                build_indices[kept] = build;
+                probe_indices[kept] = probe;
+                kept += 1;
+            }
+        }
+        self.skipped_matches += build_indices.len() - kept;
+        build_indices.truncate(kept);
+        probe_indices.truncate(kept);
+        next_offset
+    }
+
     /// Sends the pairs that are left.
     fn finish(mut self) -> Result<()> {
         self.send()
@@ -1213,12 +1331,13 @@ mod tests {
         Ok(pairs.into_iter().max().unwrap_or(0))
     }
 
-    /// A correlation scope with many build rows and many probe rows. A lookup
-    /// per probe row pairs every build row once per probe row in its scope,
-    /// even after the build row is marked. So the pairs made for one build
-    /// row grow with the number of probe rows.
+    /// A correlation scope with many build rows and many probe rows: once a
+    /// build row is marked, the later probe rows in its scope must not pair
+    /// with it again. So the pairs made for one build row must not grow with
+    /// the number of probe rows. A lookup per probe row pairs every build row
+    /// once per probe row in its scope.
     #[test]
-    fn scope_lookup_pairs_grow_with_probe_rows() -> Result<()> {
+    fn unmarked_pairs_do_not_grow_with_probe_rows() -> Result<()> {
         // Build rows 0..60 are in the scope with hash 1, rows 60..100 in the
         // scope with hash 2. The probe rows alternate between both scopes.
         let build_hashes: Vec<u64> = (0..100).map(|row| 1 + (row >= 60) as u64).collect();
@@ -1229,8 +1348,10 @@ mod tests {
                 max_pairs_per_build_row(&build_hashes, &probe_hashes, batch_size)
             });
             let (few, many) = (few?, many?);
-            assert_eq!(few, 500, "batch_size {batch_size}");
-            assert_eq!(many, 5_000, "batch_size {batch_size}");
+            assert!(
+                many <= few,
+                "batch_size {batch_size}: {few} pairs for 1000 probe rows, {many} for 10000"
+            );
         }
         Ok(())
     }