[core] Optimize data evolution compaction planning

[JingsongLi]

Summary

• plan data-evolution compaction candidates from projected manifest metadata instead of retaining full file entries for the whole scan
• select exact row-id ranges from first row ID, row count, file size, and file type, then read full metadata only for the manifests intersecting each selected batch
• move the 100,000-candidate soft batching target into DataEvolutionCompactRangePlanner; a single logical range may exceed it, while legacy manifests without row-id bounds use one full scan instead of repeatedly rescanning the same group
• group normal files by their own contiguous row-id coverage before attaching blob/vector files, so a spanning dedicated file cannot reconnect candidate ranges across a non-candidate gap
• defensively reject disjoint inputs in DataEvolutionNormalCompactTask
• keep the legacy data-evolution.compaction.rewrite-row-ids option for compatibility, but reject true; normal data-evolution compaction continues to preserve row IDs and logical deletions
• add an explicit materialize_deletion_vectors procedure for Flink and Spark to physically apply deletion vectors, assign new row IDs, remove the applied DVs, and drop affected global indexes

Why

DataEvolutionCompactCoordinator previously loaded full ManifestEntry and DataFileMeta objects for every live file before deciding which files needed compaction. Large data-evolution tables could therefore consume several gigabytes of heap and fail with OOM even when only a small subset of files were compact candidates.

The new two-phase planning flow first scans only the projected fields needed to identify exact candidate row-id ranges. Full file metadata is then loaded only for selected ranges. For manifests with row-id bounds, each range batch carries only intersecting manifests. For legacy manifests, all candidate ranges are consolidated into one full scan to avoid repeated full-group scans. Candidate metadata uses primitive arrays with 32 bytes of payload per live file, substantially reducing the planning footprint.

The full-metadata result is not a strict 100,000-file bound: one logical candidate range, or files intersecting a selected range, may exceed the soft target. This preserves row-range atomicity and is now explicit in the implementation.

The former rewrite-row-ids compaction option mixed two operations with different contracts. Regular compaction must retain stable row IDs and logical deletion state, while physically removing deleted rows necessarily assigns new row IDs. The new procedure makes that destructive operation explicit:

CALL sys.materialize_deletion_vectors(table => 'T');

Both Flink and Spark support optional partitions, options, and where arguments (partitions and where are mutually exclusive). The operation batches work by row-id/manifest ranges, rewrites only batches containing deletion vectors, and uses the existing data-evolution compaction commit preparation to remove DVs and invalidate affected global indexes. It also enables snapshot-based row-ID conflict detection so concurrent changes fail instead of committing stale results. Vector-store file materialization is rejected until it has a safe implementation.

Memory benchmark

A lightweight retained-heap benchmark used live normal files in one row-id/manifest group, which represents the old planner's problematic large-group case. Old and new modes ran in separate JVMs with Serial GC and full GC before and after allocation. The 500,000-file run used a 1.5 GiB heap; the 5,000,000-file run used a 4 GiB heap.

Live files	Old full metadata retained	New primitive metadata retained	New radix-array peak model	Retained reduction	Peak reduction
500,000	169.82 MiB (356.15 B/file)	15.27 MiB (32.01 B/file)	30.77 MiB (64.52 B/file)	91.0%	81.9%
5,000,000	1,697.61 MiB (356.01 B/file)	152.60 MiB (32.00 B/file)	183.36 MiB (38.45 B/file)	91.0%	89.2%

The old representation is a List<ManifestEntry> containing full DataFileMeta objects after copyWithoutStats. The new retained value is the measured heap for CompactCandidateRangeCollector. The radix-array peak model adds its auxiliary sort array and counters; for 5 million files, sorting is bounded to one 1-million-entry chunk at a time.

This benchmark isolates the candidate metadata retained during planning rather than measuring whole-process RSS. Actual coordinator memory also depends on manifest grouping, deleted-file identifiers, partitions, selected ranges, and the full metadata intersecting the current range batch, but the per-live-file reduction is representative of the OOM-sensitive part changed by this PR.

Impact

