[core][spark] Maintain num files and total file size on snapshots

[zhuxiangyi]

Purpose

Paimon cannot answer "how many data files does this table have, and how large are they" without folding every manifest entry.

Two places pay for it today:

• ANALYZE TABLE calls listPartitionEntries() and sums fileSizeInBytes over the whole manifest set purely to obtain one number, then uses it to estimate the merged size as totalSize * mergedRecordCount / totalRecordCount (PaimonAnalyzeTableColumnCommand.scala:62-73).
• Append compaction planning calls partitionEntries() on every discovery round — AppendTableCompact.java:115, CombinedAppendCompactSource.java:181, MultiTableAppendCompactReadOperator.java:110 — which in combined mode is one full manifest scan per table per round.

The commit path already computes this information and then throws it away. FileStoreCommitImpl builds PartitionEntry.merge(deltaFiles) on every commit, whose counts are already signed by file kind, hands the result to the catalog as partition statistics, and drops it for every catalog that does not implement partition reporting — AbstractCatalog.commitSnapshot throws UnsupportedOperationException, and RenamingSnapshotCommit.commit ignores the argument.

Summing those per-partition entries gives the table level delta, so a snapshot can carry the running totals:

• numFiles — live data files
• totalFileSizeInBytes — their total size

Both are derived as previous + delta. This adds no IO and no second pass over the delta entries; it reuses a fold that already runs.

Deriving the totals costs nothing when it works and is skipped when it does not, so the values are advisory. null means unknown, never zero, and readers fall back to scanning — PaimonAnalyzeTableColumnCommand shows the intended shape:

val totalSize = Option(currentSnapshot.totalFileSizeInBytes())
  .map(_.longValue())
  .getOrElse(table.newScan().listPartitionEntries().asScala.map(_.fileSizeInBytes()).sum)

Unknown arises for snapshots written before the fields existed, and for any commit whose predecessor was unknown. Two paths recompute instead of propagating it:

• replaceManifestList replaces the manifest layout wholesale and its callers may drop files with it — RemoveUnexistingManifestsAction does exactly that — so inheriting the previous counters would publish a number that no longer matches the live file set. It folds the manifests it is about to commit instead.
• Manifest compaction seeds unknown counters from the compacted manifest set. It rewrites the whole manifest set anyway and the compacted form is the cheapest one to fold, which makes compact_manifest the way to recover the counters on an upgraded table; the incremental chain continues on its own afterwards. To keep that reliable the procedure commits a snapshot when the counters are unknown even if no manifests need merging, and returns early exactly as before once they are known.

The two-field, fold-at-commit, advisory-and-degradable shape follows Delta Lake's version checksum file, which maintains the equivalent counters per commit and falls back rather than failing when it cannot.

Benefits

Delivered here:

• ANALYZE TABLE no longer scans the manifest set for the table size, and the value it feeds into the merged-size calculation is exact rather than a scan result summed on the spot.
• sys.snapshots gains num_files and total_file_size_in_bytes, so "which commit blew up the file count" is answerable by query, for any retained snapshot, without starting a job and without scanning manifests. Commit metrics cannot answer this: they are per-commit deltas held in a running writer's memory, carry no snapshot identity, and are absent entirely for tables written by Spark batch jobs or procedures.

Enabled next, not part of this PR:

• Compaction planning and cost-based optimization can read the counters instead of scanning, with the fallback above keeping them correct on tables that have not been seeded.
• The same fold extends naturally to a file size histogram and to deletion vector counts, which would let compaction decisions move from a file count threshold to a distribution.

Compatibility

Existing tables are unaffected. The fields are nullable and serialized only when present, so snapshots written before this change stay byte-identical, and the constructors that predate the fields are kept so no caller needs to change. Reads, writes, compaction, overwrite, tags, rollback, snapshot expiration and the sys.snapshots query all keep working on such tables, with the two columns reported as NULL.

Tests

SnapshotFileStatsTest, 20 cases:

• Five assert the values against a full manifest scan across appends, primary key compaction, overwrite, tag and rollback, and manifest compaction.
• Eleven cover tables rewritten into the pre-change snapshot format — read, write, primary key compaction, overwrite, tag, rollback, snapshot expiration, the sys.snapshots columns, mixed old/new history with time travel, and unknown JSON fields. The central one asserts that writing on top of an unknown baseline keeps reporting unknown rather than restarting from zero, which is the failure mode that would publish a plausible but wrong file count.
• The rest cover the recompute paths, including one that drops a manifest and must follow the shrink instead of reporting the stale count, and seeding a legacy table whose manifests need no merging at all.

RemoveUnexistingManifestsActionITCase and CompactManifestProcedureITCase exercise the same paths through Flink.

Full run: 293 paimon-core cases and 9 paimon-flink cases, no failures. paimon-spark-common compiles.

Documentation

docs/docs/concepts/system-tables.mdx documents the two columns, what NULL means, how it propagates, and that compact_manifest recovers the values.

[JingsongLi]

-1

This properties already existing in REST Server and REST API.

[zhuxiangyi]

Thanks for the review. You are right — the REST catalog already receives these numbers on every commit via CommitTableRequest's partition statistics and exposes them through the partition API, so maintaining them again on the snapshot is redundant. I missed that when writing this up. Closing.