[core][spark]: support rewrite_manifest procedure by zhongyujiang · Pull Request #9195 · apache/paimon

zhongyujiang · 2026-08-12T12:40:30Z

Motivation

Manifest entries are written in commit order, which can become unordered over time. This makes ManifestFileMeta statistics (partitionStats / minBucket / maxBucket / minLevel / maxLevel) loose, reducing manifest-level pruning during scan.

This PR adds a rewrite_manifest Spark procedure that reads all manifest entries, sorts them globally by partition -> bucket -> level -> fileName (canceling ADD/DELETE pairs of the same file), and writes them back as new manifest files with compact statistics.

Usage

CALL sys.rewrite_manifest(table => 'db.table')
CALL sys.rewrite_manifest(table => 'db.table', where => 'dt >= "2024-01-01" AND dt < "2024-02-01"')

An optional where clause restricts the rewrite to manifests whose partition stats may match the predicate; the remaining manifests are left untouched.

Implementation

ManifestEntrySortKey: serializable sort key (partition, bucket, level, fileName). Partition stored as byte[] for Kryo compatibility (BinaryRow's Java serialization callbacks are skipped by Kryo). RecordComparator is lazy-initialized per executor.
sortByKey + RangePartitioner: same key for ADD and DELETE of the same file guarantees they land in the same Spark partition and can be cancelled.
Streaming cancellation: per-key buffer (at most one ADD + one DELETE), order-independent mergeGroup, duplicate detection throws.
Single-file writer: each task produces one manifest via createManifestEntryWriter (non-rolling). Empty manifests are deleted.
Auto parallelism: floor(estimatedOutputSize / targetSize) - fewer, slightly-over-target manifests preferred over more under-target ones.
where clause: Spark expression parsing + transformFieldMapping (full rowType to partitionType index mapping) + PartitionPredicate stats pruning.
replaceManifest commit: optimistic concurrency with retry; conflict detection (current base must containAll removed manifests); concurrent delta manifests preserved.

Tests

ManifestEntrySortKeyTest (9): partition/bucket/level/fileName ordering, ADD/DELETE same key, serialization survival
FileStoreCommitTest (3): identity replace, concurrent delta preservation, concurrent compact conflict
RewriteManifestProcedureTest (8): DELETE cleanup, global ordering, where filtering, range where, unpartitioned table, non-partition where throws, size bounds

zhongyujiang · 2026-08-17T02:53:32Z

@JingsongLi Hi, can you help review this when you have time? I added this to sort manifest entries, which speeds up both scan and manifest commit a lot.

zhongyujiang · 2026-08-17T02:55:11Z

        procedureBuilders.put("reset_consumer", ResetConsumerProcedure::builder);
        procedureBuilders.put("mark_partition_done", MarkPartitionDoneProcedure::builder);
        procedureBuilders.put("compact_manifest", CompactManifestProcedure::builder);
+        procedureBuilders.put("rewrite_manifest", RewriteManifestProcedure::builder);


Added rewrite_manifest instead of reusing the old compact_manifest, since this operation does not necessarily produce a compaction.

JingsongLi

The core replacement protocol and sort-key behavior look consistent, and the 12 focused core tests pass. I found two blockers around public API compatibility and Spark task-attempt output ownership.

JingsongLi · 2026-08-20T02:39:19Z

+                                            FileStoreTable t = tableRef;
+                                            FileStorePathFactory pathFactory =
+                                                    t.store().pathFactory();
+                                            Path manifestPath = pathFactory.newManifestFile();


[P2] Make task-attempt manifest output abortable

Each Spark task writes its final manifest directly into the table metadata directory. If a later partition fails, collect() throws before the driver receives newManifests, so all manifests already completed by successful partitions are leaked. A retried/speculative attempt can likewise create a second file while only one attempt’s metadata is returned. This can leak a full rewrite’s worth of manifests on every failed run. Please use task-attempt temporary output plus an attempt commit/abort protocol (and clean all produced paths when the stage/commit fails), and add a test that injects a failure after another partition has finished writing.

Thanks for the review!
I've added a catch block in mapPartitions that calls writer.abort() to delete the manifest file when a task fails (including task-killed scenarios).
I didn't make the writer output to a temorary output because it would require an extra rename during the final commit, and I feel this overhead is not justified since the probability of collect() failing is relatively low. What do you think?

JingsongLi

Can you take a look to BatchTableCommit.compactManifests?

zhongyujiang · 2026-08-23T15:38:45Z

Can you take a look to BatchTableCommit.compactManifests?

I'm aware that compactManifests already supports a manifest sorting capability. The new added rewrite_manifest provides a complementary option: a one-time full distributed sort via Spark sortByKey with where support, suitable for manual maintenance on large tables. Would you prefer me to merge these into one procedure, or keep them separate?

Add a rewrite_manifest Spark procedure that reads all manifest entries, sorts them globally by partition -> bucket -> level -> fileName (canceling ADD/DELETE pairs of the same file along the way) and writes them back as new manifest files so that ManifestFileMeta statistics become more compact for scan pruning. The sort runs distributed via a Spark sortByKey shuffle. - ManifestEntrySortKey: serializable sort key with byte[] partition for Kryo compatibility and lazy RecordComparator - FileStoreCommit#replaceManifest: optimistic-concurrency commit with conflict detection (containsAll) and concurrent delta preservation - RewriteManifestProcedure: distributed sortByKey + streaming ADD/DELETE cancellation + where clause + auto parallelism - FormatTableCommit: add unsupported replaceManifests - register rewrite_manifest in SparkProcedures - tests: ManifestEntrySortKeyTest (9), FileStoreCommitTest replaceManifest conflict scenarios (3), RewriteManifestProcedureTest (8) Co-Authored-By: Claude <noreply@anthropic.com>

rewritten_manifests_count should be the number of input manifests rewritten (currentManifests), not the output. added_manifests_count should be the number of new manifests produced (newManifests), not the net delta. Also rename the post-filter variable to manifestsToRewrite for clarity. Co-Authored-By: Claude <noreply@anthropic.com>

Update the assertion to match the corrected semantics: rewritten_manifests_count == input manifest count, added_manifests_count == output manifest count (> 0). Co-Authored-By: Claude <noreply@anthropic.com>

The test expected compactManifest to change manifest file names, but with default settings the merge may be a no-op (same manifests, no new snapshot). Use a small manifest.target-file-size and manifest.full-compaction-file-size to force compactManifest to actually rewrite the manifests, making the stale manifestsBefore absent from the current base. Co-Authored-By: Claude <noreply@anthropic.com>

Apache paimon master uses ManifestAvroWriter (not ManifestEntryWriter). Replace createManifestEntryWriter(Path) with createAvroWriter(Path), and use result().isEmpty() + abort() for empty manifest cleanup instead of recordCount() + path().getName(). Co-Authored-By: Claude <noreply@anthropic.com>

compactManifest may be a no-op (same manifests, no new snapshot) so the stale manifestsBefore stays in currentBase. Instead, manually rewrite the manifests to new files via manifestFile.write(), then replaceManifest — this changes file names, guaranteeing the subsequent replaceManifest with the stale manifestsBefore throws "Manifest conflict". Co-Authored-By: Claude <noreply@anthropic.com>

When a Spark task fails during mapPartitions, call writer.abort() to delete the partially written manifest file instead of leaving it as an orphan. This addresses the reviewer's concern about leaked manifests on task failure. Co-Authored-By: Claude <noreply@anthropic.com>

zhongyujiang · 2026-08-26T12:44:51Z

Hi @JingsongLi, comments has been resolved, can you take a look when you have time? Thanks!

JingsongLi · 2026-08-30T01:58:59Z

@zhongyujiang The current manifest sort is already extremely fast; you might want to test it against your internal large tables.

JingsongLi · 2026-09-14T09:43:59Z

Take a look to #9802 , If it doesn't meet your needs, submit a new one.

zhongyujiang changed the title ~~[core][spark]: support rewrite_manifest procedure~~ [draft][core][spark]: support rewrite_manifest procedure Aug 12, 2026

zhongyujiang marked this pull request as draft August 12, 2026 12:51

zhongyujiang changed the title ~~[draft][core][spark]: support rewrite_manifest procedure~~ [core][spark]: support rewrite_manifest procedure Aug 12, 2026

zhongyujiang force-pushed the rewrite-manifest branch from 2b33320 to 960fbc2 Compare August 16, 2026 14:01

zhongyujiang marked this pull request as ready for review August 17, 2026 02:50

zhongyujiang commented Aug 17, 2026

View reviewed changes

JingsongLi requested changes Aug 20, 2026

View reviewed changes

zhongyujiang and others added 7 commits August 24, 2026 09:43

[spark]: fix test assertion for corrected output count semantics

1a7abda

Update the assertion to match the corrected semantics: rewritten_manifests_count == input manifest count, added_manifests_count == output manifest count (> 0). Co-Authored-By: Claude <noreply@anthropic.com>

zhongyujiang force-pushed the rewrite-manifest branch from c1416aa to ea88bc6 Compare August 24, 2026 01:45

JingsongLi closed this Sep 14, 2026

Navigation Menu

Repository navigation

[core][spark]: support rewrite_manifest procedure - #9195

[core][spark]: support rewrite_manifest procedure#9195
zhongyujiang wants to merge 7 commits into
apache:masterfrom
zhongyujiang:rewrite-manifest

zhongyujiang commented Aug 12, 2026 •

edited

Loading

Uh oh!

zhongyujiang commented Aug 17, 2026

Uh oh!

zhongyujiang Aug 17, 2026

Uh oh!

JingsongLi left a comment

Uh oh!

JingsongLi Aug 20, 2026

Uh oh!

zhongyujiang Aug 23, 2026

Uh oh!

JingsongLi left a comment

Uh oh!

zhongyujiang commented Aug 23, 2026

Uh oh!

zhongyujiang commented Aug 26, 2026

Uh oh!

JingsongLi commented Aug 30, 2026

Uh oh!

JingsongLi commented Sep 14, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

zhongyujiang commented Aug 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Usage

Implementation

Tests

Uh oh!

zhongyujiang commented Aug 17, 2026

Uh oh!

zhongyujiang Aug 17, 2026

Choose a reason for hiding this comment

Uh oh!

JingsongLi left a comment

Choose a reason for hiding this comment

Uh oh!

JingsongLi Aug 20, 2026

Choose a reason for hiding this comment

Uh oh!

zhongyujiang Aug 23, 2026

Choose a reason for hiding this comment

Uh oh!

JingsongLi left a comment

Choose a reason for hiding this comment

Uh oh!

zhongyujiang commented Aug 23, 2026

Uh oh!

zhongyujiang commented Aug 26, 2026

Uh oh!

JingsongLi commented Aug 30, 2026

Uh oh!

JingsongLi commented Sep 14, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

zhongyujiang commented Aug 12, 2026 •

edited

Loading