• reduces coordinator memory usage for tables with large file counts
• preserves compaction decisions for normal, blob, and vector files without allowing dedicated files to bridge normal-file gaps
• avoids repeated full-manifest scans for legacy metadata and prunes modern scans by manifest row-id bounds
• keeps compatibility for manifests without row-id bounds
• preserves stable row IDs and logical deletions during normal data-evolution compaction
• makes data-evolution.compaction.rewrite-row-ids=true fail fast and direct users to the explicit procedure
• provides a separately invoked Flink/Spark operation for DV materialization, with partition filtering, conflict detection, and documented row-ID/global-index impact

Validation

• mvn -pl paimon-core -Pfast-build -DwildcardSuites=none -Dtest=DataEvolutionCompactCoordinatorTest,CompactCandidateRangeCollectorTest,DataEvolutionCompactRangePlannerTest,DataEvolutionDeletionVectorTest,DataEvolutionTableTest,FullTextSearchBuilderTest,VectorSearchBuilderTest test (174 tests)
• focused regressions for spanning dedicated files, disjoint normal task rejection, per-batch manifest pruning, and one-scan legacy compatibility
• mvn -pl paimon-core -Pfast-build -DwildcardSuites=none -Dtest=DataEvolutionDeletionVectorTest#testMaterializeDeletionVectors+testCompactRejectsRewriteRowIdsOption test (2 tests)
• mvn -pl paimon-flink/paimon-flink-common -Pfast-build -Pflink1 -DwildcardSuites=none -Dtest=DataEvolutionDeleteSqlITCase#testMaterializeDeletionVectorsProcedure test
• mvn -pl paimon-spark/paimon-spark-3.5 -am -Pfast-build -Pspark3 -DfailIfNoTests=false -DwildcardSuites=org.apache.paimon.spark.sql.DataEvolutionDeletionTest -Dtest=none test (16 tests)
• mvn -pl paimon-docs -am -Pfast-build -DfailIfNoTests=false -DwildcardSuites=none -Dtest=ConfigOptionsDocsCompletenessITCase test
• mvn -pl paimon-api,paimon-core,paimon-flink/paimon-flink-common,paimon-spark/paimon-spark-common,paimon-spark/paimon-spark-ut -Pflink1 -Pspark3 -DskipTests spotless:check

[leaves12138]

I found a correctness issue in the two-phase planning flow when a dedicated file spans multiple normal row-id ranges.

A RangeBatch may contain multiple independent candidate ranges. The second scan uses withRowRanges, which includes every file that intersects any selected range, and then passes all returned entries to one CompactPlanner invocation. A spanning blob/vector file can therefore reconnect candidate ranges that were intentionally separated by a non-candidate normal range. CompactPlanner builds its outer range groups from all files, including dedicated files, so it may produce a normal compaction task across a row-id gap.

A minimal example is:

• normal [0, 4]: two small files, plus two updated blob files spanning [0, 12] (blob candidate)
• normal [5, 9]: one oversized non-candidate file
• normal [10, 19]: four small files (normal candidate)

The candidate collector correctly returns [0, 4] and [10, 19]. However, the second scan also returns the [0, 12] blobs. They reconnect both ranges inside CompactPlanner, which then generates a normal task containing files from [0, 4] and [10, 19], while excluding [5, 9].

DataEvolutionNormalCompactTask concatenates these disjoint ranges and assigns the output the first input row ID. This shifts the latter rows and can create an output range overlapping the untouched [5, 9] file. The commit-side existence check does not necessarily reject this because the new range is still contained in the original overall coverage.

I reproduced this deterministically and also found it with a 20,000-iteration differential test comparing:

1. CompactPlanner on all entries; and
2. candidate collection -> row-range filtering -> CompactPlanner.

The normal-only differential test passed, while the blob version found this mismatch.

I think candidate boundaries must be preserved in phase two. One possible fix is to group normal files by their own contiguous row-id coverage first, and only then attach blob/vector files to their anchor normal range, so dedicated files cannot define normal-range connectivity. It would also be valuable to add checkContiguousRowRange(compactBefore) to DataEvolutionNormalCompactTask as a defensive safeguard.

