feat(sql): support USING zonemap in CREATE INDEX

Adds `ALTER TABLE ... CREATE INDEX ... USING zonemap (col)` end-to-end with
a properly distributed multi-segment build.

Two production-side changes:

1. IndexUtils now recognises "zonemap" as a method name in both directions
   (buildIndexType → IndexType.ZONEMAP, buildScalarIndexParamType → "zonemap")
   with case-insensitive lookup.

2. AddIndexExec routes ZONEMAP through a new runZonemapDistributed path.
   Unlike the FragmentBasedIndexJob shared-UUID pattern (which BTree uses
   correctly because per-partition writes use distinct file names —
   part_<id>_page_data.lance, part_<id>_page_lookup.lance — and a merge step
   consolidates them), ZoneMap writes a fixed `zonemap.lance` filename and has
   no merge step. Sharing one UUID across N executor tasks would race on the
   same object-store path; only one fragment's data would survive.

   Each Spark task now calls dataset.createIndex with withFragmentIds=[id]
   and NO withIndexUUID. The lance-core JNI takes the execute_uncommitted
   path (skip_commit = fragment_ids.is_some()) and generates a fresh
   per-task UUID, so each task writes to its own indices_dir/<uuid>/zonemap.lance
   directory. The driver collects all N per-task ZonemapFragmentResult values
   (uuid, fragmentId, indexDetails, indexVersion, createdAt) and commits them
   as a single AddIndexOperation transaction with N IndexMetadata entries
   sharing the index name. lance-core's existing read-side infrastructure —
   describe_indices chunking by name, getZonemapStats iterating
   load_indices_by_name and reading each segment's zonemap.lance — handles
   the multi-segment shape transparently.

   mergeIndexMetadata is intentionally skipped for ZONEMAP: there is no
   per-fragment file consolidation needed, and lance-core's merge_index_metadata
   has no ZoneMap arm anyway (would throw "Unsupported index type").

Supporting refactors:

- Extract resolveFieldIdsOrThrow(dataset, columns) — shared by the parallel
  path's post-build commit and the new distributed ZoneMap path. Single source
  of truth for the "Cannot find index column in Lance schema" error.
- Extract extractNamespaceInfo(lanceDataset, readOptions) — shared between
  createIndexJob and runZonemapDistributed.
- ZonemapFragmentTask wraps execute() in try/catch that re-throws with
  fragment-id context, preserving the specific exception subclass
  (IllegalArgumentException, IllegalStateException, RuntimeException) so
  callers matching on type still match.
- Driver-side decoding of per-task results wraps decode failures with
  task-index context so deploy-skew / serialization issues are diagnosable.
- runZonemapBuild validates that createdIndex.indexDetails() is non-empty,
  mirroring the parallel path's extractIndexBuildResult guard.

Tests:

IndexUtilsTest (6 cases) — symbol-mapping unit tests for buildIndexType /
buildScalarIndexParamType: forward and reverse mappings, case-insensitive
lookup, unknown-method rejection with message substring assertions.

BaseAddIndexTest additions (5 zonemap cases):

- testCreateZonemapIndex — end-to-end USING zonemap + strict per-fragment
  coverage assertion (every indexed fragment must contribute ≥1 zone).
- testCreateZonemapOnNonExistentColumn — fail-fast IllegalArgumentException
  with "Cannot find index column" message substring.
- testCreateZonemapOnStringColumn — strict assertion that every zone's
  min/max is a non-null String (catches string-codec regressions).
- testCreateZonemapWithZoneSize — smoke test for `with (zone_size=N)`
  parameter forwarding through IndexUtils.toJson.
- testZonemapDistributedCommitShape — locks the multi-segment commit
  invariants: exactly one IndexMetadata per fragment, every segment has a
  distinct UUID, every segment's fragment-bitmap is a singleton, the union
  of segment fragment-bitmaps equals the indexed fragment set, all segments
  share one field-id list. A regression to shared-UUID single-segment
  commits — the original race — would fail every assertion in this test.

