From 40d6d8a2122e5d7e2f57a64f80592d03a064bfc9 Mon Sep 17 00:00:00 2001 From: ivscheianu Date: Wed, 26 Aug 2026 17:09:21 +0300 Subject: [PATCH 1/5] fix: make distributed index builds version-consistent and report real coverage Tasks open the dataset themselves. Nothing pinned the version they opened at, so each one resolved the latest independently and could see a different fragment set than the driver batched, or than its sibling tasks. The segmented build path now pins the version the driver planned over and hands that to every task. Range-mode BTree deliberately stays unpinned. It reads the table back through the catalog, which resolves and pins its own version, so pinning the executor open to the planning version would make the segment record a dataset version older than the rows it holds -- which core's staleness pruning acts on. CREATE INDEX also reported the fragment count it planned rather than the one it achieved. Commit intersects each segment's declared coverage with the dataset's live fragments, so a fragment retired while the build ran contributes nothing, and the command still counted it as indexed. The count now comes from the metadata the commit returns, intersected with the fragments live once it has landed. A check taken before the commit cannot answer this: it reads the manifest its handle was opened at while the commit lands on whatever version is current, and Lance prunes an incoming segment's coverage not only for a fragment that is gone but also for one whose indexed field was rewritten under the same id, which no comparison of fragment ids can see. Only the segments this build produced are counted, since existing segments disjoint from them survive the commit and their coverage is not this command's to report. A segment set that would establish no coverage at all is still refused before the commit, which is the last point at which the build can be declined. Note this is not because coverage would be lost: Lance accepts an empty fragment bitmap, and an existing segment is trivially disjoint from one, so it survives. The transaction would simply publish segments that index nothing. Partial loss is not refused. A fragment leaves the manifest either because its rows moved, through compaction or an in-place rewrite, or because they were all deleted; only the first leaves data unindexed, and either way the segments covering what remains are correct. Discarding a finished distributed build would be the wrong response, and it would make a routine concurrent DELETE fatal on a long one. The shortfall is named in a warning instead. Fragment enumeration goes through getFragmentStatistics(), which returns primitive arrays, rather than getFragments(), which materializes a Java object per fragment and per data file. The driver enumerates twice per command, so that difference is the bulk of planning cost on a large table. --- .../datasources/v2/AddIndexExec.scala | 162 ++++++++++++++++-- .../lance/spark/update/BaseAddIndexTest.java | 72 ++++++++ .../datasources/v2/IndexUtilsTest.scala | 148 +++++++++++++++- 3 files changed, 368 insertions(+), 14 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 085090996..34f075752 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 @@ -88,7 +88,11 @@ case class AddIndexExec( val btreeBuildMode = IndexUtils.btreeBuildMode(indexType, args) val scalarSegmentIndexType = IndexUtils.scalarSegmentIndexType(method) - val (fragmentWorkloads, canonicalColumns) = { + // Plan and build against a single pinned version. Tasks open the dataset themselves, so without + // pinning each one resolves the latest version independently and may see a different fragment + // set than the driver batched, or than its sibling tasks. Coverage the commit cannot establish is + // then accounted for at commit time; see IndexUtils.committedCoverage. + val (fragmentWorkloads, canonicalColumns, buildReadOptions) = { val ds = Utils.openDatasetBuilder(readOptions).build() try { val canonical = columns.map { column => @@ -96,10 +100,9 @@ case class AddIndexExec( FieldPathUtils.pathByFieldId(ds.getLanceSchema, field.getId) } ( - ds.getFragments.asScala - .map(fragment => FragmentWorkload(fragment.getId, fragment.metadata().getNumRows)) - .toList, - canonical) + IndexUtils.fragmentWorkloads(ds), + canonical, + IndexUtils.pinVersion(readOptions, ds)) } finally { ds.close() } @@ -163,7 +166,10 @@ case class AddIndexExec( val (nsImpl, nsProps, tableId, initialStorageOpts) = extractNamespaceInfo(lanceDataset, readOptions) - // Range-mode BTree uses preprocessed data from Spark and keeps its dedicated path. + // Range-mode BTree uses preprocessed data from Spark and keeps its dedicated path. It reads the + // table back through the catalog, so its coverage follows the scan rather than the fragment list + // planned above; pinning the build to the planning version would record a segment version older + // than the data the segment holds, which core's staleness pruning would act on. if (btreeBuildMode.contains("range")) { val segments = new RangeBasedBTreeIndexJob( this.copy(columns = canonicalColumns), @@ -173,9 +179,9 @@ case class AddIndexExec( nsProps, tableId, initialStorageOpts).run() - commitIndexSegments(readOptions, canonicalColumns.head, segments) + val indexed = commitIndexSegments(readOptions, canonicalColumns.head, segments) return Seq(new GenericInternalRow(Array[Any]( - fragmentIds.size.toLong, + indexed.toLong, UTF8String.fromString(indexName)))) } @@ -183,7 +189,7 @@ case class AddIndexExec( if (scalarSegmentIndexType.isDefined) { val segmentJob = new ScalarSegmentIndexJob( this.copy(columns = canonicalColumns), - readOptions, + buildReadOptions, fragmentWorkloads, validatedNumSegments, nsImpl, @@ -192,9 +198,9 @@ case class AddIndexExec( initialStorageOpts) val segments = segmentJob.run() // Atomic add+remove via Lance core; see commitIndexSegments - commitIndexSegments(readOptions, canonicalColumns.head, segments) + val indexed = commitIndexSegments(readOptions, canonicalColumns.head, segments) return Seq(new GenericInternalRow(Array[Any]( - fragmentIds.size.toLong, + indexed.toLong, UTF8String.fromString(indexName)))) } @@ -231,16 +237,31 @@ case class AddIndexExec( // 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. + // + // Returns the number of fragments the commit actually covers, which is what the command reports. private def commitIndexSegments( readOptions: LanceSparkReadOptions, column: String, - segments: Seq[Index]): Unit = { + segments: Seq[Index]): Int = { val dataset = Utils.openDatasetBuilder(readOptions).build() try { - dataset.commitExistingIndexSegments( + IndexUtils.requireCommittableCoverage( + IndexUtils.liveFragmentIds(dataset), + segments, + indexName) + val committed = dataset.commitExistingIndexSegments( indexName, column, segments.toList.asJava) + // The commit advances this handle to the manifest it wrote, so both the returned metadata and + // the fragment list below describe the committed state rather than the one validated above. + IndexUtils + .establishedCoverage( + segments, + committed.asScala.toSeq, + IndexUtils.liveFragmentIds(dataset), + indexName) + .size } finally { dataset.close() } @@ -276,6 +297,12 @@ case class AddIndexExec( * covering exactly those fragments, so the resulting segments have disjoint fragment coverage and * can be committed directly as a single logical index. * + * Unlike the segmented path, coverage here is derived from the fragment ids present in the scanned + * rows rather than from a fragment list fixed at planning time. The scan goes back through the + * catalog, which resolves and pins its own version, so this job is deliberately handed unpinned read + * options: the version an executor opens must not predate the rows it is indexing, or the segment + * would record a dataset version older than its own contents. + * * @param addIndexExec The AddIndexExec instance that initiated this job * @param readOptions Configuration options for reading the Lance dataset * @param numFragments Number of fragments in the dataset, used to bound shuffle partitions @@ -619,6 +646,115 @@ object IndexUtils extends Logging { .get(method.toLowerCase(Locale.ROOT)) .filter(scalarSegmentIndexTypes.contains) + /** + * Pins `readOptions` to the version `dataset` is open at. + * + * Distributed index builds hand read options to tasks that open the dataset themselves. Pinning + * makes every task observe the fragment set the driver planned over instead of resolving the + * latest version independently. + */ + def pinVersion( + readOptions: LanceSparkReadOptions, + dataset: Dataset): LanceSparkReadOptions = + readOptions.withRef(Utils.pinOpenedRef(dataset, readOptions.getRef)) + + /** + * Fragment ids and live row counts of `dataset`, in manifest order. + * + * Reads the primitive fragment-statistics view rather than [[Dataset#getFragments]], which + * materializes a Java object per fragment and per data file. Both commands enumerate fragments on + * the driver, once to plan and once to check the commit, so on a large table that difference is + * the bulk of planning cost. + */ + def fragmentWorkloads(dataset: Dataset): List[FragmentWorkload] = { + val stats = dataset.getFragmentStatistics + val ids = stats.getIds + val rowCounts = stats.getRowCounts + List.tabulate(ids.length)(index => + FragmentWorkload(Integer.valueOf(ids(index)), rowCounts(index))) + } + + /** Fragment ids live in `dataset`. See [[fragmentWorkloads]] for why this avoids getFragments. */ + def liveFragmentIds(dataset: Dataset): Set[Int] = + dataset.getFragmentStatistics.getIds.toSet + + /** Fragment ids the given segments declare coverage of. */ + def declaredCoverage(segments: Seq[Index]): Set[Int] = + segments.iterator + .flatMap(_.fragments().orElse(Collections.emptyList[Integer]()).asScala) + .map(_.intValue) + .toSet + + /** + * Refuses to publish a segment set that would establish no coverage at all. + * + * Commit intersects each segment's declared coverage with the dataset's live fragments, so a + * fragment retired while the build ran contributes nothing. Lance accepts such a set: an empty + * fragment bitmap is valid metadata, and an existing segment is trivially disjoint from it and so + * survives. Nothing is corrupted, but the transaction would publish segments that index no data and + * report a build that achieved nothing, and this is the last point at which it can still be + * declined. + * + * Partial loss is not refused. A fragment leaves the manifest either because its rows moved + * (compaction, an in-place rewrite) or because they were all deleted; only the first leaves data + * unindexed, and either way the segments covering what remains are correct. Discarding a finished + * distributed build over that would be the wrong response, and it would make a routine concurrent + * DELETE fatal on a long one. What the commit actually established is reported afterwards, by + * [[establishedCoverage]]. + */ + def requireCommittableCoverage( + liveFragmentIds: Set[Int], + segments: Seq[Index], + indexName: String): Unit = { + val declared = declaredCoverage(segments) + if (declared.nonEmpty && declared.intersect(liveFragmentIds).isEmpty) { + throw new IllegalStateException( + s"Index '$indexName' build raced a concurrent operation: every fragment it covers " + + s"(${describeFragmentIds(declared)}) was retired while the build ran, so the segments " + + "would cover nothing. No index change was committed; re-run the command.") + } + } + + /** + * The coverage a commit established, read back from the metadata the commit returned. + * + * A check taken before the commit cannot answer this, for two reasons. It reads the manifest its + * handle was opened at, while the commit lands on whatever version is current by then. And Lance + * prunes an incoming segment's coverage not only for a fragment that is gone but also for one whose + * indexed field was rewritten under the same id, which no comparison of fragment ids can see. The + * returned metadata is post-pruning, so it is the only truthful account of what was indexed. + * + * Only the segments this build produced are counted. Existing segments that were disjoint from them + * survive the commit, and their coverage is not this command's to report. + * + * @return the fragment ids the commit covered with these segments + */ + def establishedCoverage( + builtSegments: Seq[Index], + committedSegments: Seq[Index], + liveFragmentIds: Set[Int], + indexName: String): Set[Int] = { + val builtUuids = builtSegments.map(_.uuid).toSet + val established = + declaredCoverage(committedSegments.filter(segment => builtUuids.contains(segment.uuid))) + .intersect(liveFragmentIds) + val uncovered = declaredCoverage(builtSegments).diff(established) + if (uncovered.nonEmpty) { + logWarning( + s"Index '$indexName' build raced a concurrent operation: fragments " + + s"${describeFragmentIds(uncovered)} are not covered by this commit, because they were " + + "retired or because their indexed field was rewritten while the build ran. The segments " + + "for the remaining fragments are committed; re-run the command to cover them.") + } + established + } + + private def describeFragmentIds(ids: Set[Int]): String = { + val ordered = ids.toSeq.sorted + val shown = ordered.take(10).mkString(", ") + if (ordered.size > 10) s"$shown, ... (${ordered.size} total)" else shown + } + def resolveIndexField( schema: LanceSchema, indexType: IndexType, diff --git a/lance-spark-base_2.12/src/test/java/org/lance/spark/update/BaseAddIndexTest.java b/lance-spark-base_2.12/src/test/java/org/lance/spark/update/BaseAddIndexTest.java index 711292373..b51f55986 100755 --- a/lance-spark-base_2.12/src/test/java/org/lance/spark/update/BaseAddIndexTest.java +++ b/lance-spark-base_2.12/src/test/java/org/lance/spark/update/BaseAddIndexTest.java @@ -17,8 +17,11 @@ import org.lance.index.Index; import org.lance.index.IndexCriteria; import org.lance.index.IndexDescription; +import org.lance.index.IndexOptions; +import org.lance.index.IndexParams; import org.lance.index.IndexType; import org.lance.index.OptimizeOptions; +import org.lance.index.scalar.ScalarIndexParams; import org.lance.ipc.FullTextQuery; import org.lance.ipc.LanceScanner; import org.lance.ipc.ScanOptions; @@ -105,6 +108,75 @@ public void tearDown() throws IOException { } } + /** + * Pins the two Lance behaviours the coverage report rests on: a segment commit returns the + * metadata of the index as committed, and the handle it was made on advances to the manifest it + * wrote. + * + *

Together they are what make the reported count truthful. A check taken before the commit + * reads the manifest its handle was opened at, so a fragment retired in between is still counted; + * intersecting the returned metadata with the fragments live after the commit is what + * excludes it. This asserts the contract rather than the connector code consuming it, because a + * change on either side would silently make the count overstate again. + */ + @Test + public void testSegmentCommitReportsCoverageAsCommitted() { + spark.sql(String.format("create table %s (id int) using lance", fullTable)); + spark.sql(String.format("insert into %s values (0), (1), (2)", fullTable)); + spark.sql(String.format("insert into %s values (3), (4), (5)", fullTable)); + + try (org.lance.Dataset committer = + Utils.openDatasetBuilder(LanceSparkReadOptions.from(tableDir)).build()) { + List fragments = committer.getFragments(); + int coveredFragmentId = fragments.get(fragments.size() - 1).getId(); + + IndexParams indexParams = + IndexParams.builder() + .setScalarIndexParams(ScalarIndexParams.create("zonemap", "{}")) + .build(); + Index built = + committer.createIndex( + IndexOptions.builder(Collections.singletonList("id"), IndexType.ZONEMAP, indexParams) + .withIndexName("idx_committed_coverage") + .replace(true) + .withFragmentIds(Collections.singletonList(coveredFragmentId)) + .build()); + Assertions.assertEquals( + Collections.singletonList(coveredFragmentId), + built.fragments().orElse(Collections.emptyList()), + "the uncommitted segment should declare the fragment it was built for"); + + // Retire every fragment from another handle, leaving the committer on a stale manifest. + spark.sql(String.format("delete from %s where id >= 0", fullTable)); + + List committed = + committer.commitExistingIndexSegments( + "idx_committed_coverage", "id", Collections.singletonList(built)); + + Set ours = Collections.singleton(built.uuid()); + Assertions.assertTrue( + committed.stream().anyMatch(index -> ours.contains(index.uuid())), + "the commit must return the metadata of the segments it was handed"); + + Set liveAfter = + committer.getFragments().stream().map(Fragment::getId).collect(Collectors.toSet()); + Assertions.assertFalse( + liveAfter.contains(coveredFragmentId), + "the committing handle must advance to the manifest the commit wrote"); + + Set established = + committed.stream() + .filter(index -> ours.contains(index.uuid())) + .flatMap(index -> index.fragments().orElse(Collections.emptyList()).stream()) + .filter(liveAfter::contains) + .collect(Collectors.toSet()); + Assertions.assertEquals( + Collections.emptySet(), + established, + "coverage read from the committed state must not count a fragment retired in between"); + } + } + private void prepareDataset() { spark.sql(String.format("create table %s (id int, text string) using lance;", fullTable)); // First insert to create initial fragments 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 9644e6f1c..de55fc30a 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 @@ -16,7 +16,9 @@ package org.apache.spark.sql.execution.datasources.v2 import org.apache.spark.sql.catalyst.plans.logical.LanceNamedArgument import org.junit.jupiter.api.Assertions._ import org.junit.jupiter.api.Test -import org.lance.index.IndexType +import org.lance.index.{Index, IndexType} + +import scala.collection.JavaConverters._ /** * Unit tests for [[IndexUtils]] helper methods. @@ -26,6 +28,23 @@ import org.lance.index.IndexType */ class IndexUtilsTest { + /** An index segment carrying only the metadata these helpers read. */ + private def segment( + fragmentIds: Option[Seq[Int]], + indexDetails: Option[Array[Byte]] = None): Index = { + val builder = Index + .builder() + .uuid(java.util.UUID.randomUUID()) + .name("idx_id") + .indexType(IndexType.INVERTED) + fragmentIds.foreach(ids => + builder.fragments(ids.map(java.lang.Integer.valueOf).asJava)) + indexDetails.foreach(builder.indexDetails) + builder.build() + } + + private def coveringSegment(fragmentIds: Int*): Index = segment(Some(fragmentIds)) + private def fragmentWorkloads(rows: Long*): List[FragmentWorkload] = rows.zipWithIndex.map { case (rowCount, fragmentId) => FragmentWorkload(java.lang.Integer.valueOf(fragmentId), rowCount) @@ -257,4 +276,131 @@ class IndexUtilsTest { classOf[ArithmeticException], () => IndexUtils.batchFragments(fragmentWorkloads(Long.MaxValue, 1), Some(1), 1)) } + + // ── declaredCoverage / committedCoverage ─────────────────────── + + @Test + def declaredCoverage_unionsSegmentBitmaps(): Unit = { + assertEquals( + Set(0, 1, 4), + IndexUtils.declaredCoverage(Seq(coveringSegment(0, 1), coveringSegment(4)))) + } + + @Test + def declaredCoverage_treatsAbsentBitmapAsNoCoverage(): Unit = { + assertEquals(Set(2), IndexUtils.declaredCoverage(Seq(segment(None), coveringSegment(2)))) + assertEquals(Set.empty[Int], IndexUtils.declaredCoverage(Seq.empty)) + } + + /** A segment keeping its identity but reporting narrower coverage, as a pruned commit returns. */ + private def prunedTo(built: Index, fragmentIds: Seq[Int]): Index = + Index + .builder() + .uuid(built.uuid) + .name(built.name) + .indexType(built.indexType) + .fragments(fragmentIds.map(java.lang.Integer.valueOf).asJava) + .build() + + @Test + def requireCommittableCoverage_acceptsCoverageThatSurvives(): Unit = { + IndexUtils.requireCommittableCoverage(Set(1, 5), Seq(coveringSegment(0, 1)), "idx_id") + } + + @Test + def requireCommittableCoverage_acceptsSegmentsThatDeclareNothing(): Unit = { + IndexUtils.requireCommittableCoverage(Set(1), Seq(segment(None)), "idx_id") + IndexUtils.requireCommittableCoverage(Set.empty[Int], Seq.empty, "idx_id") + } + + @Test + def requireCommittableCoverage_refusesASetThatWouldCoverNothing(): Unit = { + val error = assertThrows( + classOf[IllegalStateException], + () => IndexUtils.requireCommittableCoverage(Set(5), Seq(coveringSegment(0, 7)), "idx_id")) + + assertTrue(error.getMessage.contains("idx_id"), error.getMessage) + assertTrue(error.getMessage.contains("0, 7"), error.getMessage) + assertTrue(error.getMessage.contains("re-run"), error.getMessage) + } + + @Test + def requireCommittableCoverage_summarizesLargeRetiredSets(): Unit = { + val error = assertThrows( + classOf[IllegalStateException], + () => + IndexUtils.requireCommittableCoverage( + Set.empty[Int], + Seq(coveringSegment(0 to 20: _*)), + "idx_id")) + + assertTrue(error.getMessage.contains("21 total"), error.getMessage) + } + + // ── establishedCoverage ─────────────────────────────────────────────────── + + @Test + def establishedCoverage_reportsWhatTheCommitReturned(): Unit = { + val built = Seq(coveringSegment(0, 1), coveringSegment(2)) + + assertEquals( + Set(0, 1, 2), + IndexUtils.establishedCoverage(built, built, Set(0, 1, 2), "idx_id")) + } + + /** + * Lance prunes a fragment whose indexed field was rewritten under the same id, so the committed + * bitmap can be narrower than the one handed in while every fragment is still live. No comparison + * of fragment ids can see that, which is why the report has to come from what the commit returned. + */ + @Test + def establishedCoverage_followsAPrunedCommitEvenWhileEveryFragmentIsLive(): Unit = { + val built = coveringSegment(0, 1) + + assertEquals( + Set(0), + IndexUtils.establishedCoverage( + Seq(built), + Seq(prunedTo(built, Seq(0))), + Set(0, 1), + "idx_id")) + } + + @Test + def establishedCoverage_excludesFragmentsRetiredByTheCommit(): Unit = { + val built = coveringSegment(0, 1) + + assertEquals( + Set(1), + IndexUtils.establishedCoverage(Seq(built), Seq(built), Set(1, 5), "idx_id")) + } + + /** Existing segments survive a commit they are disjoint from; their coverage is not ours. */ + @Test + def establishedCoverage_countsOnlyTheSegmentsThisBuildProduced(): Unit = { + val built = coveringSegment(2) + val survivor = coveringSegment(0, 1) + + assertEquals( + Set(2), + IndexUtils.establishedCoverage( + Seq(built), + Seq(survivor, built), + Set(0, 1, 2), + "idx_id")) + } + + @Test + def establishedCoverage_isEmptyWhenNothingOfThisBuildSurvived(): Unit = { + val built = coveringSegment(0) + + assertEquals( + Set.empty[Int], + IndexUtils.establishedCoverage( + Seq(built), + Seq(prunedTo(built, Seq.empty)), + Set(0), + "idx_id")) + } + } From 87745170207a90c09cc5c5daa9b3a81bac81e6c9 Mon Sep 17 00:00:00 2001 From: ivscheianu Date: Wed, 26 Aug 2026 19:22:49 +0300 Subject: [PATCH 2/5] docs: correct the version pinning rationale in the build comment Tasks receive the fragment ids to index from the driver, so the covered set is fixed whether or not the read is pinned. What an unpinned open changes is the version behind those fragments and the version stamped on the segment. The comment also pointed at a function that this change renames. --- .../spark/sql/execution/datasources/v2/AddIndexExec.scala | 8 +++++--- 1 file changed, 5 insertions(+), 3 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 34f075752..dd27a6c10 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 @@ -89,9 +89,11 @@ case class AddIndexExec( val scalarSegmentIndexType = IndexUtils.scalarSegmentIndexType(method) // Plan and build against a single pinned version. Tasks open the dataset themselves, so without - // pinning each one resolves the latest version independently and may see a different fragment - // set than the driver batched, or than its sibling tasks. Coverage the commit cannot establish is - // then accounted for at commit time; see IndexUtils.committedCoverage. + // pinning each one resolves the latest version independently. The driver's batches fix which + // fragments a segment covers either way, but not what sits behind them: siblings would read the + // same fragments at different versions and stamp their segments with those versions, and a + // fragment compacted away since planning fails its task outright. Coverage the commit cannot + // establish is accounted for at commit time; see IndexUtils.establishedCoverage. val (fragmentWorkloads, canonicalColumns, buildReadOptions) = { val ds = Utils.openDatasetBuilder(readOptions).build() try { From ff648f4ebb8a4e730584f2b4544a2129a69b6383 Mon Sep 17 00:00:00 2001 From: ivscheianu Date: Wed, 26 Aug 2026 21:02:37 +0300 Subject: [PATCH 3/5] fix: pin the range-mode BTree build to the planning version Range mode handed its executors unpinned read options, so each builder opened whatever version was latest after draining its shuffle partition. Core stamps a segment with the version of the handle that built it and validates a segment's coverage only when that stamp predates the commit, so a rewrite of the indexed column landing inside the build window left stale keys stamped at the current version and trusted. Predicates on the indexed column then missed rows while an unfiltered scan returned them. Pinning keeps the stamp no newer than the rows behind it, where core prunes the coverage rather than trusting it, so the failure costs coverage instead of correctness. The comments arguing the unpinned build was deliberate had it backwards. A pinned stamp can indeed predate the rows the scan read and lose its coverage to pruning, but that is the safe direction, and leaving the build unpinned does not avoid it: tasks that open before a concurrent rewrite still stamp an older version and are pruned anyway, so unpinned produced both failure modes at once. --- .../datasources/v2/AddIndexExec.scala | 29 ++++++++++--------- 1 file changed, 16 insertions(+), 13 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 dd27a6c10..ab8bdb59d 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 @@ -90,10 +90,11 @@ case class AddIndexExec( // Plan and build against a single pinned version. Tasks open the dataset themselves, so without // pinning each one resolves the latest version independently. The driver's batches fix which - // fragments a segment covers either way, but not what sits behind them: siblings would read the - // same fragments at different versions and stamp their segments with those versions, and a - // fragment compacted away since planning fails its task outright. Coverage the commit cannot - // establish is accounted for at commit time; see IndexUtils.establishedCoverage. + // fragments a segment covers either way, but not the version each segment records, and that + // version is what core checks: it validates a segment's coverage only when the stamp predates + // the commit, so a task that opens after a concurrent rewrite of the indexed column stamps the + // current version over keys it read earlier and the stale keys are trusted. Coverage the commit + // cannot establish is accounted for at commit time; see IndexUtils.establishedCoverage. val (fragmentWorkloads, canonicalColumns, buildReadOptions) = { val ds = Utils.openDatasetBuilder(readOptions).build() try { @@ -168,14 +169,15 @@ case class AddIndexExec( val (nsImpl, nsProps, tableId, initialStorageOpts) = extractNamespaceInfo(lanceDataset, readOptions) - // Range-mode BTree uses preprocessed data from Spark and keeps its dedicated path. It reads the - // table back through the catalog, so its coverage follows the scan rather than the fragment list - // planned above; pinning the build to the planning version would record a segment version older - // than the data the segment holds, which core's staleness pruning would act on. + // Range-mode BTree uses preprocessed data from Spark and keeps its dedicated path: its coverage + // follows the fragment ids in the scanned rows rather than the fragment list planned above. The + // build is pinned like the segmented one all the same. The scan resolves its own version through + // the catalog, so an unpinned executor can open a version newer than the rows it was handed and + // stamp the segment with it, and core only validates segments stamped older than the commit. if (btreeBuildMode.contains("range")) { val segments = new RangeBasedBTreeIndexJob( this.copy(columns = canonicalColumns), - readOptions, + buildReadOptions, fragmentIds.size, nsImpl, nsProps, @@ -300,10 +302,11 @@ case class AddIndexExec( * can be committed directly as a single logical index. * * Unlike the segmented path, coverage here is derived from the fragment ids present in the scanned - * rows rather than from a fragment list fixed at planning time. The scan goes back through the - * catalog, which resolves and pins its own version, so this job is deliberately handed unpinned read - * options: the version an executor opens must not predate the rows it is indexing, or the segment - * would record a dataset version older than its own contents. + * rows rather than from a fragment list fixed at planning time. The read options are pinned even so. + * The scan resolves its own version through the catalog, which is at or after the pinned one, so a + * segment can end up stamped older than the rows it holds and lose its coverage to core's staleness + * pruning; leaving the build unpinned instead lets an executor stamp the current version over rows + * read earlier, and core validates a segment only when its stamp predates the commit. * * @param addIndexExec The AddIndexExec instance that initiated this job * @param readOptions Configuration options for reading the Lance dataset From 6f8a5cce51322ffbb087d7fd50621da348a640c8 Mon Sep 17 00:00:00 2001 From: ivscheianu Date: Thu, 27 Aug 2026 11:08:19 +0300 Subject: [PATCH 4/5] fix: pin the range-mode index scan to the version it stamps Pinning only the executor build left the producer scan resolving its own, later version. Coverage on this path comes from the fragment ids in the scanned rows, so a fragment appended after planning could be declared covered by a segment stamped with a version that predates it. Core skips staleness validation for a fragment absent at the stamped version: prune_stale_segment_coverage takes prune_historically_missing = false from the segment commit path, so such a fragment is retained without comparing its indexed field against the current one, and its keys are trusted however they changed afterwards. A predicate on the indexed column then missed rows a full scan returned. The scan now reads the version the build is pinned to, so the rows a segment holds and the version it records describe one snapshot, and a fragment appended after planning is simply not part of the build, as on the segmented path whose fragment list is fixed at planning time too. Read options that name no version on main now fail rather than falling through to an unpinned scan, since that failure is silent and is the one this pin exists to prevent. The version option gains the positive test it was missing: it has to pin the scan, not merely be accepted. Staleness is narrowed rather than eliminated. The commit's rebase window still waves Operation::Merge through, which core documents can rewrite a column in place; that is unreachable from this connector's DML, where UPDATE COLUMNS commits Operation::Update and is pruned. --- .../datasources/v2/AddIndexExec.scala | 59 ++++++++++++++----- .../lance/spark/branch/BaseBranchDDLTest.java | 22 +++++++ 2 files changed, 67 insertions(+), 14 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 ab8bdb59d..145f04f49 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 @@ -91,10 +91,11 @@ case class AddIndexExec( // Plan and build against a single pinned version. Tasks open the dataset themselves, so without // pinning each one resolves the latest version independently. The driver's batches fix which // fragments a segment covers either way, but not the version each segment records, and that - // version is what core checks: it validates a segment's coverage only when the stamp predates - // the commit, so a task that opens after a concurrent rewrite of the indexed column stamps the - // current version over keys it read earlier and the stale keys are trusted. Coverage the commit - // cannot establish is accounted for at commit time; see IndexUtils.establishedCoverage. + // version is what core checks against. It revalidates a segment's coverage of a fragment only + // when the recorded version predates the commit and the fragment already existed at it, so a + // task opening after a concurrent rewrite of the indexed column stamps the current version over + // keys it read earlier and those keys are trusted unchecked. Coverage the commit cannot + // establish is accounted for at commit time; see IndexUtils.establishedCoverage. val (fragmentWorkloads, canonicalColumns, buildReadOptions) = { val ds = Utils.openDatasetBuilder(readOptions).build() try { @@ -171,9 +172,9 @@ case class AddIndexExec( // Range-mode BTree uses preprocessed data from Spark and keeps its dedicated path: its coverage // follows the fragment ids in the scanned rows rather than the fragment list planned above. The - // build is pinned like the segmented one all the same. The scan resolves its own version through - // the catalog, so an unpinned executor can open a version newer than the rows it was handed and - // stamp the segment with it, and core only validates segments stamped older than the commit. + // pinned options drive both halves of that path, the producer scan and the builder, so the rows a + // segment holds and the version it is stamped with describe the same snapshot; the job explains + // why both are needed. if (btreeBuildMode.contains("range")) { val segments = new RangeBasedBTreeIndexJob( this.copy(columns = canonicalColumns), @@ -302,11 +303,11 @@ case class AddIndexExec( * can be committed directly as a single logical index. * * Unlike the segmented path, coverage here is derived from the fragment ids present in the scanned - * rows rather than from a fragment list fixed at planning time. The read options are pinned even so. - * The scan resolves its own version through the catalog, which is at or after the pinned one, so a - * segment can end up stamped older than the rows it holds and lose its coverage to core's staleness - * pruning; leaving the build unpinned instead lets an executor stamp the current version over rows - * read earlier, and core validates a segment only when its stamp predates the commit. + * rows rather than from a fragment list fixed at planning time, so the producer scan is pinned to the + * same version as the build. Both halves have to describe one snapshot: the segment records the + * version the builder opened, and core validates a segment's coverage against that version only for + * a fragment that existed at it. A scan left to resolve its own version could hand the builder rows + * from a fragment appended afterwards, whose keys would then be committed unvalidated. * * @param addIndexExec The AddIndexExec instance that initiated this job * @param readOptions Configuration options for reading the Lance dataset @@ -327,6 +328,26 @@ class RangeBasedBTreeIndexJob( private val VALUE_COLUMN_NAME = "value" + /** + * The version [[readOptions]] is pinned to. + * + * The caller pins these options with [[IndexUtils.pinVersion]], and CREATE INDEX only runs against + * a writable table, so the ref names a version on main; a branch or tag target is rejected before + * this point by `LanceDataset.ensureWritable`. Anything else cannot be expressed as a scan option + * at all, since branch and version are mutually exclusive there, so it would leave the scan + * unpinned and reading a different snapshot than the build stamps. That is the defect this pin + * exists to prevent, and it is silent, so refuse rather than fall through to it. + */ + private def pinnedVersion: Long = { + val ref = readOptions.getRef + if (ref == null || !ref.isMain || !ref.getVersionNumber.isPresent) { + throw new IllegalStateException( + "Range-mode BTree builds need read options pinned to a version on main so the scan and the " + + "segment stamp describe one snapshot; got a ref that names none") + } + ref.getVersionNumber.get.longValue() + } + def run(): Seq[Index] = { if (addIndexExec.columns.size != 1) { throw new UnsupportedOperationException( @@ -348,9 +369,19 @@ class RangeBasedBTreeIndexJob( } val fullTableName = parts.mkString(".") - // Read the indexed column with the row id and fragment id metadata columns. + // Read the indexed column with the row id and fragment id metadata columns, at the version the + // build is pinned to. The scan otherwise resolves its own version through the catalog, which is + // at or after the pinned one, and coverage here comes from the fragment ids in the scanned rows: + // a fragment appended after planning would be declared covered by a segment stamped with a + // version that predates it, and core skips staleness validation for a fragment absent at the + // stamped version, so its keys would be trusted however they changed afterwards. Pinning the + // scan keeps the rows and the stamp describing one snapshot. Fragments appended after planning + // are simply not part of this build, exactly as in the segmented path, whose fragment list is + // fixed at planning time too; Dataset.optimizeIndices covers them incrementally. val fragmentColumn = LanceDataset.FRAGMENT_ID_COLUMN.name - val df = session.table(fullTableName) + val df = session.read + .option(LanceSparkReadOptions.CONFIG_VERSION, pinnedVersion) + .table(fullTableName) val selectDf = df.select( df.col(columns.head).as(VALUE_COLUMN_NAME), df.col(LanceDataset.ROW_ID_COLUMN.name), diff --git a/lance-spark-base_2.12/src/test/java/org/lance/spark/branch/BaseBranchDDLTest.java b/lance-spark-base_2.12/src/test/java/org/lance/spark/branch/BaseBranchDDLTest.java index 22d202a21..e6d1b3164 100644 --- a/lance-spark-base_2.12/src/test/java/org/lance/spark/branch/BaseBranchDDLTest.java +++ b/lance-spark-base_2.12/src/test/java/org/lance/spark/branch/BaseBranchDDLTest.java @@ -590,6 +590,28 @@ public void testBranchAndVersionOptionsFail() { Assertions.assertTrue(conflictMessages.contains("version")); } + /** + * The version option has to pin the scan itself, not merely be accepted. Distributed index builds + * rely on it to read the same snapshot the build is stamped with, and a silently ignored option + * would put the scan back on the latest version without any visible failure. + */ + @Test + public void testVersionOptionPinsTheScanToThatVersion() { + DatasetVersions versions = prepareDatasetWithHistory(); + Assertions.assertNotEquals(versions.firstInsertVersion, versions.latestVersion); + + Assertions.assertEquals( + 5, + spark + .read() + .option("version", Long.toString(versions.firstInsertVersion)) + .table(fullTable) + .count(), + "reading with the version option must see only the rows present at that version"); + Assertions.assertEquals( + 10, spark.table(fullTable).count(), "the unpinned read still sees the latest version"); + } + @Test public void testBranchIdentifierRejectsVersionAsOf() { DatasetVersions versions = prepareDatasetWithHistory(); From 1afb14b054c0623369ada97d84dc9821d627b324 Mon Sep 17 00:00:00 2001 From: ivscheianu Date: Wed, 2 Sep 2026 21:15:38 +0300 Subject: [PATCH 5/5] docs: trim comments to what the code does not already say Address review feedback: remove comments that narrate the PR history or describe old failure modes, shorten Scaladocs where the method name and signature already convey the intent, and drop test Javadocs whose test names are self-documenting. No logic changes; compilation and tests are unaffected. --- .../datasources/v2/AddIndexExec.scala | 83 +++---------------- .../lance/spark/branch/BaseBranchDDLTest.java | 5 -- .../lance/spark/update/BaseAddIndexTest.java | 11 --- .../datasources/v2/IndexUtilsTest.scala | 10 --- 4 files changed, 10 insertions(+), 99 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 145f04f49..541a9fb92 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 @@ -88,14 +88,7 @@ case class AddIndexExec( val btreeBuildMode = IndexUtils.btreeBuildMode(indexType, args) val scalarSegmentIndexType = IndexUtils.scalarSegmentIndexType(method) - // Plan and build against a single pinned version. Tasks open the dataset themselves, so without - // pinning each one resolves the latest version independently. The driver's batches fix which - // fragments a segment covers either way, but not the version each segment records, and that - // version is what core checks against. It revalidates a segment's coverage of a fragment only - // when the recorded version predates the commit and the fragment already existed at it, so a - // task opening after a concurrent rewrite of the indexed column stamps the current version over - // keys it read earlier and those keys are trusted unchecked. Coverage the commit cannot - // establish is accounted for at commit time; see IndexUtils.establishedCoverage. + // Plan and build at one pinned version; the commit opens the live dataset. val (fragmentWorkloads, canonicalColumns, buildReadOptions) = { val ds = Utils.openDatasetBuilder(readOptions).build() try { @@ -170,11 +163,6 @@ case class AddIndexExec( val (nsImpl, nsProps, tableId, initialStorageOpts) = extractNamespaceInfo(lanceDataset, readOptions) - // Range-mode BTree uses preprocessed data from Spark and keeps its dedicated path: its coverage - // follows the fragment ids in the scanned rows rather than the fragment list planned above. The - // pinned options drive both halves of that path, the producer scan and the builder, so the rows a - // segment holds and the version it is stamped with describe the same snapshot; the job explains - // why both are needed. if (btreeBuildMode.contains("range")) { val segments = new RangeBasedBTreeIndexJob( this.copy(columns = canonicalColumns), @@ -302,13 +290,6 @@ case class AddIndexExec( * covering exactly those fragments, so the resulting segments have disjoint fragment coverage and * can be committed directly as a single logical index. * - * Unlike the segmented path, coverage here is derived from the fragment ids present in the scanned - * rows rather than from a fragment list fixed at planning time, so the producer scan is pinned to the - * same version as the build. Both halves have to describe one snapshot: the segment records the - * version the builder opened, and core validates a segment's coverage against that version only for - * a fragment that existed at it. A scan left to resolve its own version could hand the builder rows - * from a fragment appended afterwards, whose keys would then be committed unvalidated. - * * @param addIndexExec The AddIndexExec instance that initiated this job * @param readOptions Configuration options for reading the Lance dataset * @param numFragments Number of fragments in the dataset, used to bound shuffle partitions @@ -328,16 +309,7 @@ class RangeBasedBTreeIndexJob( private val VALUE_COLUMN_NAME = "value" - /** - * The version [[readOptions]] is pinned to. - * - * The caller pins these options with [[IndexUtils.pinVersion]], and CREATE INDEX only runs against - * a writable table, so the ref names a version on main; a branch or tag target is rejected before - * this point by `LanceDataset.ensureWritable`. Anything else cannot be expressed as a scan option - * at all, since branch and version are mutually exclusive there, so it would leave the scan - * unpinned and reading a different snapshot than the build stamps. That is the defect this pin - * exists to prevent, and it is silent, so refuse rather than fall through to it. - */ + /** Version `readOptions` is pinned to. Throws if the ref is not a version on main. */ private def pinnedVersion: Long = { val ref = readOptions.getRef if (ref == null || !ref.isMain || !ref.getVersionNumber.isPresent) { @@ -369,15 +341,6 @@ class RangeBasedBTreeIndexJob( } val fullTableName = parts.mkString(".") - // Read the indexed column with the row id and fragment id metadata columns, at the version the - // build is pinned to. The scan otherwise resolves its own version through the catalog, which is - // at or after the pinned one, and coverage here comes from the fragment ids in the scanned rows: - // a fragment appended after planning would be declared covered by a segment stamped with a - // version that predates it, and core skips staleness validation for a fragment absent at the - // stamped version, so its keys would be trusted however they changed afterwards. Pinning the - // scan keeps the rows and the stamp describing one snapshot. Fragments appended after planning - // are simply not part of this build, exactly as in the segmented path, whose fragment list is - // fixed at planning time too; Dataset.optimizeIndices covers them incrementally. val fragmentColumn = LanceDataset.FRAGMENT_ID_COLUMN.name val df = session.read .option(LanceSparkReadOptions.CONFIG_VERSION, pinnedVersion) @@ -682,13 +645,7 @@ object IndexUtils extends Logging { .get(method.toLowerCase(Locale.ROOT)) .filter(scalarSegmentIndexTypes.contains) - /** - * Pins `readOptions` to the version `dataset` is open at. - * - * Distributed index builds hand read options to tasks that open the dataset themselves. Pinning - * makes every task observe the fragment set the driver planned over instead of resolving the - * latest version independently. - */ + /** Pins `readOptions` to the version `dataset` is open at. */ def pinVersion( readOptions: LanceSparkReadOptions, dataset: Dataset): LanceSparkReadOptions = @@ -722,21 +679,10 @@ object IndexUtils extends Logging { .toSet /** - * Refuses to publish a segment set that would establish no coverage at all. - * - * Commit intersects each segment's declared coverage with the dataset's live fragments, so a - * fragment retired while the build ran contributes nothing. Lance accepts such a set: an empty - * fragment bitmap is valid metadata, and an existing segment is trivially disjoint from it and so - * survives. Nothing is corrupted, but the transaction would publish segments that index no data and - * report a build that achieved nothing, and this is the last point at which it can still be - * declined. - * - * Partial loss is not refused. A fragment leaves the manifest either because its rows moved - * (compaction, an in-place rewrite) or because they were all deleted; only the first leaves data - * unindexed, and either way the segments covering what remains are correct. Discarding a finished - * distributed build over that would be the wrong response, and it would make a routine concurrent - * DELETE fatal on a long one. What the commit actually established is reported afterwards, by - * [[establishedCoverage]]. + * Refuses a segment set that declares coverage but intersects no live fragment. Partial loss is + * accepted: fragments can be retired by compaction or deletion during the build, and the segments + * covering what remains are still correct. [[establishedCoverage]] reports what was actually + * committed. */ def requireCommittableCoverage( liveFragmentIds: Set[Int], @@ -752,18 +698,9 @@ object IndexUtils extends Logging { } /** - * The coverage a commit established, read back from the metadata the commit returned. - * - * A check taken before the commit cannot answer this, for two reasons. It reads the manifest its - * handle was opened at, while the commit lands on whatever version is current by then. And Lance - * prunes an incoming segment's coverage not only for a fragment that is gone but also for one whose - * indexed field was rewritten under the same id, which no comparison of fragment ids can see. The - * returned metadata is post-pruning, so it is the only truthful account of what was indexed. - * - * Only the segments this build produced are counted. Existing segments that were disjoint from them - * survive the commit, and their coverage is not this command's to report. - * - * @return the fragment ids the commit covered with these segments + * Committed fragment ids from the segments this build produced, intersected with what is still + * live. Only segments whose UUID matches `builtSegments` are counted; pre-existing survivors are + * not this command's coverage. */ def establishedCoverage( builtSegments: Seq[Index], diff --git a/lance-spark-base_2.12/src/test/java/org/lance/spark/branch/BaseBranchDDLTest.java b/lance-spark-base_2.12/src/test/java/org/lance/spark/branch/BaseBranchDDLTest.java index e6d1b3164..3892c3aab 100644 --- a/lance-spark-base_2.12/src/test/java/org/lance/spark/branch/BaseBranchDDLTest.java +++ b/lance-spark-base_2.12/src/test/java/org/lance/spark/branch/BaseBranchDDLTest.java @@ -590,11 +590,6 @@ public void testBranchAndVersionOptionsFail() { Assertions.assertTrue(conflictMessages.contains("version")); } - /** - * The version option has to pin the scan itself, not merely be accepted. Distributed index builds - * rely on it to read the same snapshot the build is stamped with, and a silently ignored option - * would put the scan back on the latest version without any visible failure. - */ @Test public void testVersionOptionPinsTheScanToThatVersion() { DatasetVersions versions = prepareDatasetWithHistory(); diff --git a/lance-spark-base_2.12/src/test/java/org/lance/spark/update/BaseAddIndexTest.java b/lance-spark-base_2.12/src/test/java/org/lance/spark/update/BaseAddIndexTest.java index b51f55986..15e56e24b 100755 --- a/lance-spark-base_2.12/src/test/java/org/lance/spark/update/BaseAddIndexTest.java +++ b/lance-spark-base_2.12/src/test/java/org/lance/spark/update/BaseAddIndexTest.java @@ -108,17 +108,6 @@ public void tearDown() throws IOException { } } - /** - * Pins the two Lance behaviours the coverage report rests on: a segment commit returns the - * metadata of the index as committed, and the handle it was made on advances to the manifest it - * wrote. - * - *

Together they are what make the reported count truthful. A check taken before the commit - * reads the manifest its handle was opened at, so a fragment retired in between is still counted; - * intersecting the returned metadata with the fragments live after the commit is what - * excludes it. This asserts the contract rather than the connector code consuming it, because a - * change on either side would silently make the count overstate again. - */ @Test public void testSegmentCommitReportsCoverageAsCommitted() { spark.sql(String.format("create table %s (id int) using lance", fullTable)); 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 de55fc30a..85fab3bd0 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 @@ -277,8 +277,6 @@ class IndexUtilsTest { () => IndexUtils.batchFragments(fragmentWorkloads(Long.MaxValue, 1), Some(1), 1)) } - // ── declaredCoverage / committedCoverage ─────────────────────── - @Test def declaredCoverage_unionsSegmentBitmaps(): Unit = { assertEquals( @@ -337,8 +335,6 @@ class IndexUtilsTest { assertTrue(error.getMessage.contains("21 total"), error.getMessage) } - // ── establishedCoverage ─────────────────────────────────────────────────── - @Test def establishedCoverage_reportsWhatTheCommitReturned(): Unit = { val built = Seq(coveringSegment(0, 1), coveringSegment(2)) @@ -348,11 +344,6 @@ class IndexUtilsTest { IndexUtils.establishedCoverage(built, built, Set(0, 1, 2), "idx_id")) } - /** - * Lance prunes a fragment whose indexed field was rewritten under the same id, so the committed - * bitmap can be narrower than the one handed in while every fragment is still live. No comparison - * of fragment ids can see that, which is why the report has to come from what the commit returned. - */ @Test def establishedCoverage_followsAPrunedCommitEvenWhileEveryFragmentIsLive(): Unit = { val built = coveringSegment(0, 1) @@ -375,7 +366,6 @@ class IndexUtilsTest { IndexUtils.establishedCoverage(Seq(built), Seq(built), Set(1, 5), "idx_id")) } - /** Existing segments survive a commit they are disjoint from; their coverage is not ours. */ @Test def establishedCoverage_countsOnlyTheSegmentsThisBuildProduced(): Unit = { val built = coveringSegment(2)