fix(index): enforce CreateIndex apply-time invariants and symmetric name conflicts

- validate CreateIndex at manifest build time: indexed fields must still
  exist in the schema (closes the alter_columns type-cast gap that the
  pairwise Project check could not see, since a cast commits Merge with a
  new field id) and non-system removed segments must still be present
  (closes the mirror race where a drop racing a same-name replacement
  silently no-ops)
- collapse the frag-reuse/MemWAL/regular-name conflict booleans into one
  symmetric name-intersection rule covering both directions
- regression tests: staged commit after an alter_columns cast fails as
  incompatible; drop_index racing a replacement fails retryably and
  succeeds on retry; resolver unit test covers all conflict directions

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: ragnorc

rust/lance/src/dataset/transaction.rs

@@ -2131,6 +2131,53 @@ impl Transaction {
                 removed_indices,
             } => {
                 final_fragments.extend(maybe_existing_fragments?.clone());
+
+                // The transaction may have been staged against an older version,
+                // and the pairwise conflict checks cannot enumerate every
+                // hazardous concurrent operation. Enforce the invariants that
+                // make it safe to apply here, where the final manifest is known:
+                // every indexed field must still exist in the schema (a
+                // concurrent projection or column cast drops or re-ids fields),
+                // and every non-system segment staged for removal must still be
+                // present (a concurrent drop or replacement would otherwise be
+                // silently undone). System-index removals are reconciled during
+                // conflict rebasing and are exempt.
+                let field_ids = schema
+                    .fields_pre_order()
+                    .map(|f| f.id)
+                    .collect::<HashSet<_>>();
+                for new_index in new_indices {
+                    if let Some(field_id) = new_index
+                        .fields
+                        .iter()
+                        .find(|field_id| !field_ids.contains(*field_id))
+                    {
+                        return Err(Error::incompatible_transaction_source(
+                            format!(
+                                "CreateIndex: field {} covered by index '{}' no longer \
+                                 exists in the schema",
+                                field_id, new_index.name
+                            )
+                            .into(),
+                        ));
+                    }
+                }
+                if let Some(missing) = removed_indices.iter().find(|removed| {
+                    !is_system_index(removed)
+                        && !final_indices.iter().any(|idx| idx.uuid == removed.uuid)
+                }) {
+                    return Err(Error::retryable_commit_conflict_source(
+                        current_manifest.map(|m| m.version).unwrap_or_default(),
+                        format!(
+                            "CreateIndex: segment {} of index '{}' staged for removal was \
+                             removed by a concurrent transaction; please re-stage against \
+                             the latest version",
+                            missing.uuid, missing.name
+                        )
+                        .into(),
+                    ));
+                }
+
                 let removed_uuids = removed_indices
                     .iter()
                     .map(|old_index| old_index.uuid)

rust/lance/src/index.rs

@@ -1119,11 +1119,13 @@ impl Dataset {
     /// The transaction is a snapshot built against the current dataset version.
     /// Conflicting concurrent changes are rejected at commit time: creating or
     /// dropping a same-name index between staging and commit fails the commit
-    /// with a retryable conflict, and dropping the indexed column fails it as
-    /// incompatible. A concurrent compaction/rewrite that defers index
-    /// remapping is not conflict-checked; the committed segments may then
-    /// cover already-compacted fragments until the deferred remap catches up,
-    /// so commit promptly.
+    /// with a retryable conflict, and a schema change that drops or re-ids the
+    /// indexed column (e.g. [`Dataset::drop_columns`] or an
+    /// [`Dataset::alter_columns`] type cast) fails it as incompatible. A
+    /// concurrent compaction/rewrite that defers index remapping is not
+    /// conflict-checked; the committed segments may then cover
+    /// already-compacted fragments until the deferred remap catches up, so
+    /// commit promptly.
     ///
     /// # Side effects
     ///
@@ -7663,6 +7665,145 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_build_existing_index_segments_transaction_conflicts_with_column_cast() {
+        use crate::dataset::{ColumnAlteration, CommitBuilder};
+        use lance_datagen::{BatchCount, RowCount, array};
+
+        let test_dir = tempfile::tempdir().unwrap();
+        let reader = lance_datagen::gen_batch()
+            .col("id", array::step::<arrow_array::types::Int32Type>())
+            .col(
+                "vector",
+                array::rand_vec::<arrow_array::types::Float32Type>(8.into()),
+            )
+            .into_reader_rows(RowCount::from(10), BatchCount::from(1));
+        let mut dataset = Dataset::write(reader, test_dir.path().to_str().unwrap(), None)
+            .await
+            .unwrap();
+        let field_id = dataset.schema().field("id").unwrap().id;
+
+        let seg = write_vector_segment_metadata(
+            &dataset,
+            "id_idx",
+            field_id,
+            Uuid::new_v4(),
+            [0_u32],
+            b"seg",
+        )
+        .await;
+        let transaction = dataset
+            .build_existing_index_segments_transaction(
+                "id_idx",
+                "id",
+                vec![segment_from_metadata(&seg)],
+            )
+            .await
+            .unwrap();
+
+        // A type cast commits Operation::Merge and assigns the column a new
+        // field id, so the staged transaction's field no longer exists.
+        dataset
+            .alter_columns(&[
+                ColumnAlteration::new("id".into()).cast_to(arrow_schema::DataType::Int64)
+            ])
+            .await
+            .unwrap();
+        assert!(dataset.schema().field_by_id(field_id).is_none());
+
+        let err = CommitBuilder::new(Arc::new(dataset))
+            .execute(transaction)
+            .await
+            .expect_err("staged index commit for a re-identified column must fail");
+        assert!(
+            matches!(err, Error::IncompatibleTransaction { .. }),
+            "expected an incompatible transaction error, got: {err}"
+        );
+        assert!(
+            err.to_string().contains("no longer exists in the schema"),
+            "conflict message must identify the missing field, got: {err}"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_drop_index_conflicts_with_concurrent_replacement() {
+        use lance_datagen::{BatchCount, RowCount, array};
+
+        let test_dir = tempfile::tempdir().unwrap();
+        let reader = lance_datagen::gen_batch()
+            .col("id", array::step::<arrow_array::types::Int32Type>())
+            .col(
+                "vector",
+                array::rand_vec::<arrow_array::types::Float32Type>(8.into()),
+            )
+            .into_reader_rows(RowCount::from(10), BatchCount::from(1));
+        let mut dataset = Dataset::write(reader, test_dir.path().to_str().unwrap(), None)
+            .await
+            .unwrap();
+        let field_id = dataset.schema().field("vector").unwrap().id;
+
+        let initial = write_vector_segment_metadata(
+            &dataset,
+            "vector_idx",
+            field_id,
+            Uuid::new_v4(),
+            [0_u32],
+            b"initial",
+        )
+        .await;
+        dataset
+            .commit_existing_index_segments(
+                "vector_idx",
+                "vector",
+                vec![segment_from_metadata(&initial)],
+            )
+            .await
+            .unwrap();
+
+        // A stale handle prepares to drop while the index is replaced.
+        let mut stale = dataset.clone();
+        let replacement = write_vector_segment_metadata(
+            &dataset,
+            "vector_idx",
+            field_id,
+            Uuid::new_v4(),
+            [0_u32],
+            b"replacement",
+        )
+        .await;
+        dataset
+            .commit_existing_index_segments(
+                "vector_idx",
+                "vector",
+                vec![segment_from_metadata(&replacement)],
+            )
+            .await
+            .unwrap();
+
+        // The stale drop must conflict instead of silently no-op'ing (its
+        // removal targets a UUID the replacement already removed).
+        let err = stale
+            .drop_index("vector_idx")
+            .await
+            .expect_err("a drop racing a same-name replacement must conflict");
+        assert!(
+            matches!(err, Error::RetryableCommitConflict { .. }),
+            "expected a retryable commit conflict, got: {err}"
+        );
+
+        // Retrying against the latest version drops the replacement for real.
+        stale.checkout_latest().await.unwrap();
+        stale.drop_index("vector_idx").await.unwrap();
+        assert!(
+            stale
+                .load_indices_by_name("vector_idx")
+                .await
+                .unwrap()
+                .is_empty(),
+            "the retried drop must remove the replaced index"
+        );
+    }
+
     #[tokio::test]
     async fn test_resolve_index_column_error_cases() {
         use lance_datagen::{BatchCount, RowCount, array};

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

@@ -592,44 +592,24 @@ impl<'a> TransactionRebase<'a> {
                     new_indices: created_indices,
                     removed_indices: other_removed_indices,
                 } => {
-                    let self_has_frag_reuse = new_indices
-                        .iter()
-                        .any(|idx| idx.name == FRAG_REUSE_INDEX_NAME);
-                    // Scan the other transaction's removals too: a removal-only
-                    // transaction (e.g. drop_index) that dropped a singleton must
-                    // not be silently undone by a staged re-creation.
-                    let other_has_frag_reuse = created_indices
-                        .iter()
-                        .chain(other_removed_indices.iter())
-                        .any(|idx| idx.name == FRAG_REUSE_INDEX_NAME);
-                    let self_has_mem_wal =
-                        new_indices.iter().any(|idx| idx.name == MEM_WAL_INDEX_NAME);
-                    let other_has_mem_wal = created_indices
-                        .iter()
-                        .chain(other_removed_indices.iter())
-                        .any(|idx| idx.name == MEM_WAL_INDEX_NAME);
-                    let has_regular_name_conflict = new_indices
-                        .iter()
-                        .filter(|idx| {
-                            idx.name != FRAG_REUSE_INDEX_NAME && idx.name != MEM_WAL_INDEX_NAME
-                        })
-                        .any(|new_index| {
-                            created_indices
-                                .iter()
-                                .any(|created_index| created_index.name == new_index.name)
-                                // A concurrent transaction that removed a same-name
-                                // index (e.g. drop_index) invalidates this snapshot:
-                                // committing anyway would silently re-create the
-                                // index that was just removed.
-                                || other_removed_indices
+                    // Two index transactions conflict when they touch any index
+                    // name in common, in either direction: committing a creation
+                    // over a concurrent same-name removal would resurrect the
+                    // dropped index, and committing a removal over a concurrent
+                    // same-name replacement would silently no-op the drop (its
+                    // stale UUIDs no longer match anything at apply time).
+                    // System singletons need no special casing under this rule.
+                    let names_intersect =
+                        new_indices
+                            .iter()
+                            .chain(removed_indices.iter())
+                            .any(|self_index| {
+                                created_indices
                                     .iter()
-                                    .any(|removed_index| removed_index.name == new_index.name)
-                        });
-
-                    if (self_has_frag_reuse && other_has_frag_reuse)
-                        || (self_has_mem_wal && other_has_mem_wal)
-                        || has_regular_name_conflict
-                    {
+                                    .chain(other_removed_indices.iter())
+                                    .any(|other_index| other_index.name == self_index.name)
+                            });
+                    if names_intersect {
                         Err(self.retryable_conflict_err(other_transaction, other_version))
                     } else {
                         Ok(())
@@ -4669,7 +4649,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_create_index_conflicts_with_concurrent_index_removal() {
+    async fn test_index_transactions_conflict_on_shared_names() {
         fn index_meta(name: &str) -> IndexMetadata {
             IndexMetadata {
                 uuid: Uuid::new_v4(),
@@ -4684,44 +4664,60 @@ mod tests {
                 files: None,
             }
         }
+        fn create_op(name: &str) -> Operation {
+            Operation::CreateIndex {
+                new_indices: vec![index_meta(name)],
+                removed_indices: vec![],
+            }
+        }
+        fn removal_op(name: &str) -> Operation {
+            Operation::CreateIndex {
+                new_indices: vec![],
+                removed_indices: vec![index_meta(name)],
+            }
+        }
 
         let dataset = test_dataset(10, 1).await;
-        // Regular names and both system singletons: a removal-only concurrent
-        // transaction (e.g. drop_index) must conflict with a staged creation of
-        // the same index instead of letting the commit resurrect it.
+        // Index transactions touching the same name conflict in every
+        // direction: a creation over a concurrent removal must not resurrect
+        // the dropped index, and a removal over a concurrent replacement must
+        // not silently no-op the drop. Regular names and both system
+        // singletons behave identically.
         for name in ["my_idx", FRAG_REUSE_INDEX_NAME, MEM_WAL_INDEX_NAME] {
-            let staged = Transaction::new(
-                dataset.manifest.version,
-                Operation::CreateIndex {
-                    new_indices: vec![index_meta(name)],
-                    removed_indices: vec![],
-                },
-                None,
-            );
-            let removal_only = Transaction::new(
-                dataset.manifest.version,
-                Operation::CreateIndex {
-                    new_indices: vec![],
-                    removed_indices: vec![index_meta(name)],
-                },
-                None,
-            );
+            for (self_op, other_op) in [
+                (create_op(name), removal_op(name)),
+                (removal_op(name), create_op(name)),
+                (removal_op(name), removal_op(name)),
+            ] {
+                let staged = Transaction::new(dataset.manifest.version, self_op, None);
+                let committed = Transaction::new(dataset.manifest.version, other_op, None);
+                let mut rebase = TransactionRebase::try_new(&dataset, staged, None)
+                    .await
+                    .unwrap();
+                let err = rebase
+                    .check_txn(&committed, dataset.manifest.version + 1)
+                    .unwrap_err();
+                assert!(
+                    matches!(err, Error::RetryableCommitConflict { .. }),
+                    "{name}: expected a retryable commit conflict, got: {err}"
+                );
+                assert!(
+                    err.to_string()
+                        .contains("preempted by concurrent transaction CreateIndex"),
+                    "{name}: conflict message must identify the concurrent CreateIndex, got: {err}"
+                );
+            }
 
+            // Transactions on disjoint names never conflict, singletons included.
+            let staged = Transaction::new(dataset.manifest.version, create_op(name), None);
+            let unrelated =
+                Transaction::new(dataset.manifest.version, removal_op("other_idx"), None);
             let mut rebase = TransactionRebase::try_new(&dataset, staged, None)
                 .await
                 .unwrap();
-            let err = rebase
-                .check_txn(&removal_only, dataset.manifest.version + 1)
-                .unwrap_err();
-            assert!(
-                matches!(err, Error::RetryableCommitConflict { .. }),
-                "{name}: expected a retryable commit conflict, got: {err}"
-            );
-            assert!(
-                err.to_string()
-                    .contains("preempted by concurrent transaction CreateIndex"),
-                "{name}: conflict message must identify the concurrent CreateIndex, got: {err}"
-            );
+            rebase
+                .check_txn(&unrelated, dataset.manifest.version + 1)
+                .unwrap_or_else(|err| panic!("{name}: disjoint names must not conflict: {err}"));
         }
     }
 }