Verified end-to-end against lance-core 6.0.0-rc.2 (current upstream/main).
Closes #512. Closes #514.

Author: LuciferYang

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

@@ -82,26 +82,22 @@ case class AddIndexExec(
       return Seq(new GenericInternalRow(Array[Any](0L, UTF8String.fromString(indexName))))
     }
 
-    val uuid = UUID.randomUUID()
     val indexType = IndexUtils.buildIndexType(method)
-
     val dataset = Utils.openDatasetBuilder(readOptions).build()
 
+    if (indexType == IndexType.ZONEMAP) {
+      return runZonemapDistributed(dataset, lanceDataset, readOptions, fragmentIds)
+    }
+
+    val uuid = UUID.randomUUID()
     val indexBuildResult =
       createIndexJob(dataset, lanceDataset, readOptions, uuid.toString, fragmentIds).run()
 
     try {
-      // Merge index metadata after all fragments are indexed
+      // Merge index metadata after all fragments are indexed.
       dataset.mergeIndexMetadata(uuid.toString, indexType, Optional.empty())
 
-      val fieldIdByName = dataset.getLanceSchema.fields().asScala
-        .map(f => f.getName -> f.getId)
-        .toMap
-      val fieldIds = columns.map { column =>
-        fieldIdByName.getOrElse(
-          column,
-          throw new IllegalArgumentException(s"Cannot find index column in Lance schema: $column"))
-      }.toList
+      val fieldIds = resolveFieldIdsOrThrow(dataset, columns)
 
       val datasetVersion = dataset.version()
 
@@ -148,26 +144,152 @@ case class AddIndexExec(
       UTF8String.fromString(indexName))))
   }
 
