Skip to content

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

Open
zhongyujiang wants to merge 6 commits into
apache:masterfrom
zhongyujiang:rewrite-manifest
Open

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

Conversation

@zhongyujiang

@zhongyujiang zhongyujiang commented Aug 12, 2026

Copy link
Copy Markdown
Contributor

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 zhongyujiang changed the title [core][spark]: support rewrite_manifest procedure [draft][core][spark]: support rewrite_manifest procedure Aug 12, 2026
@zhongyujiang
zhongyujiang marked this pull request as draft August 12, 2026 12:51
@zhongyujiang zhongyujiang changed the title [draft][core][spark]: support rewrite_manifest procedure [core][spark]: support rewrite_manifest procedure Aug 12, 2026
zhongyujiang and others added 4 commits August 16, 2026 21:57
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>
zhongyujiang and others added 2 commits August 16, 2026 22:55
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>
@zhongyujiang
zhongyujiang marked this pull request as ready for review August 17, 2026 02:50
@zhongyujiang

Copy link
Copy Markdown
Contributor Author

@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.

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);

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant