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,165 @@ case class AddIndexExec(
       UTF8String.fromString(indexName))))
   }
 
+  /**
+   * ZONEMAP distributed build.
+   *
+   * <p>Each fragment is indexed in a separate Spark task, but unlike the general
+   * `FragmentBasedIndexJob` path the tasks do NOT share an index UUID. Sharing would force every
+   * task into the same `indices_dir/&lt;uuid&gt;/` directory, where they all attempt to write the
+   * fixed filename `zonemap.lance` and race — only one fragment's data would survive. BTree
+   * fragment-mode is robust under shared UUID because per-partition writes use distinct file
+   * names (`part_&lt;id&gt;_page_data.lance`, `part_&lt;id&gt;_page_lookup.lance`) and a merge
+   * step consolidates them. ZoneMap has no per-task file-name namespacing and no merge step.
+   *
+   * <p>Approach: each task generates its own UUID (lance-core does this when `withIndexUUID` is
+   * not set; the JNI `skip_commit = fragment_ids.is_some()` branch takes
+   * `execute_uncommitted`, returning IndexMetadata without committing). The driver collects all
+   * N per-task results and commits them as a single transaction with N IndexMetadata entries
+   * sharing the index name. Read consumers (`describe_indices` chunk-by-name, `getZonemapStats`
+   * iterating `load_indices_by_name`) handle the multi-segment shape transparently.
+   *
+   * @param dataset      the open Lance dataset; this method is responsible for closing it
+   * @param lanceDataset the V2 catalog table view (used to resolve namespace credentials)
+   * @param readOptions  read configuration to forward to executor tasks
+   * @param fragmentIds  fragment IDs to index, one per task
+   */
+  private def runZonemapDistributed(
+      dataset: Dataset,
+      lanceDataset: LanceDataset,
+      readOptions: LanceSparkReadOptions,
+      fragmentIds: List[Integer]): Seq[InternalRow] = {
+    try {
+      // Validate columns up-front on the driver so missing-column errors surface uniformly,
+      // without having to wait for every executor task to fail. Reused at commit time below to
+      // populate the new IndexMetadata entries — schema cannot drift within one dataset handle,
+      // so a single resolution suffices.
+      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()
+
+      // Build one Index entry per fragment, all sharing the same name. lance-core groups by
+      // name in describe_indices and per-segment reads serve the multi-segment shape.
+      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)
+  }
+
+  /**
+   * Look up Lance field IDs for the given column names, throwing IllegalArgumentException with a
+   * single-sourced error message if any column is absent. Shared by the parallel and ZoneMap build
+   * paths so that 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 +488,111 @@ case class FragmentIndexTask(
   }
 }
 
+/**
+ * Per-fragment build metadata returned by {@link ZonemapFragmentTask}.
+ *
+ * <p>Unlike the shared-UUID `FragmentIndexTask` flow, ZONEMAP requires every task to return its
+ * own UUID and fragment ID so the driver can build N independent IndexMetadata entries — one per
+ * fragment — and commit them in a single transaction. lance-core's `describe_indices` groups
+ * segments by name into one logical description, and `getZonemapStats` reads each segment's
+ * `zonemap.lance` directly via `load_indices_by_name`.
+ */
+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 the JNI path takes `execute_uncommitted` (per
+ * `should_skip_commit`/`fragment_ids.is_some()` in lance-core's blocking_dataset.rs) and
+ * lance-core generates a fresh per-task UUID. Each task writes to its own `<uuid>/zonemap.lance`
+ * directory and returns the metadata for the driver to bundle 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 +802,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 +811,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")
     }
   }