feat: add distributed zonemap index build with configurable segments

[beinan]

Summary

• Add zonemap as a new index type in CREATE INDEX DDL with distributed build support
• Batch fragments into configurable segments via num_segments option (defaults to spark.default.parallelism)
• Each segment is built in parallel on Spark executors and committed as a logical index on the driver
• Zonemap indexes currently support single column only

What Changed

• AddIndexExec.scala: Zonemap-specific path with ZonemapIndexJob/ZonemapIndexTask and commitIndexSegments
• create-index.md: Document zonemap index type, options, and usage
• Tests: unit tests for segment creation/validation and integration test

Notes

• Rebased cleanly onto current main
• Depends on lance-core 7.0.0-beta.10 or newer which includes zonemap segment support
• Supersedes PR feat: add distributed zonemap index build #473 and closed PR feat: support zonemap indexes in ALTER TABLE CREATE INDEX #466

Test plan

• CI passes (lint, unit tests, integration tests across all Spark/Scala versions)
• Zonemap index creation with default segment count
• Zonemap index creation with explicit num_segments
• Repeated zonemap index creation replaces existing segments
• Query correctness after zonemap index creation

🤖 Generated with Claude Code

[hamersaw]

This looks pretty good. Thanks for the PR! A few things that we should tighten up IMO.

[beinan]

Thanks for the thorough review! All feedback has been addressed in the latest push (force-pushed as a single clean commit on latest main):

Scan-side changes removed entirely:

• Removed useScalarIndex, forcePostScanFiltering, and shouldForcePostScanFiltering — zonemap fragment pruning works via the existing ZonemapFragmentPruner path without needing special scan flags. No scan-side files are modified in this PR anymore.

Index creation fixes:

• Race in commitIndexSegments: Now captures pre-commit UUIDs and only removes indexes with those UUIDs, so concurrent writers' segments are never deleted.
• batchFragments accuracy: Switched from ceil(N/K) to index-based slicing (slice(i*N/K, (i+1)*N/K)) to guarantee the requested segment count.
• num_segments validation: Bounds check for <= 0 and type validation, passed through constructor (no duplicate extraction).
• Segment failure handling: try/catch with clear error message about Lance GC cleanup.

[beinan]

@hamersaw All review feedback has been addressed — the key change since your last review is simplifying commitIndexSegments to rely on Lance core's built-in atomic replacement (single transaction for add+remove). Would you mind taking another look when you get a chance? Thanks!

[hamersaw]

@LuciferYang do you mind making a pass here? Specifically, I'm interested with how this compares to your proposal (#513) to support building distributed ZoneMap indexes.

[LuciferYang]

@LuciferYang do you mind making a pass here? Specifically, I'm interested with how this compares to your proposal (#513) to support building distributed ZoneMap indexes.

will give feedback later today.

[LuciferYang]

@hamersaw @beinan Had a closer look. After the May 12 force-push on #516, the two PRs are adjacent rather than overlapping.

	#516	#513 distributed	#513 consolidated (opt-in)
Tasks	1 per num_segments batch	1 per fragment	1 per fragment (compute), driver write
Segments on disk	num_segments (default = min(fragments, defaultParallelism))	= fragment count	1
Commit API	commitExistingIndexSegments	manual AddIndexOperation + Transaction + CommitBuilder	same as distributed
Upstream blocker	none (project's already on lance-core 7.0.0-beta.10)	none	lance-format/lance#6779 + #6780, both unmerged

The main difference between the two distributed paths is that num_segments on #516 is one knob doing two jobs: parallelism and segment count are the same lever. At num_segments=1 you get one segment, but the work serialises onto a single executor — not the same as #513's consolidated path, which keeps compute parallel and only centralises the write. That decoupled corner only opens up once #6779/#6780 land.

For reference, sf=100 store_sales (234 fragments, ~288M rows) under #513: distributed = 15.0 s / 234 segments / 1.1 MB; consolidated = 28.1 s / 1 segment / 138 KB. The ~8× footprint drop is per-file header amortisation.

What I'd want to inherit from #516:

• It uses commitExistingIndexSegments. That's the right API; feat(sql): support USING zonemap with distributed and consolidated build paths #513 reimplements the add+remove transaction by hand and should be rebased onto it.
• Dropping the scan-side useScalarIndex / forcePostScanFiltering machinery in the May 12 force-push was the right call — pruning works through ZonemapFragmentPruner without it.

Suggested path:

1. Land feat: add distributed zonemap index build with configurable segments #516 as-is.
2. After it lands, I'll rebase feat(sql): support USING zonemap with distributed and consolidated build paths #513 distributed onto commitExistingIndexSegments. The fragment-id exception wrapping in ZonemapFragmentTask is worth porting onto ZonemapIndexTask while we're there. After that, feat(sql): support USING zonemap with distributed and consolidated build paths #513 only carries the consolidated path.
3. Consolidated lands as an opt-in (spark.lance.zonemap.consolidate.enabled) once #6779/#6780 release. Default-off — driver allocator holds every per-fragment Arrow batch until writeZonemapIndexFromBatches consumes them, so it can regress at very-high fragment counts.

One nit on #516, will leave inline: targetTasks = math.min(fragmentIds.size, n) silently clamps num_segments=1000 to fragment count. The doc string reads like num_segments is a target, not an upper bound. Either log when clamping or reword the doc.

[beinan]

@LuciferYang Thanks for the thorough comparison — the side-by-side table is really helpful.

You're right about num_segments doing double duty. The latest push addresses the nit:

• num_segments doc now clarifies it's an upper bound clamped to fragment count
• Added a log message when clamping occurs so it's not silent
• Switched batchFragments to index-based slicing to guarantee the requested segment count

Agreed on the suggested path — happy to land #516 as the distributed foundation, then have #513 rebase its distributed path onto commitExistingIndexSegments and carry the consolidated path as an opt-in once #6779/#6780 land. The fragment-id exception wrapping from ZonemapFragmentTask is a good addition to port over as well.

[LuciferYang]

Does the PR description also need to be updated?

[LuciferYang]

The "What Changed" section still references LanceScanBuilder.java, LanceScan.java, LanceInputPartition.java, LanceFragmentScanner.java, and LanceCountStarPartitionReader.java, plus the bullet "Add segmented zonemap scan support with Spark-side post-scan filtering fallback" in Summary. None of this is in the current diff.

others LGTM

[beinan]

The "What Changed" section still references LanceScanBuilder.java, LanceScan.java, LanceInputPartition.java, LanceFragmentScanner.java, and LanceCountStarPartitionReader.java, plus the bullet "Add segmented zonemap scan support with Spark-side post-scan filtering fallback" in Summary. None of this is in the current diff.

others LGTM

Sorry for my delay, just updated. can we merge this pr? @LuciferYang @hamersaw

[hamersaw]

Thanks for the iterations on this!