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

["zhuxiangyi (via GitHub)" <gi...@apache.org>]

zhuxiangyi opened a new pull request, #9659:
URL: https://github.com/apache/paimon/pull/9659

   ### 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:
   
   ```scala
   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.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [core][spark] Maintain num files and total file size on snapshots [paimon]

["zhuxiangyi (via GitHub)" <gi...@apache.org>]

zhuxiangyi commented on PR #9659:
URL: https://github.com/apache/paimon/pull/9659#issuecomment-5582614889

   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.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [core][spark] Maintain num files and total file size on snapshots [paimon]

["zhuxiangyi (via GitHub)" <gi...@apache.org>]

zhuxiangyi closed pull request #9659: [core][spark] Maintain num files and total file size on snapshots
URL: https://github.com/apache/paimon/pull/9659


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org