[nyl3532016]

Equal-sized fragments are currently assigned to alternating index segments (S0, S1, S0, S1, ...). Lance's compaction planner cannot merge adjacent fragments with different index coverage, so eight small fragments can produce zero compaction tasks after distributed indexing.

Partition selected fragments in dataset order into deterministic contiguous groups, minimizing the largest group's row count. Preserve the existing fragment_group_size API and its group count, ceil(fragment_count / fragment_group_size). Scalar and vector indexing share this policy. Use a uniform estimate of 1 when a row count is unavailable.

Document the compaction tradeoff, partial-build gaps, and the need to rebuild existing interleaved indexes. Add exhaustive cut-placement oracle tests for grouping and end-to-end regressions for two aligned BTREE indexes and an IVF_FLAT index: eight fragments compact into two, with data and indexed query results preserved.

Related fix in lance-ray: lance-format/lance-ray#5319

Validation

Linux Docker container, Python 3.12, pylance 12.0.0, Daft 0.7.14, Ray 2.55.1; dependencies from uv.lock:

• Full NativeRunner suite: 606 passed, 10 skipped, 2 xfailed, 2 xpassed.
• Fragment distribution and compaction tests on Ray with 2 CPUs: 59 passed.
• Replaced the grouping function with the original implementation: the equal-fragment, BTREE, and IVF_FLAT regressions all fail with alternating segment coverage.
• ruff check --no-cache --no-fix daft_lance tests: passed.
• ruff format --check for changed Python files: passed.
• mypy daft_lance/ tests/io/lance/test_fragment_distribution.py: passed (16 source files).
• The broader mypy daft_lance/ tests/ check reports the same 582 errors in 24 files on both unmodified main and this branch; no new typing errors.

[FANNG1]

Required change: split the heaviest run when runs fall short of the segment count

lance-spark's IndexUtils.balancedRunLengths (AddIndexExec.scala) implements the same policy, but when the greedy pass yields fewer runs than the requested segment count, it pops the heaviest run off a priority queue and cuts it at its balancePoint. This PR instead reserves one fragment per remaining group, which front-loads everything and emits a tail of single-fragment segments — exactly the shape CandidateBin::is_noop() discards:

sizes = [5000] + [100]*9, segments=5      # one compacted fragment + small appends
  this PR     : [(1,5000), (6,600), (1,100), (1,100), (1,100)]
  lance-spark : [(1,5000), (3,300), (2,200), (2,200), (2,200)]

sizes = [1e6] + [1000]*99, segments=10
  this PR     : [(1,1e6), (91,91000), (1,1000) x8]
  lance-spark : [(1,1e6), (7,7000), (6,6000), (12,12000), ...]

Pairs are (fragments, rows). Cross-checking 3000 random inputs after porting: the max group row count is identical in every case (both optimal), but this PR produces more single-fragment segments in ~6% of them. End to end, with one 5000-row fragment plus nine 100-row appends and fragment_group_size=2, the segments are [[0],[1..6],[7],[8],[9]] and compaction leaves [5000, 100, 100, 100, 600] — three small fragments that cannot merge with their neighbors under the current coverage layout (a full rebuild can still change that; they are not permanently unmergeable).

So "optimal max group rows" is not sufficient evidence that the PR's compaction goal is met. The splitting step costs nothing in max group rows, and while it cannot mathematically eliminate every single-fragment segment (segments == fragment count is all singletons by definition), it does eliminate the tail singletons this grouping policy creates. Please also add the "large fragment + several small appends" shape to test_fragment_distribution.py.

Suggestion: sort by fragment_id inside the function

lance-spark does fragments.sortBy(_.fragmentId) explicitly, and lance guarantees this is equivalent to manifest order (final_fragments.sort_by_key(|frag| frag.id) at the end of manifest_build.rs, present in both v12.0.0 and HEAD).

Both call sites currently filter the dataset's fragment list, so the order is correct — this is not a defect in the existing fragment_ids path, but a defensive improvement: it puts the invariant in code so a later refactor that builds the list from a user-supplied id order cannot silently produce interleaved coverage. If you take it, update the docstring and the descending-id case in test_contiguous_groups_minimize_maximum_rows (100 - i*3 is an input lance cannot produce; keeping it as an order-preservation test is fine, but it deserves a comment saying so).

Nits

• The per-group statistics log still says Worker %d; these are index segments now.
• README covers the compaction benefit only. Worth one line on the cost in the other direction: too many segments can increase index loading and query overhead (lance-spark's docs note the same for num_segments), so fragment_group_size shouldn't be set too small.
• For the determinism note, state it plainly in the README: multiple indexes on one dataset should be built from the same dataset state with the same grouping parameters.
• Falling back to 1 when count_rows() fails is a clear improvement, and the fallback branch should stay — physical_rows / num_deletions are rt().block_on(...) async calls in the Python binding, returning PyResult and able to raise PyIOError; property syntax doesn't imply no I/O. Just leave a TODO here:

# TODO(pylance): once pylance exposes a bulk fragment-statistics API (the Java
# side already has Dataset.getFragmentStatistics(): metadata-only, returning
# ids + row counts in manifest order), bump the dependency and fetch the stats
# in one call instead of per-fragment count_rows(). Same semantics (physical
# rows - deleted), so grouping boundaries are unchanged; purely a performance
# change, and the failure fallback must stay.

Neither pylance>=12.0.0 nor 14.0.0-beta exposes it on the Python side, so this isn't a blocker for this PR.