From 11e965934713dae65037c4ae9005eee5cfc3fbcd Mon Sep 17 00:00:00 2001 From: ivscheianu Date: Sun, 13 Sep 2026 19:34:32 +0300 Subject: [PATCH 1/3] refactor: decouple index build utilities for shared use ScalarSegmentIndexJob now takes a SparkContext and primitive parameters instead of the AddIndexExec instance, so REFRESH INDEX can share the same distributed build path without coupling to CREATE INDEX. Move extractNamespaceInfo and isSystemIndex to IndexUtils, extract parseNumSegments, and add methodForIndexType (reverse IndexType lookup), resolveRetainedSegments (commit-time index re-resolution), requireUniformSegmentDetails (FTS segment compatibility guard), and improve balancedRunLengths to split the heaviest run first via a priority queue instead of left-to-right. --- .../datasources/v2/AddIndexExec.scala | 276 +++++++++++++----- .../datasources/v2/ShowIndexesExec.scala | 9 +- .../datasources/v2/IndexUtilsTest.scala | 220 ++++++++++++++ 3 files changed, 429 insertions(+), 76 deletions(-) diff --git a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AddIndexExec.scala b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AddIndexExec.scala index 68d158743..dc3bfa1eb 100755 --- a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AddIndexExec.scala +++ b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AddIndexExec.scala @@ -18,6 +18,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode import org.apache.arrow.c.{ArrowArrayStream, Data} import org.apache.arrow.vector.VectorSchemaRoot import org.apache.arrow.vector.ipc.ArrowReader +import org.apache.spark.SparkContext import org.apache.spark.internal.Logging import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.{Attribute, GenericInternalRow} @@ -40,7 +41,7 @@ import org.lance.spark.write.SingleBatchArrowReader import java.util.{Collections, Locale, UUID} import scala.collection.JavaConverters._ -import scala.collection.mutable.ArrayBuffer +import scala.collection.mutable.{ArrayBuffer, PriorityQueue} import scala.reflect.ClassTag /** @@ -60,10 +61,10 @@ import scala.reflect.ClassTag *

Deferred training ({@code WITH (train=false)}): commits an empty index on the driver * with an empty fragment bitmap (all rows appear unindexed), skipping data processing. Supported * for all supported scalar index methods. Empty tables use the same path even when - * {@code train=true}, since there are no fragments to train. Populate the index later by re-running - * {@code CREATE INDEX} with the same name (a full distributed build that replaces the empty index) - * or, for incremental coverage of appended fragments, by {@code Dataset.optimizeIndices} (the SQL - * {@code OPTIMIZE} only compacts fragments). {@code num_segments} is rejected with + * {@code train=true}, since there are no fragments to train. Populate the index later with + * {@code REFRESH INDEX} (a distributed build of the fragments the index does not cover) or by + * re-running {@code CREATE INDEX} with the same name (a full distributed rebuild); the SQL + * {@code OPTIMIZE} only compacts fragments. {@code num_segments} is rejected with * {@code train=false}, since no segmented build occurs. * *

The following options are consumed at the Spark execution layer and are never forwarded @@ -126,22 +127,7 @@ case class AddIndexExec( throw new IllegalArgumentException( "num_segments is not supported with train=false: a deferred index performs no segmented build") } - val validatedNumSegments: Option[Int] = numSegmentsOpt.map { arg => - arg.value match { - case null => - throw new IllegalArgumentException( - "num_segments must be a positive integer, got: null") - case n: Number => - val asLong = n.longValue() - if (asLong < 1L || asLong > Int.MaxValue) - throw new IllegalArgumentException( - s"num_segments must be a positive integer that fits in Int, got: $asLong") - asLong.toInt - case other => - throw new IllegalArgumentException( - s"num_segments must be a positive integer, got: $other") - } - } + val validatedNumSegments: Option[Int] = numSegmentsOpt.map(IndexUtils.parseNumSegments) // train=false, or an empty table: commit an empty index on the driver and skip data // processing. Index and option validation above still applies to empty tables. @@ -161,7 +147,7 @@ case class AddIndexExec( } val (nsImpl, nsProps, tableId, initialStorageOpts) = - extractNamespaceInfo(lanceDataset, readOptions) + IndexUtils.extractNamespaceInfo(catalog, lanceDataset, readOptions) if (btreeBuildMode.contains("range")) { val segments = new RangeBasedBTreeIndexJob( @@ -181,7 +167,11 @@ case class AddIndexExec( // Scalar segment indexes use the logical segment commit path. if (scalarSegmentIndexType.isDefined) { val segmentJob = new ScalarSegmentIndexJob( - this.copy(columns = canonicalColumns), + session.sparkContext, + indexName, + method, + canonicalColumns.toList, + IndexUtils.toJson(args), buildReadOptions, fragmentWorkloads, validatedNumSegments, @@ -260,24 +250,6 @@ case class AddIndexExec( } } - 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) - } - } - } /** @@ -498,9 +470,16 @@ case class RangeBTreeIndexBuilder( * A job implementation for creating scalar segment indexes using logical segment commit. * Fragments are batched into segments, each built in parallel, and committed * as a logical index on the driver. + * + * Shared by CREATE INDEX, which passes every fragment, and REFRESH INDEX, which passes only the + * fragments an existing index does not cover. */ class ScalarSegmentIndexJob( - addIndexExec: AddIndexExec, + sc: SparkContext, + indexName: String, + method: String, + columns: List[String], + argsJson: String, readOptions: LanceSparkReadOptions, fragmentWorkloads: List[FragmentWorkload], numSegments: Option[Int], @@ -510,24 +489,22 @@ class ScalarSegmentIndexJob( initialStorageOpts: Option[Map[String, String]]) { def run(): Seq[Index] = { - val indexType = IndexUtils.scalarSegmentIndexType(addIndexExec.method).getOrElse { + val indexType = IndexUtils.scalarSegmentIndexType(method).getOrElse { throw new UnsupportedOperationException( - s"Unsupported Lance index method: ${addIndexExec.method}") + s"Unsupported Lance index method: $method") } val encodedReadOptions = encode(readOptions) - val columns = addIndexExec.columns.toList - val argsJson = IndexUtils.toJson(addIndexExec.args) val fragmentBatches = IndexUtils.batchFragments( fragmentWorkloads, numSegments, - addIndexExec.session.sparkContext.defaultParallelism) + sc.defaultParallelism) val tasks = fragmentBatches.map { batch => ScalarSegmentIndexTask( encodedReadOptions, - addIndexExec.indexName, + indexName, columns, - addIndexExec.method, + method, argsJson, batch, nsImpl, @@ -537,7 +514,7 @@ class ScalarSegmentIndexJob( }.toSeq IndexUtils.runSegmentTasks( - addIndexExec.session.sparkContext, + sc, tasks, s"${indexType.name()} index build failed. Uncommitted segments are not " + "visible to readers and will not affect query correctness.")(_.execute()) @@ -548,6 +525,14 @@ final private[v2] case class FragmentWorkload(fragmentId: Integer, numRows: Long /** * A task to create a scalar index segment on a batch of fragments. + * + * The segment is named after the logical index it will join, and {@code replace} is set because that + * index may already exist: on the uncommitted build path Lance consults {@code replace} only to + * reject a name that is already taken, which is precisely the normal case for a REFRESH, and for a + * CREATE INDEX that replaces an index of the same name. Nothing is removed here — the driver's + * single {@code commitExistingIndexSegments} transaction decides which existing segments to keep. + * Leaving the name unset instead makes Lance derive its own default (`{column}_idx`) and reject the + * build whenever an index of that name exists on the column. */ case class ScalarSegmentIndexTask( encodedReadOptions: String, @@ -644,11 +629,48 @@ object IndexUtils extends Logging { IndexType.RTREE -> "rtree", IndexType.INVERTED -> "inverted") + // Reverse of `methodToIndexTypes`, for commands that resolve an already-created index and need + // the method name back. INVERTED maps to the canonical "fts" spelling rather than its alias. + private val methodByIndexType: Map[IndexType, String] = Map( + IndexType.BTREE -> "btree", + IndexType.ZONEMAP -> "zonemap", + IndexType.BITMAP -> "bitmap", + IndexType.LABEL_LIST -> "label_list", + IndexType.NGRAM -> "ngram", + IndexType.BLOOM_FILTER -> "bloomfilter", + IndexType.RTREE -> "rtree", + IndexType.INVERTED -> "fts") + def scalarSegmentIndexType(method: String): Option[IndexType] = methodToIndexTypes .get(method.toLowerCase(Locale.ROOT)) .filter(scalarSegmentIndexTypes.contains) + /** The SQL method name for an index type that supports segmented builds, if any. */ + def methodForIndexType(indexType: IndexType): Option[String] = + Option(indexType).filter(scalarSegmentIndexTypes.contains).flatMap(methodByIndexType.get) + + private val SystemIndexNames: Set[String] = Set("__lance_frag_reuse", "__lance_mem_wal") + + /** True for indexes Lance maintains itself, which user commands must not target. */ + def isSystemIndex(indexName: String): Boolean = + indexName != null && SystemIndexNames.exists(_.equalsIgnoreCase(indexName)) + + /** Parses and range-checks the `num_segments` option shared by the segmented build paths. */ + def parseNumSegments(arg: LanceNamedArgument): Int = arg.value match { + case null => + throw new IllegalArgumentException("num_segments must be a positive integer, got: null") + case n: Number => + val asLong = n.longValue() + if (asLong < 1L || asLong > Int.MaxValue) { + throw new IllegalArgumentException( + s"num_segments must be a positive integer that fits in Int, got: $asLong") + } + asLong.toInt + case other => + throw new IllegalArgumentException(s"num_segments must be a positive integer, got: $other") + } + /** Pins `readOptions` to the version `dataset` is open at. */ def pinVersion( readOptions: LanceSparkReadOptions, @@ -732,6 +754,106 @@ object IndexUtils extends Logging { if (ordered.size > 10) s"$shown, ... (${ordered.size} total)" else shown } + // Index types whose segments must all describe the same build configuration. Lance builds each + // segment independently and queries most of them independently too, so for those, segments built + // with different options differ in performance only. An inverted (FTS) index is the exception: its + // read path loads one set of index details for the whole logical index and rejects a set whose + // segments disagree, which fails every full-text query on the column. Kept explicit rather than + // inferred; extend it if core grows another type with the same requirement. + private val uniformDetailsIndexTypes: Set[IndexType] = Set(IndexType.INVERTED) + + /** True when segments of this index type must all share one build configuration. */ + def requiresUniformSegmentDetails(indexType: IndexType): Boolean = + indexType != null && uniformDetailsIndexTypes.contains(indexType) + + /** + * Fails before commit when newly built segments describe a different build configuration than the + * segments they are about to join, for an index type that requires them to agree. + * + * Index details are the serialized build parameters, so this compares what was built rather than + * what was asked for, and it runs while the new segments are still uncommitted: a rejection leaves + * the index exactly as it was. Segments without index details predate the field, and an index + * being replaced wholesale has nothing to agree with; both cases defer to Lance core, which + * validates the segment set it is handed. + */ + def requireUniformSegmentDetails( + indexType: IndexType, + indexName: String, + method: String, + retainedSegments: Seq[Index], + builtSegments: Seq[Index]): Unit = { + if (!requiresUniformSegmentDetails(indexType)) { + return + } + val retained = distinctIndexDetails(retainedSegments) + val built = distinctIndexDetails(builtSegments) + if (retained.isEmpty || built.isEmpty) { + return + } + if (retained.size > 1 || built.size > 1 || retained != built) { + throw new IllegalArgumentException( + s"Index '$indexName' uses the $method method, whose segments must all share one " + + "configuration, and the segments this build produced are configured differently from the " + + "ones they would join. Nothing was committed and the index is unchanged. Re-run with the " + + "options the index was created with, or rebuild it in full with " + + "ALTER TABLE ... CREATE INDEX.") + } + } + + private def distinctIndexDetails(segments: Seq[Index]): Set[Seq[Byte]] = + segments.iterator + .flatMap(segment => Option(segment.indexDetails().orElse(null))) + .map(_.toSeq) + .toSet + + /** + * Segments of `indexName` that a segment commit will keep, resolved against `dataset` as it is now. + * + * Fails when the index is gone. A refresh resolves its target before the distributed build and + * commits after it, and `commitExistingIndexSegments` against a name Lance no longer knows creates + * that index rather than extending it: a DROP INDEX during the build would otherwise be undone, + * leaving the index back in place with only the coverage this build happened to plan for. + */ + def resolveRetainedSegments( + dataset: Dataset, + indexName: String, + liveFragmentIds: Set[Int]): Seq[Index] = + retainedSegments(dataset.getIndexes.asScala.toSeq, indexName, liveFragmentIds) + + /** Dataset-free form of [[resolveRetainedSegments]]. */ + def retainedSegments( + allSegments: Seq[Index], + indexName: String, + liveFragmentIds: Set[Int]): Seq[Index] = { + val current = allSegments.filter(segment => indexName == segment.name()) + if (current.isEmpty) { + throw new IllegalStateException( + s"Index '$indexName' no longer exists: it was dropped or replaced while the build was " + + "running. Nothing was committed; re-create it with ALTER TABLE ... CREATE INDEX.") + } + current.filter(segment => declaredCoverage(Seq(segment)).exists(liveFragmentIds.contains)) + } + + /** Namespace and storage context that tasks need to reopen the dataset on an executor. */ + def extractNamespaceInfo( + catalog: TableCatalog, + 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) + } + } + def resolveIndexField( schema: LanceSchema, indexType: IndexType, @@ -747,7 +869,7 @@ object IndexUtils extends Logging { * Extracts the `train` option from named arguments, defaulting to `true`. * * When `train=false`, index creation registers an empty index without processing any data. - * All existing rows will be unindexed and covered by a subsequent OPTIMIZE INDEX call. + * All existing rows are left unindexed until a subsequent REFRESH INDEX covers them. */ def extractTrain(args: Seq[LanceNamedArgument]): Boolean = args.find(_.name == "train") match { @@ -848,14 +970,16 @@ object IndexUtils extends Logging { /** * Splits `fragments` into `numSegments` batches, each a contiguous run of fragment ids, chosen so - * that the heaviest batch is as light as any contiguous split allows. + * that the heaviest batch is as light as possible. * * Contiguity is not cosmetic. Lance's compaction planner only groups fragments that are covered by * the identical set of index segments, so batches whose fragment ids interleave leave every * adjacent pair of fragments in a different group and make OPTIMIZE a no-op for the whole table. - * It does cost some balance. `[10, 9, 8, 7]` into two batches is 19/15 here, where the previous - * least-loaded-first assignment reached 17/17 by interleaving. Optimal among contiguous splits is - * the guarantee, not optimal overall. + * Since an index accumulates one segment set per build, and REFRESH INDEX adds more over time, + * interleaved coverage would permanently block compaction on exactly the append-heavy tables this + * command exists for. Balance is not sacrificed to get it: the optimal contiguous partition is + * found exactly, so a workload the previous least-loaded-first assignment balanced perfectly + * still is. * * Assignment is deterministic: the same fragments and segment count always produce the same * batches, whatever order `fragments` arrives in. Every batch holds at least one fragment, so the @@ -907,9 +1031,8 @@ object IndexUtils extends Logging { * is the widest single fragment, below which no packing exists. * * Packing at that budget can use fewer runs than were asked for, which would cost parallelism, so - * runs are divided until the count is reached. Which run gets divided does not matter: every run - * already fits the budget, so both halves of any split fit it too, and the budget is minimal, so - * the heaviest run cannot fall below it either. + * the remainder are split at their own balance points. A split only ever lowers the heaviest run, + * so optimality survives it. */ private def balancedRunLengths(rowsUpTo: Array[Long], segmentCount: Int): Seq[Int] = { val fragmentCount = rowsUpTo.length - 1 @@ -943,18 +1066,33 @@ object IndexUtils extends Logging { start += length } - // Divide from the left until the count is reached. Splitting the heaviest run first would be no - // better, since the budget already bounds every run. - var splitAt = 0 - while (runCount < segmentCount && splitAt < fragmentCount) { - val length = runLengthAt(splitAt) - if (length > 1) { - val cut = balancePoint(rowsUpTo, splitAt, length) - runLengthAt(splitAt) = cut - runLengthAt(splitAt + cut) = length - cut + if (runCount < segmentCount) { + // Heaviest splittable run first, so each extra batch is spent where it helps most. Ties break + // on length then on the earlier run, to keep the result independent of heap internals. + val splittable = PriorityQueue.empty[(Long, Int, Int)]( + Ordering.by[(Long, Int, Int), (Long, Int, Int)] { + case (rows, length, begin) => (rows, length, -begin) + }) + var scan = 0 + while (scan < fragmentCount) { + val length = runLengthAt(scan) + if (length > 1) { + splittable.enqueue((rowsOf(scan, length), length, scan)) + } + scan += length + } + while (runCount < segmentCount && splittable.nonEmpty) { + val (_, length, begin) = splittable.dequeue() + val cut = balancePoint(rowsUpTo, begin, length) + runLengthAt(begin) = cut + runLengthAt(begin + cut) = length - cut runCount += 1 - } else { - splitAt += length + if (cut > 1) { + splittable.enqueue((rowsOf(begin, cut), cut, begin)) + } + if (length - cut > 1) { + splittable.enqueue((rowsOf(begin + cut, length - cut), length - cut, begin + cut)) + } } } diff --git a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowIndexesExec.scala b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowIndexesExec.scala index 3015b584e..2b19c4288 100755 --- a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowIndexesExec.scala +++ b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowIndexesExec.scala @@ -24,12 +24,7 @@ import org.lance.spark.utils.{FieldPathUtils, Utils} import scala.collection.JavaConverters._ -object ShowIndexesExec { - private val SystemIndexNames = Set("__lance_frag_reuse", "__lance_mem_wal") - - private def isSystemIndex(indexName: String): Boolean = - indexName != null && SystemIndexNames.exists(_.equalsIgnoreCase(indexName)) -} +object ShowIndexesExec /** * Physical execution of SHOW INDEXES for Lance datasets. @@ -54,7 +49,7 @@ case class ShowIndexesExec( val dataset = Utils.openDatasetBuilder(readOptions).build() try { val indexes = dataset.getIndexes.asScala.toSeq - .filterNot(idx => ShowIndexesExec.isSystemIndex(idx.name())) + .filterNot(idx => IndexUtils.isSystemIndex(idx.name())) .groupBy(_.name()) .toSeq .sortBy(_._1) diff --git a/lance-spark-base_2.12/src/test/scala/org/apache/spark/sql/execution/datasources/v2/IndexUtilsTest.scala b/lance-spark-base_2.12/src/test/scala/org/apache/spark/sql/execution/datasources/v2/IndexUtilsTest.scala index a09aba6ca..11df42edd 100644 --- a/lance-spark-base_2.12/src/test/scala/org/apache/spark/sql/execution/datasources/v2/IndexUtilsTest.scala +++ b/lance-spark-base_2.12/src/test/scala/org/apache/spark/sql/execution/datasources/v2/IndexUtilsTest.scala @@ -539,4 +539,224 @@ class IndexUtilsTest { "idx_id")) } + // ── methodForIndexType ──────────────────────────────────────────────────── + + @Test + def methodForIndexType_roundTripsEverySegmentBuildableType(): Unit = { + val types = Seq( + IndexType.BTREE, + IndexType.ZONEMAP, + IndexType.BITMAP, + IndexType.LABEL_LIST, + IndexType.NGRAM, + IndexType.BLOOM_FILTER, + IndexType.RTREE, + IndexType.INVERTED) + + types.foreach { indexType => + val method = IndexUtils.methodForIndexType(indexType) + assertTrue(method.isDefined, s"expected a method name for $indexType") + assertEquals( + Some(indexType), + IndexUtils.scalarSegmentIndexType(method.get), + s"method '${method.get}' should resolve back to $indexType") + } + } + + @Test + def methodForIndexType_mapsInvertedToCanonicalFtsSpelling(): Unit = { + assertEquals(Some("fts"), IndexUtils.methodForIndexType(IndexType.INVERTED)) + } + + @Test + def methodForIndexType_returnsNoneForVectorType(): Unit = { + assertEquals(None, IndexUtils.methodForIndexType(IndexType.VECTOR)) + } + + @Test + def methodForIndexType_returnsNoneForNull(): Unit = { + assertEquals(None, IndexUtils.methodForIndexType(null)) + } + + // ── isSystemIndex ───────────────────────────────────────────────────────── + + @Test + def isSystemIndex_matchesLanceMaintainedIndexesCaseInsensitively(): Unit = { + assertTrue(IndexUtils.isSystemIndex("__lance_frag_reuse")) + assertTrue(IndexUtils.isSystemIndex("__LANCE_MEM_WAL")) + } + + @Test + def isSystemIndex_rejectsUserIndexesAndNull(): Unit = { + assertFalse(IndexUtils.isSystemIndex("idx_id")) + assertFalse(IndexUtils.isSystemIndex(null)) + } + + // ── retainedSegments ────────────────────────────────────────────────────── + + private def namedSegment(name: String, fragmentIds: Int*): Index = + Index + .builder() + .uuid(java.util.UUID.randomUUID()) + .name(name) + .indexType(IndexType.ZONEMAP) + .fragments(fragmentIds.map(java.lang.Integer.valueOf).asJava) + .build() + + @Test + def retainedSegments_keepsOnlySegmentsOfTheNamedIndexThatStillCoverLiveFragments(): Unit = { + val kept = namedSegment("idx_id", 0, 1) + val allRetired = namedSegment("idx_id", 7) + val otherIndex = namedSegment("idx_other", 0) + + assertEquals( + Seq(kept), + IndexUtils.retainedSegments(Seq(kept, allRetired, otherIndex), "idx_id", Set(0, 1))) + } + + @Test + def retainedSegments_failsWhenTheIndexIsGone(): Unit = { + val error = assertThrows( + classOf[IllegalStateException], + () => IndexUtils.retainedSegments(Seq(namedSegment("idx_other", 0)), "idx_id", Set(0))) + + assertTrue(error.getMessage.contains("idx_id"), error.getMessage) + assertTrue(error.getMessage.contains("no longer exists"), error.getMessage) + assertTrue(error.getMessage.contains("CREATE INDEX"), error.getMessage) + } + + @Test + def retainedSegments_matchesIndexNameExactly(): Unit = { + assertThrows( + classOf[IllegalStateException], + () => IndexUtils.retainedSegments(Seq(namedSegment("IDX_ID", 0)), "idx_id", Set(0))) + } + + // ── requireUniformSegmentDetails ────────────────────────────────────────── + + @Test + def requiresUniformSegmentDetails_onlyForInverted(): Unit = { + assertTrue(IndexUtils.requiresUniformSegmentDetails(IndexType.INVERTED)) + Seq( + IndexType.BTREE, + IndexType.ZONEMAP, + IndexType.BITMAP, + IndexType.LABEL_LIST, + IndexType.NGRAM, + IndexType.BLOOM_FILTER, + IndexType.RTREE, + IndexType.VECTOR).foreach { indexType => + assertFalse(IndexUtils.requiresUniformSegmentDetails(indexType), indexType.name()) + } + assertFalse(IndexUtils.requiresUniformSegmentDetails(null)) + } + + @Test + def requireUniformSegmentDetails_acceptsMatchingDetails(): Unit = { + val details = Array[Byte](1, 2, 3) + IndexUtils.requireUniformSegmentDetails( + IndexType.INVERTED, + "idx_id", + "fts", + Seq(segment(Some(Seq(0)), Some(details.clone()))), + Seq(segment(Some(Seq(1)), Some(details.clone())))) + } + + @Test + def requireUniformSegmentDetails_rejectsDifferingDetails(): Unit = { + val error = assertThrows( + classOf[IllegalArgumentException], + () => + IndexUtils.requireUniformSegmentDetails( + IndexType.INVERTED, + "idx_id", + "fts", + Seq(segment(Some(Seq(0)), Some(Array[Byte](1, 2, 3)))), + Seq(segment(Some(Seq(1)), Some(Array[Byte](1, 2, 4)))))) + + assertTrue(error.getMessage.contains("idx_id"), error.getMessage) + assertTrue(error.getMessage.contains("fts"), error.getMessage) + assertTrue(error.getMessage.contains("CREATE INDEX"), error.getMessage) + } + + @Test + def requireUniformSegmentDetails_rejectsDisagreementWithinEitherSide(): Unit = { + val a = Array[Byte](1) + val b = Array[Byte](2) + assertThrows( + classOf[IllegalArgumentException], + () => + IndexUtils.requireUniformSegmentDetails( + IndexType.INVERTED, + "idx_id", + "fts", + Seq(segment(Some(Seq(0)), Some(a)), segment(Some(Seq(1)), Some(b))), + Seq(segment(Some(Seq(2)), Some(a))))) + } + + @Test + def requireUniformSegmentDetails_ignoresIndexTypesWithoutTheRequirement(): Unit = { + IndexUtils.requireUniformSegmentDetails( + IndexType.ZONEMAP, + "idx_id", + "zonemap", + Seq(segment(Some(Seq(0)), Some(Array[Byte](1)))), + Seq(segment(Some(Seq(1)), Some(Array[Byte](2))))) + } + + @Test + def requireUniformSegmentDetails_defersWhenEitherSideHasNoDetails(): Unit = { + IndexUtils.requireUniformSegmentDetails( + IndexType.INVERTED, + "idx_id", + "fts", + Seq(segment(Some(Seq(0)))), + Seq(segment(Some(Seq(1)), Some(Array[Byte](1))))) + IndexUtils.requireUniformSegmentDetails( + IndexType.INVERTED, + "idx_id", + "fts", + Seq.empty, + Seq(segment(Some(Seq(1)), Some(Array[Byte](1))))) + } + + // ── parseNumSegments ────────────────────────────────────────────────────── + + @Test + def parseNumSegments_acceptsPositiveIntegers(): Unit = { + assertEquals( + 8, + IndexUtils.parseNumSegments(LanceNamedArgument("num_segments", java.lang.Long.valueOf(8)))) + } + + @Test + def parseNumSegments_rejectsNonPositiveValues(): Unit = { + Seq(0L, -1L).foreach { value => + assertThrows( + classOf[IllegalArgumentException], + () => + IndexUtils.parseNumSegments( + LanceNamedArgument("num_segments", java.lang.Long.valueOf(value)))) + } + } + + @Test + def parseNumSegments_rejectsValuesWiderThanInt(): Unit = { + assertThrows( + classOf[IllegalArgumentException], + () => + IndexUtils.parseNumSegments( + LanceNamedArgument("num_segments", java.lang.Long.valueOf(Int.MaxValue.toLong + 1)))) + } + + @Test + def parseNumSegments_rejectsNonNumericAndNullValues(): Unit = { + assertThrows( + classOf[IllegalArgumentException], + () => IndexUtils.parseNumSegments(LanceNamedArgument("num_segments", "eight"))) + assertThrows( + classOf[IllegalArgumentException], + () => IndexUtils.parseNumSegments(LanceNamedArgument("num_segments", null))) + } + } From 8931e7d86ef4164ccef19747216b5c2ac0dc3666 Mon Sep 17 00:00:00 2001 From: ivscheianu Date: Wed, 16 Sep 2026 11:27:34 +0300 Subject: [PATCH 2/3] ci: re-trigger checks (flaky Lance index-build OOM) From b03a94cf786b5855b9c42f00bce69a72a203880f Mon Sep 17 00:00:00 2001 From: ivscheianu Date: Wed, 16 Sep 2026 14:35:12 +0300 Subject: [PATCH 3/3] refactor: trim verbose comments, add missing test, remove dead code Trim ~40 lines of prose comments down to constraint-only notes, add a test for retainedSegments returning empty when all segments cover only retired fragments, and remove the empty ShowIndexesExec companion object. --- .../datasources/v2/AddIndexExec.scala | 71 ++++--------------- .../datasources/v2/ShowIndexesExec.scala | 2 - .../datasources/v2/IndexUtilsTest.scala | 9 +++ 3 files changed, 23 insertions(+), 59 deletions(-) diff --git a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AddIndexExec.scala b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AddIndexExec.scala index dc3bfa1eb..ea9738f33 100755 --- a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AddIndexExec.scala +++ b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AddIndexExec.scala @@ -523,17 +523,9 @@ class ScalarSegmentIndexJob( final private[v2] case class FragmentWorkload(fragmentId: Integer, numRows: Long) -/** - * A task to create a scalar index segment on a batch of fragments. - * - * The segment is named after the logical index it will join, and {@code replace} is set because that - * index may already exist: on the uncommitted build path Lance consults {@code replace} only to - * reject a name that is already taken, which is precisely the normal case for a REFRESH, and for a - * CREATE INDEX that replaces an index of the same name. Nothing is removed here — the driver's - * single {@code commitExistingIndexSegments} transaction decides which existing segments to keep. - * Leaving the name unset instead makes Lance derive its own default (`{column}_idx`) and reject the - * build whenever an index of that name exists on the column. - */ +// Named after the index it joins with replace=true: on the uncommitted path Lance uses replace only +// to skip its name collision check. Without it, Lance derives `{column}_idx` and rejects the build +// when that name is taken. case class ScalarSegmentIndexTask( encodedReadOptions: String, indexName: String, @@ -629,8 +621,7 @@ object IndexUtils extends Logging { IndexType.RTREE -> "rtree", IndexType.INVERTED -> "inverted") - // Reverse of `methodToIndexTypes`, for commands that resolve an already-created index and need - // the method name back. INVERTED maps to the canonical "fts" spelling rather than its alias. + // INVERTED maps to the canonical "fts" spelling. private val methodByIndexType: Map[IndexType, String] = Map( IndexType.BTREE -> "btree", IndexType.ZONEMAP -> "zonemap", @@ -754,28 +745,15 @@ object IndexUtils extends Logging { if (ordered.size > 10) s"$shown, ... (${ordered.size} total)" else shown } - // Index types whose segments must all describe the same build configuration. Lance builds each - // segment independently and queries most of them independently too, so for those, segments built - // with different options differ in performance only. An inverted (FTS) index is the exception: its - // read path loads one set of index details for the whole logical index and rejects a set whose - // segments disagree, which fails every full-text query on the column. Kept explicit rather than - // inferred; extend it if core grows another type with the same requirement. + // FTS loads one config for all segments and rejects disagreement; other types query segments + // independently, so mismatched options only affect performance. private val uniformDetailsIndexTypes: Set[IndexType] = Set(IndexType.INVERTED) /** True when segments of this index type must all share one build configuration. */ def requiresUniformSegmentDetails(indexType: IndexType): Boolean = indexType != null && uniformDetailsIndexTypes.contains(indexType) - /** - * Fails before commit when newly built segments describe a different build configuration than the - * segments they are about to join, for an index type that requires them to agree. - * - * Index details are the serialized build parameters, so this compares what was built rather than - * what was asked for, and it runs while the new segments are still uncommitted: a rejection leaves - * the index exactly as it was. Segments without index details predate the field, and an index - * being replaced wholesale has nothing to agree with; both cases defer to Lance core, which - * validates the segment set it is handed. - */ + /** Pre-commit guard: rejects built segments whose configuration differs from retained ones. */ def requireUniformSegmentDetails( indexType: IndexType, indexName: String, @@ -807,12 +785,8 @@ object IndexUtils extends Logging { .toSet /** - * Segments of `indexName` that a segment commit will keep, resolved against `dataset` as it is now. - * - * Fails when the index is gone. A refresh resolves its target before the distributed build and - * commits after it, and `commitExistingIndexSegments` against a name Lance no longer knows creates - * that index rather than extending it: a DROP INDEX during the build would otherwise be undone, - * leaving the index back in place with only the coverage this build happened to plan for. + * Segments of `indexName` that the commit will keep. Fails if the index was dropped: committing + * against an unknown name creates a new index rather than extending the existing one. */ def resolveRetainedSegments( dataset: Dataset, @@ -972,18 +946,8 @@ object IndexUtils extends Logging { * Splits `fragments` into `numSegments` batches, each a contiguous run of fragment ids, chosen so * that the heaviest batch is as light as possible. * - * Contiguity is not cosmetic. Lance's compaction planner only groups fragments that are covered by - * the identical set of index segments, so batches whose fragment ids interleave leave every - * adjacent pair of fragments in a different group and make OPTIMIZE a no-op for the whole table. - * Since an index accumulates one segment set per build, and REFRESH INDEX adds more over time, - * interleaved coverage would permanently block compaction on exactly the append-heavy tables this - * command exists for. Balance is not sacrificed to get it: the optimal contiguous partition is - * found exactly, so a workload the previous least-loaded-first assignment balanced perfectly - * still is. - * - * Assignment is deterministic: the same fragments and segment count always produce the same - * batches, whatever order `fragments` arrives in. Every batch holds at least one fragment, so the - * result always has exactly `segmentCount` entries. + * Contiguity matters: Lance's compaction planner only groups fragments covered by the identical + * set of index segments, so interleaved batches make OPTIMIZE a no-op for the whole table. */ def batchFragments( fragments: List[FragmentWorkload], @@ -1025,14 +989,8 @@ object IndexUtils extends Logging { /** * Lengths of exactly `segmentCount` contiguous runs over `rowsUpTo`, minimising the heaviest run. * - * The smallest row budget a contiguous packing can respect is found by binary search, which is - * exact rather than approximate: for a fixed budget, extending each run as far as it will go uses - * the fewest runs, so the smallest feasible budget is the optimal maximum. The floor of the search - * is the widest single fragment, below which no packing exists. - * - * Packing at that budget can use fewer runs than were asked for, which would cost parallelism, so - * the remainder are split at their own balance points. A split only ever lowers the heaviest run, - * so optimality survives it. + * Binary-searches the smallest budget a contiguous packing can respect, then splits the heaviest + * runs until the count is reached. */ private def balancedRunLengths(rowsUpTo: Array[Long], segmentCount: Int): Seq[Int] = { val fragmentCount = rowsUpTo.length - 1 @@ -1067,8 +1025,7 @@ object IndexUtils extends Logging { } if (runCount < segmentCount) { - // Heaviest splittable run first, so each extra batch is spent where it helps most. Ties break - // on length then on the earlier run, to keep the result independent of heap internals. + // Heaviest splittable run first. Ties break on length then position for determinism. val splittable = PriorityQueue.empty[(Long, Int, Int)]( Ordering.by[(Long, Int, Int), (Long, Int, Int)] { case (rows, length, begin) => (rows, length, -begin) diff --git a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowIndexesExec.scala b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowIndexesExec.scala index 2b19c4288..d8aef2430 100755 --- a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowIndexesExec.scala +++ b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowIndexesExec.scala @@ -24,8 +24,6 @@ import org.lance.spark.utils.{FieldPathUtils, Utils} import scala.collection.JavaConverters._ -object ShowIndexesExec - /** * Physical execution of SHOW INDEXES for Lance datasets. * diff --git a/lance-spark-base_2.12/src/test/scala/org/apache/spark/sql/execution/datasources/v2/IndexUtilsTest.scala b/lance-spark-base_2.12/src/test/scala/org/apache/spark/sql/execution/datasources/v2/IndexUtilsTest.scala index 11df42edd..625b6540b 100644 --- a/lance-spark-base_2.12/src/test/scala/org/apache/spark/sql/execution/datasources/v2/IndexUtilsTest.scala +++ b/lance-spark-base_2.12/src/test/scala/org/apache/spark/sql/execution/datasources/v2/IndexUtilsTest.scala @@ -625,6 +625,15 @@ class IndexUtilsTest { assertTrue(error.getMessage.contains("CREATE INDEX"), error.getMessage) } + @Test + def retainedSegments_returnsEmptyWhenAllSegmentsCoverOnlyRetiredFragments(): Unit = { + val result = IndexUtils.retainedSegments( + Seq(namedSegment("idx_id", 5, 6)), + "idx_id", + Set(0, 1)) + assertEquals(Seq.empty, result) + } + @Test def retainedSegments_matchesIndexNameExactly(): Unit = { assertThrows(