Fix spurious slice is not sorted assert in Subset::intersect (#971)

CLAUDE.md

@@ -18,4 +18,4 @@ When you edit the files, make sure to respect the following:
 - When running tests, always use the `--release` mode. Alternatively, you can also run `make test`.
 - If your change is performance-critical, use `script/bench.py` as the ground truth to evaluate the performance impact.
 - Keep your documentation concise and avoid duplicate information. The `tidy-diff-docs` skill (`.claude/skills/tidy-diff-docs/`) cleans the doc and code comments in a diff down to the caller-facing contract.
-- Update CHANGELOG.md with a concise bullet when you make major changes (e.g., breaking changes or new features added) in the codebase.
+- Update CHANGELOG.md with a concise bullet when you make major changes (e.g., breaking changes or new features added) in the codebase. Don't update CHANGELOG.md if it's just a bug fix or if it's not a user-facing change. However, major performance improvements (e.g., using a new join algorithm or introducing a new sort of indices) should be documented.

core-relations/src/offsets/mod.rs

@@ -468,6 +468,12 @@ impl Subset {
                 } else if cur_len > other_len {
                     // other is much smaller: iterate other and gallop in cur.
                     // O(other_len * log(cur_len / other_len)) vs O(cur_len) for retain.
+                    // NB: the result is compacted into `cur` in place, so `cur` as a
+                    // whole is transiently unsorted while the loop runs. `write` only
+                    // ever advances to `ci` (a match at `found >= ci` writes to
+                    // `write <= found` and then sets `ci = found + 1`), so the
+                    // *suffix* `cur.0[ci..]` is always untouched, and searching it is
+                    // equivalent to searching all of `cur` from `ci`.
                     let mut write = 0usize;
                     let mut ci = 0usize;
                     #[allow(clippy::needless_range_loop)]
@@ -476,15 +482,19 @@ impl Subset {
                             break;
                         }
                         let target = other_inner[oi];
-                        let result = cur.slice().scan_for_offset(ci, target);
+                        debug_assert!(write <= ci);
+                        // SAFETY: `cur.0[ci..]` is an unmodified suffix of the
+                        // original sorted vector, per the note above.
+                        let suffix = unsafe { SortedOffsetSlice::new_unchecked(&cur.0[ci..]) };
+                        let result = suffix.scan_for_offset(0, target);
                         match result {
                             Ok(found) => {
                                 cur.0[write] = target;
                                 write += 1;
-                                ci = found + 1;
+                                ci += found + 1;
                             }
                             Err(next_ci) => {
-                                ci = next_ci;
+                                ci += next_ci;
                             }
                         }
                     }

core-relations/src/offsets/tests.rs

@@ -73,6 +73,10 @@ fn intersect() {
         Vec::from_iter(0..5),
         Vec::from_iter(0..100), // 20x skew → galloping path
         Vec::from_iter((0..100).filter(|x| x % 7 == 0)),
+        // Both sides Sparse (i.e. non-contiguous) with >4x skew: galloping in
+        // `cur` while compacting it in place.
+        Vec::from_iter((0..6).chain(11..22)),
+        vec![13, 20, 21],
     ];
     let all_elts: Vec<&Vec<usize>> = elts.iter().chain(skewed.iter()).collect();
 
@@ -133,3 +137,31 @@ fn iter_bounded() {
     let expected = Vec::from_iter((2..12).map(|x| RowId::new(x * 2)));
     assert_eq!(got, expected);
 }
+
+#[test]
+fn intersect_sparse_sparse_skewed() {
+    // Regression test for #971. `cur` must be Sparse and >4x longer than
+    // `other` (which must also be Sparse) to reach the "gallop in cur" branch
+    // of `Subset::intersect`.
+    let cur_rows: Vec<usize> = (0..6).chain(11..22).collect();
+    let other_rows: Vec<usize> = vec![13, 20, 21];
+
+    let mut cur = Subset::empty();
+    for row in &cur_rows {
+        cur.add_row_sorted(o(*row));
+    }
+    let mut other = Subset::empty();
+    for row in &other_rows {
+        other.add_row_sorted(o(*row));
+    }
+    assert!(matches!(cur, Subset::Sparse(..)));
+    assert!(matches!(other, Subset::Sparse(..)));
+
+    cur.intersect(
+        other.as_ref(),
+        &with_pool_set(|pool_set| pool_set.get_pool()),
+    );
+    let mut got = Vec::new();
+    cur.offsets(|row| got.push(row.index()));
+    assert_eq!(got, other_rows);
+}