fix: reject Rewrite vs CreateIndex when FRI groups straddle bitmap (#6610)

## Summary

`check_rewrite_txn` returned `Ok` for any `(None, Some(_))` — a Rewrite
with a deferred-remap FRI committing against a concurrently committed
CreateIndex — on the assumption that the FRI would correct index bitmaps
post-commit. That is only safe when each rewrite group's old fragments
are fully inside or fully outside the new index's bitmap. If a group
straddles (some fragments indexed, some not), `load_indices` later fails
with `split of indexed and non-indexed data` and queries against the
index break.

This PR rejects the straddling case as a retryable conflict so the
rewrite is replanned against the post-CreateIndex view.

Production trigger (with `defer_index_remap=true` now enabled):
1. Compaction plans `[frag1, frag2]` while both are unindexed.
2. `optimize_indices` commits a CreateIndex covering `frag1` only (ran
on an older version that didn't see `frag2`).
3. Compaction's Rewrite commits against a read_version earlier than the
CreateIndex; the conflict resolver incorrectly returns COMPATIBLE.
4. Queries fail: the FRI group `[frag1, frag2]` straddles the new index
bitmap.

Also fixes an `.unwrap()` in `remap_fragment_bitmap` so the underlying
error propagates instead of panicking.

## Test plan

- [x] New test `test_rewrite_fri_vs_create_index_conflict` in
`rust/lance/src/dataset/optimize.rs` reproduces the production sequence
and asserts `commit_compaction` returns `RetryableCommitConflict`
(before the fix, commit succeeded and the query failed with "split of
indexed and non-indexed data").
- [x] Existing `dataset::optimize::` and `io::commit::` tests pass.
- [x] `cargo clippy -p lance --tests -- -D warnings`

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: wjones127

rust/lance/src/dataset/optimize.rs

@@ -4564,4 +4564,104 @@ mod tests {
         // validate() should have turned off materialize_deletions since threshold >= 1.0
         assert!(!plan.options.materialize_deletions);
     }
+
+    // check_rewrite_txn takes the (None, Some(_)) branch when a Rewrite with
+    // defer_index_remap=true is committed against a previously committed
+    // CreateIndex, declaring COMPATIBLE without verifying that the Rewrite's
+    // FRI groups don't straddle the CreateIndex's fragment bitmap. When a
+    // group mixes indexed and unindexed fragments, commit succeeds and later
+    // queries fail at load_indices with "split of indexed and non-indexed
+    // data".
+    #[tokio::test]
+    async fn test_rewrite_fri_vs_create_index_conflict() {
+        use crate::index::DatasetIndexExt;
+        use crate::index::vector::VectorIndexParams;
+        use futures::TryStreamExt;
+        use lance_datagen::{BatchCount, Dimension, RowCount, array, gen_batch};
+        use lance_index::IndexType;
+        use lance_linalg::distance::MetricType;
+
+        async fn append_fragment(uri: &str, rows: u64) -> Dataset {
+            let reader = gen_batch()
+                .col("vec", array::rand_vec::<Float32Type>(Dimension::from(16)))
+                .into_reader_rows(RowCount::from(rows), BatchCount::from(1));
+            let params = WriteParams {
+                max_rows_per_file: rows as usize,
+                mode: WriteMode::Append,
+                ..Default::default()
+            };
+            Dataset::write(reader, uri, Some(params)).await.unwrap()
+        }
+
+        let tmpdir = TempStrDir::default();
+        let uri = format!("file://{}", tmpdir.as_str());
+
+        // frag0 (256 rows) with a base IVF index.
+        let reader = gen_batch()
+            .col("vec", array::rand_vec::<Float32Type>(Dimension::from(16)))
+            .into_reader_rows(RowCount::from(256), BatchCount::from(1));
+        let mut dataset = Dataset::write(
+            reader,
+            &uri,
+            Some(WriteParams {
+                max_rows_per_file: 256,
+                mode: WriteMode::Overwrite,
+                ..Default::default()
+            }),
+        )
+        .await
+        .unwrap();
+        let index_params = VectorIndexParams::ivf_pq(2, 8, 2, MetricType::L2, 50);
+        dataset
+            .create_index(&["vec"], IndexType::Vector, None, &index_params, true)
+            .await
+            .unwrap();
+
+        // Append frag1 (unindexed), snapshot a stale handle pointing here,
+        // then append frag2 (also unindexed).
+        dataset = append_fragment(&uri, 64).await;
+        let mut stale = dataset.clone();
+        dataset = append_fragment(&uri, 64).await;
+
+        // Plan + execute compaction of frag1+frag2 with deferred remap.
+        let options = CompactionOptions {
+            defer_index_remap: true,
+            ..Default::default()
+        };
+        let plan = plan_compaction(&dataset, &options).await.unwrap();
+        assert!(!plan.tasks.is_empty());
+        let snapshot = dataset.clone();
+        let completed: Vec<RewriteResult> = futures::stream::iter(plan.tasks.into_iter())
+            .map(|task| rewrite_files(Cow::Borrowed(&snapshot), task, &options))
+            .buffer_unordered(1)
+            .try_collect()
+            .await
+            .unwrap();
+
+        // optimize_indices on the stale handle indexes frag1 only (frag2
+        // didn't exist at that version), commits as CreateIndex. `dataset`
+        // stays at its pre-optimize version so the Rewrite commit has to
+        // conflict-check against this CreateIndex.
+        stale
+            .optimize_indices(&lance_index::optimize::OptimizeOptions::append())
+            .await
+            .unwrap();
+
+        // Commit the pre-executed Rewrite. The FRI group [frag1, frag2]
+        // straddles the new CreateIndex bitmap (frag1 indexed, frag2 not), so
+        // check_rewrite_txn must reject this as a retryable conflict rather
+        // than letting it commit into a broken state that fails queries.
+        let err = commit_compaction(
+            &mut dataset,
+            completed,
+            Arc::new(DatasetIndexRemapperOptions::default()),
+            &options,
+        )
+        .await
+        .expect_err("commit should fail with retryable conflict");
+        assert!(
+            matches!(err, Error::RetryableCommitConflict { .. }),
+            "unexpected error: {err}"
+        );
+    }
 }

rust/lance/src/index.rs

@@ -923,11 +923,11 @@ impl DatasetIndexExt for Dataset {
                 })
                 .await?;
             let mut indices = indices.as_ref().clone();
-            indices.iter_mut().for_each(|idx| {
+            for idx in indices.iter_mut() {
                 if let Some(bitmap) = idx.fragment_bitmap.as_mut() {
-                    frag_reuse_index.remap_fragment_bitmap(bitmap).unwrap();
+                    frag_reuse_index.remap_fragment_bitmap(bitmap)?;
                 }
-            });
+            }
             Ok(Arc::new(indices))
         } else {
             Ok(indices)

rust/lance/src/io/commit/conflict_resolver.rs

@@ -746,9 +746,38 @@ impl<'a> TransactionRebase<'a> {
                                 .push(committed_fri.clone());
                             Ok(())
                         }
-                        // If rewrite defers index remap,
-                        // then it does not conflict with index creation
-                        (None, Some(_)) => Ok(()),
+                        // If rewrite defers index remap, the FRI handles the
+                        // post-commit bitmap update — but only if each rewrite
+                        // group is fully inside or fully outside each new
+                        // index's fragment bitmap. A group that straddles
+                        // would produce a bitmap with a mix of indexed and
+                        // non-indexed fragments, which load_indices rejects.
+                        (None, Some(_)) => {
+                            for index in new_indices {
+                                let Some(frag_bitmap) = &index.fragment_bitmap else {
+                                    return Err(self
+                                        .retryable_conflict_err(other_transaction, other_version));
+                                };
+                                for group in groups {
+                                    let mut indexed = 0usize;
+                                    let mut unindexed = 0usize;
+                                    for frag in &group.old_fragments {
+                                        if frag_bitmap.contains(frag.id as u32) {
+                                            indexed += 1;
+                                        } else {
+                                            unindexed += 1;
+                                        }
+                                    }
+                                    if indexed > 0 && unindexed > 0 {
+                                        return Err(self.retryable_conflict_err(
+                                            other_transaction,
+                                            other_version,
+                                        ));
+                                    }
+                                }
+                            }
+                            Ok(())
+                        }
                         // Rewrite with remapping and frag_reuse_index creation can commit without conflict
                         (Some(_), None) => {
                             // this should not happen today since we don't support committing