[WIP][core][spark] Measure format table partitions in MSCK REPAIR TABLE - #9297
Open
sundapeng wants to merge 3 commits into
Open
[WIP][core][spark] Measure format table partitions in MSCK REPAIR TABLE#9297sundapeng wants to merge 3 commits into
sundapeng wants to merge 3 commits into
Conversation
sundapeng
changed the title
sundapeng
changed the title
sundapeng
force-pushed
the
upstream/s5-measure-partitions-in-msck
branch
6 times, most recently
from
August 19, 2026 02:14
36c77c0 to
e3ccd68
Compare
A catalog-managed format table registers the partitions a commit wrote. It has no snapshot, so the channel data tables use to report statistics — commitSnapshot — does not exist for it, and there is nowhere else for the numbers to go. This adds them to the registration itself. CreatePartitionsRequest gains an optional partitionStatistics list and a statisticsMode. Registration and statistics then land in one request and one server-side transaction, which is the point: "the statistics failed but the partition registered" is not a state anyone has to handle, and no extra round trip is paid for the numbers. The alternative, a separate endpoint, buys only the ability to report statistics without registering — which is exactly the state worth not having. Rather than grow a third overload on each layer, the reporting arguments fold into the ignoreIfExists overloads that apache#8707 added and nothing yet calls. Catalog keeps the long-standing two-argument createPartitions plus the reporting one, and RESTApi is back to a single method; CachingCatalog, DelegateCatalog and RESTCatalog each lose a forwarder, and two Spark tests that registered partitions through the removed overload now call the two-argument one. No released signature changes. While here, supportsPartitionModification says in its javadoc that createPartitions and alterPartitions do nothing when it is false, which stopped being true once a REST catalog started registering the partitions of a Format Table while still reporting false. What the flag actually gates is the PartitionModification handler a table is given, so committing a table maintains no partitions; calling the catalog methods directly is unaffected. The javadoc now says that. The mode says how a report combines with what the server holds, per field: ADD sums the three counts and takes the later creation time, SET replaces. Which one a writer sends follows from what it accounted for: an appending commit (INSERT INTO) reports ADD, since it knows the files it wrote and not the ones the partition already held, and one statement whose writers commit separately reports each writer's share; an overwriting commit (INSERT OVERWRITE, the partition pinned statically or left to the data) reports SET, since it left the partition holding what it wrote and nothing else. Reporting a field as unknown leaves the stored value alone under either mode, so a reporter that can measure only some fields never erases the others. Statistics never create or remove a partition row, whatever the numbers say. The mode travels as a string rather than an enum so a value the server has no name for deserializes instead of failing the request it arrived in: such a report is dropped and the partitions are still registered. Compatibility runs both ways and neither direction errors. A new client against an old server: the field is ignored, the statistics are dropped and the columns stay unknown. The client is not told, because the response carries no acknowledgement — an observation that is missing is a thing consumers can handle, an observation that is wrong is not, and an acknowledgement would be a second contract to keep in step with the first. An old client against a new server: no statistics arrive and the columns stay unknown. The existing three-argument createPartitions keeps working and keeps sending exactly the request it sends today, so a caller that reports nothing does not change the shape the server sees. A non-empty ADD report declares itself unsafe to retry, so an automatically replayed POST cannot count the same increment twice. Tests: HttpClientRetrySafetyTest covers which requests declare themselves unsafe to retry — only a non-empty ADD report does — and that the flag itself never reaches the wire. RESTApiJsonTest covers the request shape, including a mode the server has no name for surviving deserialization. MockRESTCatalogTest covers reporting against the mock server end to end: ADD accumulating, SET replacing, unknown fields leaving the stored value alone, statistics for a partition the request does not register being dropped, a report that only partly matches not being applied at all, and an unreadable mode still registering. Its assertions are the contract the modes are described by. CatalogFormatTablePartitionManagerTest covers what the manager sends: statistics riding in the request of their own partitions, a batch that reports nothing sending an empty list rather than null, and the rejections of malformed reports. CachingCatalogTest and DelegateCatalogTest cover the new overload forwarding through and invalidating the partition cache.
The writer already counted the rows and the bytes, the commit already knows which partitions it wrote, and the catalog now has somewhere to put both. This connects them, behind format-table.commit.report-partition-statistics. Which mode a commit uses follows from what it did to the partition. An appending commit saw only its own files, so it reports an increment: a Flink sink commits once per writer subtask, and N increments over one partition add up to what the job wrote. An overwriting commit replaced everything the partitions held, so what it wrote is the total and it reports that. Static prefix overwrite is why a pure increment cannot express this. Clearing a prefix empties every partition beneath it, including ones this commit writes nothing to; their old data is gone and no increment says so. Those partitions report zero — an exact zero, they really are empty — and stay registered. The partition set is the catalog's authority and statistics never touch it; only ADD PARTITION and DROP PARTITION do. The directories emptied that way come out of the deletion listing, which already had to walk them, so the numbers cost no extra IO: the row count and byte size come from the writer, the file count from the commit messages, and the last file creation time from the commit's own clock. The option is off by default. A non-empty ADD report makes the create-partitions request unsafe to replay, so a 429 or 503 the client used to retry by itself now surfaces to the caller instead of risking a double-counted increment. That is a change to how an existing write path fails, and it should be opted into rather than inherited. An increment can also drift for reasons no client can see: a writer that is not Paimon, a file deleted out of band. Convergence is a later full report over the same partition, which a follow-up will teach MSCK REPAIR and ANALYZE to produce; a deduplication token in front of every commit would cost more than the drift does. Tests: FormatTableCommitStatisticsTest covers append, dynamic overwrite, static prefix overwrite of a partition this commit does not write, the summation of the independent increments of concurrent writers of one partition, a listing that answers under another scheme, a nested directory in the value-only layout, and reporting turned off.
A commit reports what it wrote. Nothing reports what is already there, and for a format table plenty is: partitions written by something that is not Paimon, files deleted out of band, an increment redelivered and counted twice. MSCK REPAIR TABLE is already the command that reconciles the partition set against the directories, so it is the natural place to reconcile the numbers too. Off by default, behind spark.paimon.format-table.repair.collect-statistics, because measuring changes what a repair costs: the plain diff lists partition directories, and measuring lists the files inside every one of them. That is a different order of magnitude on a table with many partitions, and a repair should not silently become that. When it is on, spark.paimon.format-table.statistics.parallelism caps how many partitions are measured at once, at 8: listing one is a round trip the driver spends waiting on, and the cap keeps a table with many partitions from turning that wait into a burst of requests. When on it measures every partition that ends up registered with a directory behind it, not only the ones it just added — the stale numbers of partitions written outside Paimon are exactly what a repair exists to correct. Without ADD it stays inside the already-registered set, so measuring never registers a partition the command was not asked to. The collector reports what a reader would see. File count, byte size and last file creation time come from the listing. It stops there: the row count needs a file footer, which is what ANALYZE is for. A listing failure aborts the whole collection rather than reporting what it managed to see, because a truncated listing is indistinguishable from a partition that lost files. A partition whose directory is gone measures as an exact zero, with no last file to date. Tests: FormatTablePartitionStatsCollectorTest covers the staging trees a committer leaves behind, exact parquet row counts, an unreadable footer, a missing directory, the one-for-one alignment of the result with the given specs, a spec that omits a partition key, and a listing failure aborting the whole collection on both the serial and the parallel path. FormatTablePartitionRepairTest covers measuring every partition on disk rather than only the additions, a repair without ADD registering nothing, and a listing failure leaving the catalog untouched. CatalogManagedPartitionMsckRepairTest covers the command end to end with the option off and on.
sundapeng
force-pushed
the
upstream/s5-measure-partitions-in-msck
branch
from
August 19, 2026 02:40
e3ccd68 to
42f3394
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
A commit reports only what it wrote. A format table also holds partitions written outside Paimon,
files deleted out of band, and increments that drifted. MSCK REPAIR TABLE already reconciles the
partition set against the directories, so it is the place to reconcile the numbers too.
Off by default, behind
spark.paimon.format-table.repair.collect-statistics: a plain repair listspartition directories, while measuring lists the files inside every one of them.
spark.paimon.format-table.statistics.parallelismcaps how many partitions are measured at once, at8.
When on, it measures every registered partition that has a directory, not only the ones it just
added, and never registers a partition the repair did not ask for. It reports what a listing sees:
file count, byte size and last file creation time. Row count needs a file footer, which is what
ANALYZE is for. A partition whose directory is gone measures as an exact zero, and a listing failure
aborts the collection, because a truncated listing looks the same as a partition that lost files.
Tests
FormatTablePartitionStatsCollectorTest: staging trees, parquet row counts, an unreadable footer,a missing directory, alignment with the given specs, a spec missing a partition key, and a listing
failure aborting both the serial and the parallel path.
FormatTablePartitionRepairTest: measuring every partition on disk, a repair without ADDregistering nothing, and a listing failure leaving the catalog untouched.
CatalogManagedPartitionMsckRepairTest: the command end to end, with the option off and on.API and Format
Two new Spark options. No API or format change.