[mattfaltyn]

Fixes #2287.

Problem

Parquet scan planning can return multiple byte-range tasks for one source file. Compaction counted and packed those ranges as independent files, then read only each range while scheduling the entire source file for replacement. Partial-progress commits could therefore publish one range's rows and fail on a later group, leaving unread rows absent from the active snapshot.

Change

• Collapse split tasks by source path before compaction counts, candidate selection and bin packing. Preserve per-file delete and lineage metadata without mutating caller tasks.
• Reject incomplete ranges, duplicate files and residual row filters in direct compaction worker groups before I/O. Keep the existing zero-range shorthand for complete files.
• Document the complete-file input contract and add planner, executor and end-to-end atomic/partial-progress regressions.

Tasks passed to the planner come from one coherent scan plan. The fix preserves the exported API; compaction reads complete source files, with concurrency across files.

Validation

The new row-preservation regression fails on the base: the split partial-progress case returns an error after committing only 1 of 12 rows; the unsplit controls pass. The regression preserves all twelve rows with this change, with both atomic and partial-progress execution.

Passed locally (Go checks use Go 1.26.3):

• make test, make test-race, make test-assert, and make vet (including integration-tagged code).
• make lint (golangci-lint v2.14.0; zero issues).
• go build ./... and GOOS=linux GOARCH=s390x CGO_ENABLED=0 go build ./....
• dev/check-api-compat.sh origin/main (no incompatible API changes).
• dev/check-license, mdbook build, and git diff --check.
• make integration-test against Spark 3.5 and Spark 4.0 in a local Linux/arm64 Docker runner. The Spark 4 run disables test-result caching.
• make integration-external-rest against Apache Polaris 1.8.0.

CI results are available in the PR checks.

[laskoviymishka]

Thanks @mattfaltyn for the split x partial regression. As I read the base path, both split cases fail there: partial progress commits 1 of 12 rows, and atomic fails at commit because the same file is queued for deletion 12 times. I think this is fine overall, but I'd like to strengthen the first-task test a bit before approving. The residual question on the collapse loop is worth covering too, though it isn't a blocker.

[laskoviymishka]

Should PlanCompaction return ErrInvalidOperation when a later task for the same path carries different delete files or FirstRowID? Today it keeps the first and skips the rest without comparing, while ExecuteCompactionGroup rejects a duplicate path outright. Splits from one scan are copies of a single task, so nothing in-repo hits this. The check would enforce the "one coherent scan plan" precondition instead of only documenting it.

[laskoviymishka]

Could this also set task.Residual = nil, to match Java's planner, which scans with ignoreResiduals()? As it stands, a task planned from Scan(WithRowFilter(...)) with a data-column predicate keeps its residual through planning and is only rejected later by the new check in ExecuteCompactionGroup. Analyze scans unfiltered, so this isn't a blocker. It's more a question of where we want that contract to live.

[laskoviymishka]

Could the two same-path tasks differ in more than the range? second := first copies the DV, FirstRowID and sequence number, and require.Contains compares with reflect.DeepEqual, so this passes whether the planner keeps the earlier task or the later one. Since second sits first in tasks, giving first its own FirstRowID after the copy would pin the first-task rule the doc comment describes.

[mattfaltyn]

Thanks @laskoviymishka! All three points are addressed: the planner rejects conflicting delete or lineage metadata, clears residuals for whole-file rewrites, and the first-task test now checks object identity. Added regressions for metadata conflicts and filtered rewrites in both atomic and partial-progress modes. You're right that both split cases fail on the base. Full local checks and CI pass.

[zeroshade]

Thanks, this closes both loss paths: split ranges were packed as independent files, and residual filters dropped rows while the whole source file was replaced. I merged it with current main locally and the tests pass. With compaction.go and rewrite_data_files.go reverted, TestCompactionPreservesSplitFileRows fails in all four split cases and in both unsplit filtered cases, so it pins both.

One non-blocking nit: sameCompactionMetadata compares delete files with reflect.DeepEqual, but dataFile fills some fields lazily (initPartition/initColumnStats and the stats maps). Two identical delete files stop comparing equal once one has had its bounds or partition read. Local split tasks share pointers, so DeepEqual short-circuits, but delete files decoded from separate REST responses don't, so it could raise a spurious "different metadata" error. Comparing identity fields (path, content offset/size) would be sturdier, in a follow-up if you like. LGTM.