[core][spark]: support rewrite_manifest procedure

[zhongyujiang]

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