+  /**
+   * Distributed ZONEMAP build. Each task indexes one fragment with a fresh per-task UUID (no
+   * `withIndexUUID`), so each writes to its own `<uuid>/zonemap.lance` directory. The driver
+   * commits N IndexMetadata entries under a shared name in one transaction; lance-core's read
+   * path serves multi-segment indexes natively.
+   *
+   * <p>A shared UUID would race on the fixed `zonemap.lance` filename — ZoneMap, unlike BTree, has
+   * no per-task file-name namespacing or merge step.
+   *
+   * @param dataset      open Lance dataset; this method closes it
+   * @param lanceDataset V2 catalog view, for namespace credentials
+   * @param readOptions  read config forwarded to executor tasks
+   * @param fragmentIds  fragments to index, one task each
+   */
+  private def runZonemapDistributed(
+      dataset: Dataset,
+      lanceDataset: LanceDataset,
+      readOptions: LanceSparkReadOptions,
+      fragmentIds: List[Integer]): Seq[InternalRow] = {
+    try {
+      // Validate columns up-front (fail-fast before parallelize) and reuse at commit time —
+      // schema cannot drift within one dataset handle.
+      val fieldIds = resolveFieldIdsOrThrow(dataset, columns)
+
+      val (nsImpl, nsProps, tableId, initialStorageOpts) =
+        extractNamespaceInfo(lanceDataset, readOptions)
+      val encodedReadOptions = encode(readOptions)
+      val argsJson = IndexUtils.toJson(args)
+      val tasks = fragmentIds.map { fid =>
+        ZonemapFragmentTask(
+          encodedReadOptions,
+          columns.toList,
+          method,
+          argsJson,
+          indexName,
+          fid,
+          nsImpl,
+          nsProps,
+          tableId,
+          initialStorageOpts)
+      }
+
+      val encodedResults: Array[String] = session.sparkContext
+        .parallelize(tasks, tasks.size)
+        .map(_.execute())
+        .collect()
+      val perFragment: Array[ZonemapFragmentResult] = encodedResults.zipWithIndex.map {
+        case (encoded, idx) =>
+          try {
+            decode[ZonemapFragmentResult](encoded)
+          } catch {
+            case e: Exception =>
+              throw new IllegalStateException(
+                s"Failed to decode ZONEMAP build result at task index $idx: ${e.getMessage}",
+                e)
+          }
+      }
+
+      val datasetVersion = dataset.version()
+
+      // One Index entry per fragment, all sharing the index name; the read path groups by name.
+      val newIndexes = perFragment.toList.map { r =>
+        val builder = Index.builder()
+          .uuid(UUID.fromString(r.uuid))
+          .name(indexName)
+          .fields(fieldIds.map(java.lang.Integer.valueOf).asJava)
+          .datasetVersion(datasetVersion)
+          .indexDetails(r.indexDetails)
+          .indexVersion(r.indexVersion)
+          .indexType(IndexType.ZONEMAP)
+          .fragments(Collections.singletonList(java.lang.Integer.valueOf(r.fragmentId)))
+        r.createdAt.foreach(builder.createdAt)
+        builder.build()
+      }.asJava
+
+      val removedIndices = dataset.getIndexes.asScala
+        .filter(_.name() == indexName)
+        .toList.asJava
+
+      val op = AddIndexOperation.builder()
+        .withNewIndices(newIndexes)
+        .withRemovedIndices(removedIndices)
+        .build()
+      val txn = new Transaction.Builder()
+        .readVersion(datasetVersion)
+        .operation(op)
+        .build()
+      try {
+        val newDataset = new CommitBuilder(dataset)
+          .writeParams(readOptions.getStorageOptions)
+          .execute(txn)
+        newDataset.close()
+      } finally {
+        txn.close()
+      }
+    } finally {
+      dataset.close()
+    }
+    Seq(new GenericInternalRow(Array[Any](
+      fragmentIds.size.toLong,
+      UTF8String.fromString(indexName))))
+  }
+
+  /** Extract namespace credentials info from the catalog, mirroring `createIndexJob`. */
+  private def extractNamespaceInfo(
+      lanceDataset: LanceDataset,
+      readOptions: LanceSparkReadOptions): (
+      Option[String],
+      Option[Map[String, String]],
+      Option[List[String]],
+      Option[Map[String, String]]) = catalog match {
+    case nsCatalog: BaseLanceNamespaceSparkCatalog =>
+      (
+        Option(nsCatalog.getNamespaceImpl),
+        Option(nsCatalog.getNamespaceProperties).map(_.asScala.toMap),
+        Option(readOptions.getTableId).map(_.asScala.toList),
+        Option(lanceDataset.getInitialStorageOptions).map(_.asScala.toMap))
+    case _ => (None, None, None, None)
+  }
+
+  /**
+   * Resolve column names to Lance field IDs. Throws IllegalArgumentException if any column is
+   * absent; shared by both build paths so failure modes stay uniform.
+   */
+  private def resolveFieldIdsOrThrow(
+      dataset: Dataset,
+      columnsToResolve: Seq[String]): List[Int] = {
+    val fieldIdByName = dataset.getLanceSchema.fields().asScala
+      .map(f => f.getName -> f.getId)
+      .toMap
+    columnsToResolve.map { column =>
+      fieldIdByName.getOrElse(
+        column,
+        throw new IllegalArgumentException(s"Cannot find index column in Lance schema: $column"))
+    }.toList
+  }
+
   private def createIndexJob(
       dataset: Dataset,
       lanceDataset: LanceDataset,
       readOptions: LanceSparkReadOptions,
       uuid: String,
       fragmentIds: List[Integer]): IndexJob = {
     // Get namespace info from catalog if available (for credential vending on workers)
-    val (nsImpl, nsProps, tableId, initialStorageOpts): (
-        Option[String],
-        Option[Map[String, String]],
-        Option[List[String]],
-        Option[Map[String, String]]) = catalog match {
-      case nsCatalog: BaseLanceNamespaceSparkCatalog =>
-        (
-          Option(nsCatalog.getNamespaceImpl),
-          Option(nsCatalog.getNamespaceProperties).map(_.asScala.toMap),
-          Option(readOptions.getTableId).map(_.asScala.toList),
-          Option(lanceDataset.getInitialStorageOptions).map(_.asScala.toMap))
-      case _ => (None, None, None, None)
-    }
+    val (nsImpl, nsProps, tableId, initialStorageOpts) =
+      extractNamespaceInfo(lanceDataset, readOptions)
 
     IndexUtils.buildIndexType(method) match {
       case IndexType.BTREE =>
@@ -353,6 +475,101 @@ case class FragmentIndexTask(
   }
 }
 
+/** Per-fragment build metadata returned by {@link ZonemapFragmentTask}. */
+case class ZonemapFragmentResult(
+    uuid: String,
+    fragmentId: Int,
+    indexDetails: Array[Byte],
+    indexVersion: Int,
+    createdAt: Option[Instant])
+  extends Serializable
+
+/**
+ * Per-fragment ZONEMAP build task. Calls `dataset.createIndex` with `withFragmentIds=[id]` and no
+ * `withIndexUUID`, so lance-core takes the uncommitted path with a fresh per-task UUID. The
+ * returned metadata is collected by the driver into a multi-segment commit.
+ */
+case class ZonemapFragmentTask(
+    encodedReadOptions: String,
+    columns: List[String],
+    method: String,
+    argsJson: String,
+    indexName: String,
+    fragmentId: Int,
+    namespaceImpl: Option[String],
+    namespaceProperties: Option[Map[String, String]],
+    tableId: Option[List[String]],
+    initialStorageOptions: Option[Map[String, String]])
+  extends Serializable {
+
+  def execute(): String = {
+    try {
+      buildAndEncode()
+    } catch {
+      // Preserve the specific exception subclass so callers matching on type still match —
+      // only the message gains fragment-id context.
+      case e: IllegalArgumentException =>
+        throw new IllegalArgumentException(
+          s"ZONEMAP build failed for fragment $fragmentId: ${e.getMessage}",
+          e)
+      case e: IllegalStateException =>
+        throw new IllegalStateException(
+          s"ZONEMAP build failed for fragment $fragmentId: ${e.getMessage}",
+          e)
+      case e: RuntimeException =>
+        throw new RuntimeException(
+          s"ZONEMAP build failed for fragment $fragmentId: ${e.getMessage}",
+          e)
+    }
+  }
+
+  private def buildAndEncode(): String = {
+    val readOptions = decode[LanceSparkReadOptions](encodedReadOptions)
+    val params = IndexParams.builder()
+      .setScalarIndexParams(ScalarIndexParams.create(
+        IndexUtils.buildScalarIndexParamType(method),
+        argsJson))
+      .build()
+
+    val indexOptions = IndexOptions
+      .builder(java.util.Arrays.asList(columns: _*), IndexType.ZONEMAP, params)
+      .replace(true)
+      .withIndexName(indexName)
+      .withFragmentIds(Collections.singletonList(java.lang.Integer.valueOf(fragmentId)))
+      .build()
+
+    val dataset = Utils.openDatasetBuilder(readOptions)
+      .initialStorageOptions(initialStorageOptions.map(_.asJava).orNull)
+      .runtimeNamespace(
+        namespaceImpl.orNull,
+        namespaceProperties.map(_.asJava).orNull,
+        tableId.map(_.asJava).orNull)
+      .build()
+
+    try {
+      val createdIndex = dataset.createIndex(indexOptions)
+      val details = createdIndex.indexDetails()
+      if (!details.isPresent || details.get().length == 0) {
+        throw new IllegalStateException(
+          s"Index ${createdIndex.name()} was created without index details")
+      }
+      val createdAt = if (createdIndex.createdAt().isPresent) {
+        Some(createdIndex.createdAt().get())
+      } else {
+        None
+      }
+      encode(ZonemapFragmentResult(
+        uuid = createdIndex.uuid().toString,
+        fragmentId = fragmentId,
+        indexDetails = details.get(),
+        indexVersion = createdIndex.indexVersion(),
+        createdAt = createdAt))
+    } finally {
+      dataset.close()
+    }
+  }
+}
+
 /**
  * A job implementation for creating range-based BTree indexes using preprocessed, globally sorted data.
  * This approach distributes data across multiple partitions based on ranges of values and creates
@@ -562,6 +779,7 @@ object IndexUtils {
     method.toLowerCase match {
       case "btree" => IndexType.BTREE
       case "fts" => IndexType.INVERTED
+      case "zonemap" => IndexType.ZONEMAP
       case other => throw new UnsupportedOperationException(s"Unsupported index method: $other")
     }
   }
@@ -570,6 +788,7 @@ object IndexUtils {
     method.toLowerCase match {
       case "btree" => "btree"
       case "fts" => "inverted"
+      case "zonemap" => "zonemap"
       case other => throw new UnsupportedOperationException(s"Unsupported index method: $other")
     }
   }

lance-spark-base_2.12/src/test/java/org/lance/spark/index/IndexUtilsTest.java

@@ -0,0 +1,72 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.lance.spark.index;
+
+import org.lance.index.IndexType;
+
+import org.apache.spark.sql.execution.datasources.v2.IndexUtils;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class IndexUtilsTest {
+
+  @Test
+  public void testBuildIndexTypeKnown() {
+    assertEquals(IndexType.BTREE, IndexUtils.buildIndexType("btree"));
+    assertEquals(IndexType.INVERTED, IndexUtils.buildIndexType("fts"));
+    assertEquals(IndexType.ZONEMAP, IndexUtils.buildIndexType("zonemap"));
+  }
+
+  @Test
+  public void testBuildIndexTypeCaseInsensitive() {
+    assertEquals(IndexType.ZONEMAP, IndexUtils.buildIndexType("ZoneMap"));
+    assertEquals(IndexType.BTREE, IndexUtils.buildIndexType("BTREE"));
+    assertEquals(IndexType.INVERTED, IndexUtils.buildIndexType("FTS"));
+  }
+
+  @Test
+  public void testBuildIndexTypeUnknown() {
+    UnsupportedOperationException ex =
+        assertThrows(UnsupportedOperationException.class, () -> IndexUtils.buildIndexType("hash"));
+    assertTrue(ex.getMessage().contains("Unsupported index method"));
+    assertTrue(ex.getMessage().contains("hash"));
+  }
+
+  @Test
+  public void testBuildScalarIndexParamTypeKnown() {
+    assertEquals("btree", IndexUtils.buildScalarIndexParamType("btree"));
+    assertEquals("inverted", IndexUtils.buildScalarIndexParamType("fts"));
+    assertEquals("zonemap", IndexUtils.buildScalarIndexParamType("zonemap"));
+  }
+
+  @Test
+  public void testBuildScalarIndexParamTypeCaseInsensitive() {
+    assertEquals("zonemap", IndexUtils.buildScalarIndexParamType("ZoneMap"));
+    assertEquals("btree", IndexUtils.buildScalarIndexParamType("BTREE"));
+    assertEquals("inverted", IndexUtils.buildScalarIndexParamType("FTS"));
+  }
+
+  @Test
+  public void testBuildScalarIndexParamTypeUnknown() {
+    UnsupportedOperationException ex =
+        assertThrows(
+            UnsupportedOperationException.class,
+            () -> IndexUtils.buildScalarIndexParamType("hash"));
+    assertTrue(ex.getMessage().contains("Unsupported index method"));
+    assertTrue(ex.getMessage().contains("hash"));
+  }
+}