fix: make distributed index builds version-consistent and report real coverage by ivscheianu · Pull Request #795 · lance-format/lance-spark

ivscheianu · 2026-08-26T14:12:59Z

Part of #789.

What

Three problems in the distributed CREATE INDEX path, plus the driver-side cost of enumerating fragments.

Tasks did not agree on a version. Each task opens the dataset itself, and nothing pinned the version it opened at, so every task resolved the latest independently. Which fragments a segment covers is fixed by the driver's batch either way, but the version each segment records is not, and that version is what core checks against.

Range mode read one snapshot and stamped another. Its coverage comes from the fragment ids in the scanned rows, and the scan resolves its own version through the catalog, 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, because the segment commit path passes prune_historically_missing = false. Those keys were then trusted however they changed afterwards, and a predicate on the indexed column could miss rows a full scan returned.

The command reported the fragment count it planned, not the one it achieved. Commit intersects each segment's declared coverage with the live fragments, so a fragment retired while the build ran contributes nothing, yet fragments_indexed still counted it.

Change

Both distributed build paths now run against the version the driver planned over. The segmented path hands the pinned options to every task. Range mode pins its producer scan as well, so the rows a segment holds and the version it records describe one snapshot; a fragment appended after planning is simply not part of the build, which was already true of the segmented path.

Read options that name no version on main are refused rather than quietly falling back to an unpinned scan. That fallback is invisible, and it is the exact failure the pin exists to prevent.

commitIndexSegments computes the coverage the commit will actually establish, reports that, and names anything retired in a warning.

Retirement alone does not discard the build. 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. Throwing away a finished distributed build over that would make a routine concurrent DELETE fatal on a long one. Only a total loss is refused, and not because anything would be corrupted: an existing segment is trivially disjoint from an empty bitmap and survives. The transaction would just publish an index over no data and report a build that achieved nothing, and this is the last point where it can be declined.

Fragment enumeration goes through Dataset.getFragmentStatistics(), which returns primitive arrays, instead of getFragments(), which materializes a Java object per fragment and per data file. The driver enumerates twice per command, so on a large table that is the bulk of planning cost.

Not closed

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. Nothing in this connector's DML reaches it, since UPDATE COLUMNS commits Operation::Update and is pruned.

Worth raising upstream separately: the segment commit path validates coverage with prune_historically_missing = false while the merge path passes true, so a caller that hands core a bitmap naming a fragment absent at the segment's own version gets it committed unvalidated. The connector can avoid producing that state, but it cannot detect it.

… 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.

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.

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.

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.

ivscheianu · 2026-09-02T15:05:12Z

hello @geruh, thanks a lot for the review! I'd appreciate if you can take another look when you find some time, thanks a lot!

geruh

Thanks @ivscheianu! Did a closer pass here, and the logic looks great. Just left some nits.

geruh · 2026-09-02T16:05:14Z

+    // 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


pin now sense here now. Can we add a test for this similar to the gatekeeper repro?

The mechanism is tested by testVersionOptionPinsTheScanToThatVersion in BaseBranchDDLTest, same API call this path uses.

I tried to think of a targeted test here but the false-negative scenario needs a concurrent rewrite landing between the scan and the build, and I don't think I can control that timing from a single CREATE INDEX statement without something flaky. Once #794 lands a coverage assertion might work, but right now I'm not sure what to assert that the mechanism test doesn't already cover.

Open to ideas if you see something I missed!

geruh · 2026-09-02T16:08:51Z

      .filter(scalarSegmentIndexTypes.contains)

+  /**
+   * Pins `readOptions` to the version `dataset` is open at.


this is backwards now. We can probably drop the latter half of this comment

geruh · 2026-09-02T16:09:28Z

    val scalarSegmentIndexType = IndexUtils.scalarSegmentIndexType(method)

-    val (fragmentWorkloads, canonicalColumns) = {
+    // Plan and build against a single pinned version. Tasks open the dataset themselves, so without


nit: this comment talks about the old behavior. Let's drop or at the very leas say something like:

// plan and build at one version, and commit stays on the live dataset.

geruh · 2026-09-02T16:17:29Z

+    // 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


nit: similar to the run comment we can drop

geruh · 2026-09-02T16:22:16Z

 * 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


nit: we can drop this section since this is the old failure mode

geruh · 2026-09-02T16:28:23Z

+    }
+  }
+
+  /**


nit: name kind of implies this we can say this builds committed frag ids, and instersected with what is still live or something along those lines.

geruh · 2026-09-02T16:29:20Z

  }

+  /**
+   * Pins the two Lance behaviours the coverage report rests on: a segment commit returns the


nit: java doc on the test. test name is essentially the comment.

geruh · 2026-09-02T16:30:07Z

      () => IndexUtils.batchFragments(fragmentWorkloads(Long.MaxValue, 1), Some(1), 1))
  }
+
+  // ── declaredCoverage / committedCoverage ───────────────────────


nit: do we need the section headers in these files?

not sure, I see that's the pattern on the test files, e.g.:

// ── extractTrain ──────────────────

// ── toJson — SparkOnlyOptions filtering ───────

but happy to remove 👍🏻

geruh · 2026-09-02T16:30:37Z

+  }
+
+  /**
+   * Lance prunes a fragment whose indexed field was rewritten under the same id, so the committed


nit: test should imply this comment

geruh · 2026-09-02T16:31:39Z

  }

+  /**
+   * The version option has to pin the scan itself, not merely be accepted. Distributed index builds


nit: we can drop the java doc here since the test name & logic implies behavior

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.

lance-gatekeeper

⚠️ Gate recommendation: approve with a non-blocking risk.

The latest revision only trims comments and test documentation; the verified snapshot pinning and committed-coverage behavior are unchanged. The range-mode producer scan and executor builder still use the same planning snapshot while the final commit remains live for coverage reconciliation.

The author-accepted chance that a final total-retirement race can publish zero-coverage segments while reporting zero also remains unchanged. No further change is requested for this pull request.

ivscheianu · 2026-09-02T18:29:41Z

Thanks a lot @geruh for the review, really appreciate it! Addressed the comments, please have a look when you get some time. Since I got your attention, I'd appreciate if you can also have a look on the follow-ups, they're small 😄
#793
#794

geruh

LGTM thanks @ivscheianu!

ivscheianu mentioned this pull request Aug 26, 2026

Distributed REFRESH INDEX for incremental scalar index maintenance #789

Open

6 tasks

github-actions Bot added the bug Something isn't working label Aug 26, 2026

ivscheianu mentioned this pull request Aug 26, 2026

feat: add distributed REFRESH INDEX for incremental index maintenance #784

Open