Two additional concerns:

• data-evolution.compaction.rewrite-row-ids=true now fails with guidance to use a separate deletion-vector materialization operation, but I could not find an exposed Flink/Spark action or procedure providing that replacement. This changes an existing capability into a hard failure without a migration path.
• Each candidate batch rescans the complete manifest group. For legacy manifests without row-id bounds, all manifests form one group, so more than 100,000 candidate files may cause repeated full-manifest scans. The 100,000-file value is also an estimated candidate count rather than a strict bound on full metadata returned by the intersecting-range scan.

[JingsongLi]

Thanks for the detailed reproduction. Addressed in 129068954c:

1. Dedicated-file bridge correctness: CompactPlanner now derives contiguous groups from normal files only, then associates BLOB/vector files with the normal file containing their first row ID. Dedicated files can no longer define normal-range connectivity. DataEvolutionNormalCompactTask also calls checkContiguousRowRange as a defensive fail-fast. The regression test follows the reported shape: two candidate normal ranges separated by an omitted oversized range, with BLOB files spanning across the gap.

2. Rewrite-row-ids migration wording: confirmed that there is currently no exposed Flink/Spark operation equivalent to the removed physical DV materialization path. reassign_row_id is metadata-only and is not a replacement. I removed the inaccurate “run a separate operation” guidance from the exception, option description, generated configuration, and user docs; they now explicitly state that data-evolution compaction preserves logical deletions and that no standalone materialization action is currently exposed. Keeping true unsupported is intentional here; restoring that capability would be a separate compatibility decision rather than pointing users at a nonexistent path.

3. Manifest rescans / batching: for manifests with row-id bounds, each RangeBatch now carries only manifests intersecting its candidate ranges. For legacy manifests without bounds, all candidate ranges are consolidated into one batch so the full manifest group is scanned once instead of once per candidate batch. The constant is renamed to CANDIDATE_FILES_PER_BATCH and documented as a soft target, since a logical range or its intersecting full metadata can exceed it.

Added regression coverage for the bridge case, disjoint-task rejection, manifest pruning, and legacy one-scan behavior. The related 174 core tests, dependent Flink/Spark compile, Spotless, and ConfigOptionsDocsCompletenessITCase pass locally.

[leaves12138]

Rechecked the latest update at 129068954c.

The previously reported correctness blocker is resolved:

• normal-range connectivity is now derived exclusively from normal files, so a spanning BLOB/vector file cannot reconnect candidate ranges across an omitted normal range;
• dedicated files are associated only after the normal contiguous groups are established;
• DataEvolutionNormalCompactTask now fails fast on disjoint row-id inputs;
• bounded manifests are pruned per candidate batch, while legacy manifests without row-id bounds consolidate candidates into one full scan.

I reran the affected 174 core tests successfully. I also reran the 20,000-iteration normal and BLOB differential checks against full planning; both passed with the fix. The original spanning-BLOB reproduction now produces two independent normal tasks and no row-id-shifting task.

I did not find another correctness issue in the updated implementation. The removal of rewrite-row-ids=true remains an explicit compatibility/product decision, but the documentation and exception no longer suggest a nonexistent replacement operation.

[leaves12138]

Rechecked the latest update. The reported correctness issue is resolved, and the affected tests and differential checks pass.

[JingsongLi]

Added the explicit DV materialization path in 3946674: Flink and Spark now expose CALL sys.materialize_deletion_vectors(...). Normal data-evolution compaction still preserves row IDs and rejects data-evolution.compaction.rewrite-row-ids=true; the new procedure physically applies DVs, reassigns surviving row IDs, drops affected global indexes, and enables snapshot conflict checks. Core, Flink end-to-end, Spark 3.5 DataEvolutionDeletionTest (16 tests), docs completeness, and Spotless all pass locally. The PR description and user docs have been updated.

[leaves12138]

I found a blocker in the Spark multi-batch execution path for deletion-vector materialization.

