fix(compaction): split tasks to fit source budgets

[everySympathy]

Summary

• keep CandidateBin metadata until source budgets are applied
• trim an oversized candidate task to its largest useful Fragment prefix
• preserve strict cumulative limits for source Fragment count, live rows, and bytes
• avoid emitting a single ordinary small Fragment as a no-op task

Problem

TaskData was formed only from target_rows_per_fragment before per-run source budgets were checked. If the first TaskData exceeded a budget, the complete task was rejected even when a useful prefix of adjacent Fragments fit the configured limit. This could leave a hard-budget compaction plan empty unnecessarily.

For example, a row-sized task containing three Fragments was fully rejected by max_source_fragments = 2. The planner now emits the useful two-Fragment prefix and stops before the third Fragment.

A single Fragment that independently requires rewriting remains eligible. A lone CompactWithNeighbors Fragment remains a no-op and is not emitted.

This addresses the splittable TaskData case discussed alongside #8651. Soft budgets are still useful when one indivisible Fragment itself exceeds a row or byte budget.

Testing

• cargo fmt --all -- --check
• cargo test -p lance test_max_source_ --lib
• cargo test -p lance dataset::optimize::tests --lib (136 passed)
• cargo clippy --all --tests --benches -- -D warnings
• cargo clippy -p lance --all-targets -- -D warnings

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The partial-prefix mechanism fixes the task-granularity stall, but the new execution path does not preserve the compaction contract. A safe revision needs both an ordered partial-commit mechanism—the unresolved bounded-compaction contract in #8400—and prefix selection that converges under repeated hard-budget runs without regressing the invariant fixed by #8513. Please cover the completed behavior with execute-until-noop regressions.

[lance-gatekeeper]

is_noop considers every multi-Fragment prefix useful, but a neighbor-only prefix can produce a below-target Fragment that later becomes permanently isolated. With five 100-row Fragments, target_rows_per_fragment = 250, and max_source_fragments = 2, repeated runs stop at [200, 300]; the 200-row Fragment is still below target but has no candidate neighbor, so planning remains empty. This recreates the failure reported in #8506 and fixed by #8513.

Reproducer

I added this temporary unit test on the observed head:

let mut dataset = lance_datagen::gen_batch()
    .col("a", lance_datagen::array::step::<Int32Type>())
    .into_ram_dataset(FragmentCount::from(5), FragmentRowCount::from(100))
    .await
    .unwrap();
let options = CompactionOptions {
    target_rows_per_fragment: 250,
    max_source_fragments: Some(2),
    ..Default::default()
};
for _ in 0..10 {
    if compact_files(&mut dataset, options.clone(), None).await.unwrap()
        == CompactionMetrics::default()
    {
        break;
    }
}
let sizes = dataset.get_fragments().iter()
    .map(|f| f.metadata.physical_rows.unwrap())
    .collect::<Vec<_>>();
assert!(sizes.iter().all(|rows| *rows >= 250), "{sizes:?}");

Command: cargo test -p lance gate_repro_budget_prefix_does_not_strand_subtarget_fragment --lib -- --nocapture

Expected: once compaction reports no more work, the exactly divisible 500 rows have no sub-target Fragment.

Observed: the assertion failed with [200, 300].

Please define usefulness so neighbor-only prefixes cannot create a non-convergent layout under the configured bound, and add a repeat-until-noop regression for this case.

[lance-gatekeeper]

Pushing an early prefix makes a partial rewrite executable even though replacement Fragments receive IDs above the manifest high-water mark and the manifest remains ID-sorted. The replacement therefore moves behind the untouched suffix and changes scan order. Base returned an empty plan for this tight budget; this head newly exposes the ordering failure on that input.

Reproducer

I added and ran a temporary unit test that created five ordered 100-row Fragments, captured a scan, executed one compaction with target_rows_per_fragment = 250 and max_source_fragments = 2, then asserted the next scan was identical:

let before = dataset.scan().try_into_batch().await.unwrap();
compact_files(&mut dataset, options, None).await.unwrap();
let after = dataset.scan().try_into_batch().await.unwrap();
assert_eq!(after, before);

Command: cargo test -p lance gate_repro_budget_prefix_preserves_row_order --lib -- --nocapture

Expected: values remained 0..499.

Observed: the assertion failed; the scan began at 200, and values 0..199 moved to the end.

The broader ordered bounded-commit contract remains unresolved in #8400. Please keep this new partial-prefix path from becoming executable until a compatible ordered-commit solution exists, or sequence this change after that prerequisite is resolved.

[everySympathy]

Thanks for the concrete reproducers. I confirmed that prefix trimming cannot safely become executable under the current ID-sorted manifest contract: fresh replacement Fragment IDs move an early rewrite behind the untouched suffix, and repeated two-Fragment runs can strand the [200, 300] layout. I will not paper over either invariant locally. This change needs to be sequenced after the ordered partial-commit contract in #8400 is resolved, or replaced by an explicitly approved broader design.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The author’s follow-up confirms that this head changes row order for an early partial rewrite and can converge to a stranded [200, 300] layout. It also clarifies that a local workaround would not preserve the required invariants, so the current revision remains unsafe to accept.

The safe path is to sequence this change after #8400 resolves the ordered partial-commit contract, or replace it with an explicitly approved broader design. The eventual implementation must also retain the convergence invariant fixed by #8513, with execute-until-noop coverage.