fix: simplify commitIndexSegments to use Lance core's atomic replacement

Lance core's commitExistingIndexSegments already handles atomic
replacement — it finds existing segments with overlapping fragments
and removes them in the same CreateIndex transaction. The Spark-side
manual cleanup (second dataset open + removal transaction) was
redundant and introduced a race condition.

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

Author: beinan

lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AddIndexExec.scala

@@ -203,57 +203,17 @@ case class AddIndexExec(
       UTF8String.fromString(indexName))))
   }
 
-  // Note: There is a narrow race window between commitExistingIndexSegments and
-  // the subsequent dataset re-open to remove old segments. A concurrent writer
-  // could modify the index state in between. Lance's optimistic concurrency
-  // control will surface a conflict error if this happens.
+  // Lance core's commitExistingIndexSegments handles atomic replacement:
+  // it finds existing segments whose fragments overlap with incoming ones
+  // and removes them in the same CreateIndex transaction.
   private def commitIndexSegments(
       readOptions: LanceSparkReadOptions,
       column: String,
       segments: Seq[Index]): Unit = {
     val dataset = Utils.openDatasetBuilder(readOptions).build()
     try {
-      val existingIndices = dataset.getIndexes.asScala
-        .filter(_.name() == indexName)
-        .toList
-
-      val committedSegments =
-        dataset.commitExistingIndexSegments(
-          indexName,
-          column,
-          segments.toList.asJava).asScala.toList
-
-      if (existingIndices.nonEmpty) {
-        val committedUuids = committedSegments.map(_.uuid()).toSet
-        val refreshedDataset = Utils.openDatasetBuilder(readOptions).build()
-        try {
-          val removedIndices = refreshedDataset.getIndexes.asScala
-            .filter(index => index.name() == indexName && !committedUuids.contains(index.uuid()))
-            .toList
-            .asJava
-
-          if (!removedIndices.isEmpty) {
-            val op = AddIndexOperation.builder()
-              .withNewIndices(Collections.emptyList())
-              .withRemovedIndices(removedIndices)
-              .build()
-            val txn = new Transaction.Builder()
-              .readVersion(refreshedDataset.version())
-              .operation(op)
-              .build()
-            try {
-              val newDataset = new CommitBuilder(refreshedDataset)
-                .writeParams(readOptions.getStorageOptions)
-                .execute(txn)
-              newDataset.close()
-            } finally {
-              txn.close()
-            }
-          }
-        } finally {
-          refreshedDataset.close()
-        }
-      }
+      dataset.commitExistingIndexSegments(
+        indexName, column, segments.toList.asJava)
     } finally {
       dataset.close()
     }