[WIP][spark] Support ANALYZE TABLE on catalog-managed format tables - #9298
Open
sundapeng wants to merge 4 commits into
Open
[WIP][spark] Support ANALYZE TABLE on catalog-managed format tables#9298sundapeng wants to merge 4 commits into
sundapeng wants to merge 4 commits into
Conversation
sundapeng
changed the title
sundapeng
changed the title
sundapeng
force-pushed
the
upstream/s6-analyze-format-table-partitions
branch
6 times, most recently
from
August 19, 2026 02:14
f3d5eb5 to
86101ab
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.
Spark rejects ANALYZE TABLE for every v2 table, so a user reaching for the standard command on a format table gets told it is unsupported. It is the command that exists to recompute statistics, and after this it does. What it means here is narrower than for a Paimon table, and the difference is worth stating. A format table has no snapshot to carry table-level statistics and no column statistics to compute; analyzing it measures its partitions. So both modifiers carry weight: NOSCAN stops at the listing, and a full ANALYZE also reads file footers for exact row counts — which no listing can give and no commit can give for data it did not write. PARTITION(...) must name a leading run of the partition keys, because that is the shape the catalog can select on. Anything else is rejected rather than quietly widened to more partitions than the user asked about; naming a partition that does not exist is rejected too, rather than reporting success for having measured nothing. Following Spark, a partition column named without a value means every value of it, and the column names are resolved case-insensitively like every other command's. A format table that discovers its partitions from the filesystem has no catalog to write to and keeps the upstream rejection. Measuring replaces rather than accumulates, so running it twice is running it once. Row counts stay unknown for formats without a footer, and a footer that cannot be read makes the whole partition's row count unknown rather than short. Tests: CatalogManagedPartitionAnalyzeTest covers the command end to end — a full spec measuring only its own partition and a leading prefix measuring every partition under it, a non-leading column and a column that is not a partition column both rejected rather than silently widened, a partition that is not registered failing instead of reporting success for having measured nothing, column names resolved the way the rest of Spark resolves them, NOSCAN keeping a row count that is already known while a full ANALYZE reads the one a NOSCAN cannot, a second run reporting the same measurement rather than accumulating, files a committer left staged not being counted, the registered set never changing, and a filesystem-discovering format table keeping the upstream rejection.
sundapeng
force-pushed
the
upstream/s6-analyze-format-table-partitions
branch
from
August 19, 2026 02:40
86101ab to
651c754
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
Spark rejects ANALYZE TABLE for every v2 table, so the standard command does not work on a format
table. This PR makes it work on catalog-managed ones.
A format table has no snapshot to carry table-level statistics and no column statistics to compute,
so analyzing it measures its partitions. NOSCAN stops at the listing; a full ANALYZE also reads file
footers for exact row counts.
PARTITION(...)must name a leading run of the partition keys, because that is what the catalog canselect on. Any other shape is rejected rather than widened to more partitions than the user asked
for, and so is a partition that is not registered. A partition column named without a value means
every value of it, and column names are resolved case-insensitively, both following Spark.
Measuring replaces rather than accumulates, so running it twice is the same as running it once. Row
counts stay unknown for formats without a footer, and an unreadable footer makes the whole
partition's row count unknown rather than short. A format table that discovers partitions from the
filesystem has no catalog to write to and keeps the upstream rejection.
Tests
CatalogManagedPartitionAnalyzeTest: a full spec and a leading prefix, a non-leading column and anon-partition column both rejected, an unregistered partition failing, case-insensitive column
names, NOSCAN against a full ANALYZE, a second run not accumulating, staged files not counted, the
registered set unchanged, and a filesystem-discovering table keeping the rejection.
API and Format
No API or format change. A command Spark used to reject now runs on catalog-managed format tables.