DataEvolutionDeletionVectorMaterializeCoordinator intentionally returns full-metadata work in batches. However, CompactProcedure.executeDataEvolutionCompaction commits every taskPlanner.get() result separately, while every batch runs DataEvolutionCompactionCommitPreparation(table, snapshot) against the same original snapshot.

This breaks when two batches from the same partition touch deletion vectors stored in the same DV index file:

1. Batch 1 materializes its range. The preparation rewrites the shared DV index file, removing batch 1's DV while preserving batch 2's DV, and the commit replaces the original index file.
2. Batch 2 still prepares from the original snapshot. It therefore rewrites the already-replaced index file and also carries forward batch 1's now-stale DV.
3. The second commit fails deterministically with a file-deletion conflict; my reproduction reaches Trying to create deletion vector on file ... which is not previously added.

I reproduced the actual planner/commit loop by using one partition with two disjoint manifest groups, putting both DVs in one index file, and reducing FILES_PER_BATCH to force two plan() calls. The first batch commits and the second batch fails. With the production limit, the same path is reached when an earlier manifest-group batch contains at least 100,000 live files and a later group in the same partition shares a DV index file.

The existing tests collect every planned task and prepare/commit them once, so they do not exercise the Spark behavior at lines 561-639.

Please either prepare and commit all materialization batches atomically (similar to the Flink non-parallel preparation operator), or maintain DV rewrite state correctly across batches without reloading the original DV index state. The latter must still reject an external concurrent DV update. A configurable batch-size regression test would be valuable. The same stale-snapshot pattern should also be checked for regular Spark data-evolution compaction when a DV index file spans planner batches.

[leaves12138]

Rechecked the update at 1e12e1e759 and the previous Spark multi-batch correctness blocker is resolved.

The Spark loop now advances the snapshot used for DV/index commit preparation after each successful batch, so a shared DV index file is maintained from the state produced by the preceding batch. At the same time, materialization continues to use the original planning snapshot as the row-ID conflict baseline, which preserves detection of external concurrent updates instead of treating earlier batches from this operation as conflicts.

I verified both new Spark 3.5 regressions independently:

• deletion-vector materialization across planner batches;
• normal data-evolution compaction preserving deletion vectors across planner batches.

I also reran 65 relevant core tests covering candidate collection, range batching, planner boundaries, dedicated-file association, DV rewriting/materialization, and serialization. All passed.

The overall separation now looks sound: projected metadata selects logical ranges, full metadata is read only for selected batches, normal-file connectivity is established before attaching dedicated files, and commit preparation centrally maintains DVs and invalidates global indexes. The optimistic commit provider also refreshes global-index deletions from the latest snapshot on retries. I did not find another correctness blocker in the current design.

[leaves12138]

The DV-first batching direction is a good optimization, and I rechecked the Spark/Flink separation and the existing multi-batch regressions. However, there is still a correctness blocker when a dedicated file spans beyond the selected DV anchor range. Please see the inline comment for a concrete reproduction.

I locally verified that the existing core materialization tests and both Spark 3.5 multi-batch tests still pass, while the dedicated-file boundary reproduction fails consistently.

[leaves12138]

The selected ranges are derived only from DV anchor files, but the resulting scan can include a dedicated file whose physical range extends beyond those anchors without including all normal files needed to cover that dedicated file. This creates an incomplete logical read group.

A local reproduction is:

1. normal ranges: [0,4], [5,9], [10,14];
2. write a partial BLOB update as one file covering [5,14];
3. create a DV only for the anchor in [5,9];
4. run materialization.

This scan selects [5,9], includes the BLOB file [5,14] because it intersects, but excludes the normal file [10,14]. DataEvolutionMaterializeDeletionCompactTask then fails in DataEvolutionSplitRead with The merged rowCount 10 of blob file bunch should be aligned with normal files 5.

Please expand the batch to the complete normal-file coverage required by every included dedicated file (and repeat to closure if necessary), or otherwise ensure a materialization task never contains a partial dedicated-file read group. A regression with a BLOB/vector file spanning adjacent normal ranges would be useful.