From b59b9e5a4f7ccfeec26ca92689b5986a1ed89cd8 Mon Sep 17 00:00:00 2001 From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com> Date: Sun, 6 Sep 2026 14:52:46 +0800 Subject: [PATCH 1/3] [core][spark] Maintain num files and total file size on snapshots Answering "how many files does this table have, and how large are they" currently requires folding every manifest entry: ANALYZE TABLE scans the whole manifest set just to sum one number, and append compaction planning repeats that scan on every discovery round. The commit path already folds the same information. It builds per-partition entries whose counts are signed by file kind, hands them to the catalog as partition statistics, and then discards them for every catalog that does not implement partition reporting. Summing those entries gives the table level delta for free, so a snapshot can carry the running totals: numFiles live data files totalFileSizeInBytes their total size Both are derived as previous + delta, at no extra IO and without walking the delta entries a second time. Deriving them costs nothing when it works and is skipped when it does not, so they are advisory: null means unknown, never zero, and readers fall back to scanning. PaimonAnalyzeTableColumnCommand shows the shape - use the snapshot value, otherwise scan. Unknown propagates in three cases: snapshots written before these fields existed, snapshots whose predecessor was unknown, and replaceManifestList, which swaps the manifest layout wholesale and therefore cannot carry counters over. RemoveUnexistingManifestsAction goes through that path and does drop files, so inheriting the previous numbers there would publish a count that does not match the live file set. Old tables are unaffected. The fields are nullable and serialized only when present, so snapshots written before this change are byte-identical, and the constructors that predate the fields are kept. Reads, writes, compaction, overwrite, tags, rollback, snapshot expiration and the snapshots system table all keep working on such tables, with the two columns reported as NULL. --- docs/docs/concepts/system-tables.mdx | 12 + .../main/java/org/apache/paimon/Snapshot.java | 166 ++++- .../java/org/apache/paimon/Changelog.java | 12 +- .../paimon/operation/FileStoreCommitImpl.java | 43 +- .../paimon/table/system/SnapshotsTable.java | 8 +- .../main/java/org/apache/paimon/tag/Tag.java | 12 +- .../paimon/table/SnapshotFileStatsTest.java | 627 ++++++++++++++++++ .../PaimonAnalyzeTableColumnCommand.scala | 18 +- 8 files changed, 878 insertions(+), 20 deletions(-) create mode 100644 paimon-core/src/test/java/org/apache/paimon/table/SnapshotFileStatsTest.java diff --git a/docs/docs/concepts/system-tables.mdx b/docs/docs/concepts/system-tables.mdx index 0bb6d2629cc4..fd952d46a85e 100644 --- a/docs/docs/concepts/system-tables.mdx +++ b/docs/docs/concepts/system-tables.mdx @@ -70,6 +70,18 @@ logical rows to a table with one [dedicated BLOB column](../multimodal-table/blo adds `N` records to the regular data files and `N` records to the BLOB files, so `delta_record_count` increases by `2 * N`. Use `COUNT(*)` when you need the logical row count. +`num_files` and `total_file_size_in_bytes` describe the live data files of the snapshot. They are +folded incrementally at commit time from the files the commit adds and removes, so reading them +costs nothing extra, and they cover data files only: changelog files, index files and deletion +vectors are not counted. + +Both columns are nullable, and `NULL` means unknown rather than zero. It appears for snapshots +written before these columns existed, and for commits that could not derive the values from the +previous snapshot, such as metadata repair operations that replace the manifest layout wholesale. +Once a snapshot reports unknown, the snapshots committed after it report unknown as well, until the +values are recomputed. Fall back to scanning the manifests when you read `NULL`; never read it as +zero. + ### Schemas Table You can query the historical schemas of the table through schemas table. diff --git a/paimon-api/src/main/java/org/apache/paimon/Snapshot.java b/paimon-api/src/main/java/org/apache/paimon/Snapshot.java index d9de932506cd..3ae7cb4bd5f2 100644 --- a/paimon-api/src/main/java/org/apache/paimon/Snapshot.java +++ b/paimon-api/src/main/java/org/apache/paimon/Snapshot.java @@ -73,6 +73,8 @@ public class Snapshot implements Serializable { protected static final String FIELD_PROPERTIES = "properties"; protected static final String FIELD_NEXT_ROW_ID = "nextRowId"; protected static final String FIELD_OPERATION = "operation"; + protected static final String FIELD_NUM_FILES = "numFiles"; + protected static final String FIELD_TOTAL_FILE_SIZE_IN_BYTES = "totalFileSizeInBytes"; // version of snapshot @JsonProperty(FIELD_VERSION) @@ -203,6 +205,19 @@ public class Snapshot implements Serializable { @Nullable protected final Operation operation; + // number of live data files in this snapshot, null when it could not be derived incrementally + @JsonInclude(JsonInclude.Include.NON_NULL) + @JsonProperty(FIELD_NUM_FILES) + @Nullable + protected final Long numFiles; + + // total size of live data files in this snapshot, null when it could not be derived + // incrementally + @JsonInclude(JsonInclude.Include.NON_NULL) + @JsonProperty(FIELD_TOTAL_FILE_SIZE_IN_BYTES) + @Nullable + protected final Long totalFileSizeInBytes; + public Snapshot( long id, long schemaId, @@ -226,6 +241,58 @@ public Snapshot( @Nullable Map properties, @Nullable Long nextRowId, @Nullable Operation operation) { + this( + id, + schemaId, + baseManifestList, + baseManifestListSize, + deltaManifestList, + deltaManifestListSize, + changelogManifestList, + changelogManifestListSize, + indexManifest, + commitUser, + writerVersion, + commitIdentifier, + commitKind, + timeMillis, + totalRecordCount, + deltaRecordCount, + changelogRecordCount, + watermark, + statistics, + properties, + nextRowId, + operation, + null, + null); + } + + public Snapshot( + long id, + long schemaId, + String baseManifestList, + @Nullable Long baseManifestListSize, + String deltaManifestList, + @Nullable Long deltaManifestListSize, + @Nullable String changelogManifestList, + @Nullable Long changelogManifestListSize, + @Nullable String indexManifest, + String commitUser, + @Nullable String writerVersion, + long commitIdentifier, + CommitKind commitKind, + long timeMillis, + long totalRecordCount, + long deltaRecordCount, + @Nullable Long changelogRecordCount, + @Nullable Long watermark, + @Nullable String statistics, + @Nullable Map properties, + @Nullable Long nextRowId, + @Nullable Operation operation, + @Nullable Long numFiles, + @Nullable Long totalFileSizeInBytes) { this( CURRENT_VERSION, UUID.randomUUID().toString(), @@ -250,7 +317,67 @@ public Snapshot( statistics, properties, nextRowId, - operation); + operation, + numFiles, + totalFileSizeInBytes); + } + + /** + * Kept so that callers written before {@link #numFiles()} and {@link #totalFileSizeInBytes()} + * existed keep compiling; both are left unknown. + */ + public Snapshot( + int version, + @Nullable String uuid, + long id, + long schemaId, + String baseManifestList, + @Nullable Long baseManifestListSize, + String deltaManifestList, + @Nullable Long deltaManifestListSize, + @Nullable String changelogManifestList, + @Nullable Long changelogManifestListSize, + @Nullable String indexManifest, + String commitUser, + @Nullable String writerVersion, + long commitIdentifier, + CommitKind commitKind, + long timeMillis, + long totalRecordCount, + long deltaRecordCount, + @Nullable Long changelogRecordCount, + @Nullable Long watermark, + @Nullable String statistics, + @Nullable Map properties, + @Nullable Long nextRowId, + @Nullable Operation operation) { + this( + version, + uuid, + id, + schemaId, + baseManifestList, + baseManifestListSize, + deltaManifestList, + deltaManifestListSize, + changelogManifestList, + changelogManifestListSize, + indexManifest, + commitUser, + writerVersion, + commitIdentifier, + commitKind, + timeMillis, + totalRecordCount, + deltaRecordCount, + changelogRecordCount, + watermark, + statistics, + properties, + nextRowId, + operation, + null, + null); } @JsonCreator @@ -279,7 +406,9 @@ public Snapshot( @JsonProperty(FIELD_STATISTICS) @Nullable String statistics, @JsonProperty(FIELD_PROPERTIES) @Nullable Map properties, @JsonProperty(FIELD_NEXT_ROW_ID) @Nullable Long nextRowId, - @JsonProperty(FIELD_OPERATION) @Nullable Operation operation) { + @JsonProperty(FIELD_OPERATION) @Nullable Operation operation, + @JsonProperty(FIELD_NUM_FILES) @Nullable Long numFiles, + @JsonProperty(FIELD_TOTAL_FILE_SIZE_IN_BYTES) @Nullable Long totalFileSizeInBytes) { this.version = version; this.uuid = uuid; this.id = id; @@ -304,6 +433,8 @@ public Snapshot( this.properties = properties; this.nextRowId = nextRowId; this.operation = operation; + this.numFiles = numFiles; + this.totalFileSizeInBytes = totalFileSizeInBytes; } @JsonGetter(FIELD_VERSION) @@ -439,6 +570,29 @@ public Operation operation() { return operation; } + /** + * Number of live data files in this snapshot, maintained incrementally at commit time. + * + *

Returns null when the value is unknown: snapshots written before this field existed, and + * commits whose previous snapshot had no value to derive from. Callers must fall back to + * scanning manifests instead of treating null as zero. + */ + @JsonGetter(FIELD_NUM_FILES) + @Nullable + public Long numFiles() { + return numFiles; + } + + /** + * Total size in bytes of the live data files in this snapshot, maintained incrementally at + * commit time. Null has the same meaning as in {@link #numFiles()}. + */ + @JsonGetter(FIELD_TOTAL_FILE_SIZE_IN_BYTES) + @Nullable + public Long totalFileSizeInBytes() { + return totalFileSizeInBytes; + } + public String toJson() { return JsonSerdeUtil.toJson(this); } @@ -469,7 +623,9 @@ public int hashCode() { statistics, properties, nextRowId, - operation); + operation, + numFiles, + totalFileSizeInBytes); } @Override @@ -504,7 +660,9 @@ public boolean equals(Object o) { && Objects.equals(statistics, that.statistics) && Objects.equals(properties, that.properties) && Objects.equals(nextRowId, that.nextRowId) - && operation == that.operation; + && operation == that.operation + && Objects.equals(numFiles, that.numFiles) + && Objects.equals(totalFileSizeInBytes, that.totalFileSizeInBytes); } /** Type of changes in this snapshot. */ diff --git a/paimon-core/src/main/java/org/apache/paimon/Changelog.java b/paimon-core/src/main/java/org/apache/paimon/Changelog.java index 4433c806195b..dd6af336af4e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/Changelog.java +++ b/paimon-core/src/main/java/org/apache/paimon/Changelog.java @@ -66,7 +66,9 @@ public Changelog(Snapshot snapshot) { snapshot.statistics(), snapshot.properties, snapshot.nextRowId, - snapshot.operation); + snapshot.operation, + snapshot.numFiles, + snapshot.totalFileSizeInBytes); } @JsonCreator @@ -95,7 +97,9 @@ public Changelog( @JsonProperty(FIELD_STATISTICS) @Nullable String statistics, @JsonProperty(FIELD_PROPERTIES) Map properties, @JsonProperty(FIELD_NEXT_ROW_ID) @Nullable Long nextRowId, - @JsonProperty(FIELD_OPERATION) @Nullable Operation operation) { + @JsonProperty(FIELD_OPERATION) @Nullable Operation operation, + @JsonProperty(FIELD_NUM_FILES) @Nullable Long numFiles, + @JsonProperty(FIELD_TOTAL_FILE_SIZE_IN_BYTES) @Nullable Long totalFileSizeInBytes) { super( version, uuid, @@ -120,7 +124,9 @@ public Changelog( statistics, properties, nextRowId, - operation); + operation, + numFiles, + totalFileSizeInBytes); } public static Changelog fromJson(String json) { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index 2d9c94ec72fc..d2d1b23e12af 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -1107,9 +1107,16 @@ CommitResult tryCommitOnce( long nextRowIdStart = firstRowIdStart; try { long previousTotalRecordCount = 0L; + // A brand new table starts from an empty file set. An existing snapshot that carries no + // file statistics (written before the fields existed, or by a commit that could not + // derive them) breaks the chain, and the new snapshot stays unknown as well. + Long previousNumFiles = 0L; + Long previousTotalFileSizeInBytes = 0L; Long currentWatermark = watermark; if (latestSnapshot != null) { previousTotalRecordCount = latestSnapshot.totalRecordCount(); + previousNumFiles = latestSnapshot.numFiles(); + previousTotalFileSizeInBytes = latestSnapshot.totalFileSizeInBytes(); // read all previous manifest files mergeBeforeManifests = manifestList.readDataManifests(latestSnapshot); Long latestWatermark = latestSnapshot.watermark(); @@ -1127,6 +1134,9 @@ CommitResult tryCommitOnce( mergeBeforeManifests = emptyList(); mergeAfterManifests = emptyList(); oldIndexManifest = null; + // the previous file set is dropped entirely, so the counters restart from zero + previousNumFiles = 0L; + previousTotalFileSizeInBytes = 0L; } else { ManifestMergeReuse manifestMergeReuse = tryReuseManifestMergeResult(retryResult, mergeBeforeManifests); @@ -1174,6 +1184,21 @@ CommitResult tryCommitOnce( // write new delta files into manifest files deltaPartitionEntries = new ArrayList<>(PartitionEntry.merge(deltaFiles)); + + // reuse the per-partition fold above: its counts are already signed by file kind, so + // summing them gives the table level delta without walking deltaFiles again + long deltaNumFiles = 0L; + long deltaFileSizeInBytes = 0L; + for (PartitionEntry entry : deltaPartitionEntries) { + deltaNumFiles += entry.fileCount(); + deltaFileSizeInBytes += entry.fileSizeInBytes(); + } + Long numFiles = previousNumFiles == null ? null : previousNumFiles + deltaNumFiles; + Long totalFileSizeInBytes = + previousTotalFileSizeInBytes == null + ? null + : previousTotalFileSizeInBytes + deltaFileSizeInBytes; + deltaManifestList = manifestList.write(manifestFile.write(deltaFiles)); // write changelog into manifest files @@ -1241,7 +1266,9 @@ CommitResult tryCommitOnce( // if empty properties, just set to null properties.isEmpty() ? null : properties, nextRowIdStart, - operation); + operation, + numFiles, + totalFileSizeInBytes); } catch (Throwable e) { // fails when preparing for commit, we should clean up commitCleaner.cleanUpReuseTmpManifests( @@ -1409,6 +1436,10 @@ public boolean replaceManifestList( // if empty properties, just set to null latest.properties(), nextRowId, + null, + // the manifest layout is rewritten wholesale here, so the incremental file + // statistics cannot be carried over; leave them unknown + null, null); return commitSnapshotImpl(latest, newSnapshot, emptyList()); @@ -1495,7 +1526,10 @@ public boolean rollbackToAsLatest(Snapshot targetSnapshot) { targetSnapshot.statistics(), targetSnapshot.properties(), nextRowId, - null); + null, + // the live file set is restored to the target snapshot's + targetSnapshot.numFiles(), + targetSnapshot.totalFileSizeInBytes()); // The rollback is an overwrite from the previous latest to the target, so the base files, // delta files and index changes describe the transition the callbacks need. These are @@ -1644,7 +1678,10 @@ private boolean compactManifestOnce() { latestSnapshot.statistics(), latestSnapshot.properties(), latestSnapshot.nextRowId(), - null); + null, + // manifest compaction only rewrites metadata, the data files are untouched + latestSnapshot.numFiles(), + latestSnapshot.totalFileSizeInBytes()); return commitSnapshotImpl(latestSnapshot, newSnapshot, emptyList()); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java index 75d46b708e1d..75946b190828 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java @@ -111,7 +111,9 @@ public class SnapshotsTable implements ReadonlyTable { new DataField(13, "next_row_id", new BigIntType(true)), new DataField(14, "operation", SerializationUtils.newStringType(true)), new DataField( - 15, "writer_version", SerializationUtils.newStringType(true)))); + 15, "writer_version", SerializationUtils.newStringType(true)), + new DataField(16, "num_files", new BigIntType(true)), + new DataField(17, "total_file_size_in_bytes", new BigIntType(true)))); private final FileIO fileIO; private final Path location; @@ -346,7 +348,9 @@ private InternalRow toRow(Snapshot snapshot) { snapshot.operation() == null ? null : BinaryString.fromString(snapshot.operation().toString()), - BinaryString.fromString(snapshot.writerVersion())); + BinaryString.fromString(snapshot.writerVersion()), + snapshot.numFiles(), + snapshot.totalFileSizeInBytes()); } } } diff --git a/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java b/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java index ac435782d05a..fc38f18cdf95 100644 --- a/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java +++ b/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java @@ -83,6 +83,8 @@ public Tag( @JsonProperty(FIELD_PROPERTIES) Map properties, @JsonProperty(FIELD_NEXT_ROW_ID) @Nullable Long nextRowId, @JsonProperty(FIELD_OPERATION) @Nullable Operation operation, + @JsonProperty(FIELD_NUM_FILES) @Nullable Long numFiles, + @JsonProperty(FIELD_TOTAL_FILE_SIZE_IN_BYTES) @Nullable Long totalFileSizeInBytes, @JsonProperty(FIELD_TAG_CREATE_TIME) @Nullable LocalDateTime tagCreateTime, @JsonProperty(FIELD_TAG_TIME_RETAINED) @Nullable Duration tagTimeRetained) { super( @@ -109,7 +111,9 @@ public Tag( statistics, properties, nextRowId, - operation); + operation, + numFiles, + totalFileSizeInBytes); this.tagCreateTime = tagCreateTime; this.tagTimeRetained = tagTimeRetained; } @@ -151,6 +155,8 @@ public static Tag fromSnapshotAndTagTtl( snapshot.properties(), snapshot.nextRowId(), snapshot.operation(), + snapshot.numFiles(), + snapshot.totalFileSizeInBytes(), tagCreateTime, tagTimeRetained); } @@ -180,7 +186,9 @@ public Snapshot trimToSnapshot() { statistics, properties, nextRowId, - operation); + operation, + numFiles, + totalFileSizeInBytes); } @Override diff --git a/paimon-core/src/test/java/org/apache/paimon/table/SnapshotFileStatsTest.java b/paimon-core/src/test/java/org/apache/paimon/table/SnapshotFileStatsTest.java new file mode 100644 index 000000000000..dbce22dbe1b6 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/SnapshotFileStatsTest.java @@ -0,0 +1,627 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.Snapshot; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.fs.Path; +import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.ManifestList; +import org.apache.paimon.manifest.PartitionEntry; +import org.apache.paimon.operation.FileStoreCommitImpl; +import org.apache.paimon.options.ExpireConfig; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.sink.TableCommitImpl; +import org.apache.paimon.tag.Tag; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.SnapshotManager; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; + +/** + * Tests for {@link Snapshot#numFiles()} and {@link Snapshot#totalFileSizeInBytes()}, which are + * folded incrementally at commit time. + * + *

Two things are checked: the values agree with a full manifest scan whenever they are present, + * and tables whose snapshots predate the fields keep reading and writing exactly as before, with + * the values reported as unknown rather than as a wrong number. + */ +public class SnapshotFileStatsTest extends TableTestBase { + + private static final String commitUser = "test-commit-user"; + + // ------------------------------------------------------------------------------------------ + // the statistics agree with a full manifest scan + // ------------------------------------------------------------------------------------------ + + @Test + public void testAppendTable() throws Exception { + FileStoreTable table = createTable("append_table", false); + + write(table, row(1, "a"), row(2, "b")); + assertStatsMatchManifests(table); + + write(table, row(3, "c")); + assertStatsMatchManifests(table); + + write(table, row(4, "d"), row(5, "e"), row(6, "f")); + assertStatsMatchManifests(table); + + assertThat(latest(table).numFiles()).isGreaterThan(0L); + } + + @Test + public void testPrimaryKeyTableAcrossCompaction() throws Exception { + FileStoreTable table = createTable("pk_table", true); + + // enough commits to trigger compaction, which deletes files as well as adding them + for (int i = 0; i < 12; i++) { + write(table, row(i % 3, "v" + i)); + assertStatsMatchManifests(table); + } + + // guard the premise: without a COMPACT snapshot this test only covers appends + assertThat(commitKindCount(table, Snapshot.CommitKind.COMPACT)).isGreaterThan(0); + } + + @Test + public void testOverwrite() throws Exception { + FileStoreTable table = createTable("overwrite_table", false); + + write(table, row(1, "a"), row(2, "b")); + assertThat(latest(table).numFiles()).isGreaterThan(0L); + + overwrite(table, row(9, "z")); + assertStatsMatchManifests(table); + } + + @Test + public void testStatsSurviveTagAndRollback() throws Exception { + FileStoreTable table = createTable("tag_rollback", false); + + write(table, row(1, "a")); + long targetId = latest(table).id(); + Long targetNumFiles = latest(table).numFiles(); + table.createTag("t1"); + + write(table, row(2, "b")); + assertStatsMatchManifests(table); + + // a tag carries the statistics of the snapshot it was taken from + Tag tag = table.tagManager().get("t1").orElseThrow(IllegalStateException::new); + assertThat(tag.numFiles()).isEqualTo(targetNumFiles); + + // rolling back restores the target snapshot's file set, so its statistics come along + table.rollbackTo(targetId); + table = reload("tag_rollback"); + assertStatsMatchManifests(table); + } + + // ------------------------------------------------------------------------------------------ + // tables written before these fields existed + // ------------------------------------------------------------------------------------------ + + @Test + public void testLegacySnapshotOmitsTheFieldsEntirely() throws Exception { + FileStoreTable table = createTable("legacy_json", false); + write(table, row(1, "a")); + + String json = legacyJsonOf(latest(table)); + assertThat(json).doesNotContain("numFiles").doesNotContain("totalFileSizeInBytes"); + + Snapshot parsed = Snapshot.fromJson(json); + assertThat(parsed.numFiles()).isNull(); + assertThat(parsed.totalFileSizeInBytes()).isNull(); + } + + @Test + public void testReadLegacyTable() throws Exception { + FileStoreTable table = createTable("legacy_read", false); + write(table, row(1, "a"), row(2, "b")); + write(table, row(3, "c")); + List before = readRows(table); + + table = degradeToLegacy("legacy_read"); + + assertThat(latest(table).numFiles()).isNull(); + assertThat(latest(table).totalFileSizeInBytes()).isNull(); + assertThat(readRows(table)).containsExactlyInAnyOrderElementsOf(before); + } + + @Test + public void testWriteToLegacyTableStaysUnknownAndNeverWrong() throws Exception { + FileStoreTable table = createTable("legacy_write", false); + write(table, row(1, "a"), row(2, "b")); + + table = degradeToLegacy("legacy_write"); + + // writing on top of a snapshot without statistics must succeed, and must report unknown + // rather than a count derived from a zero baseline + for (int i = 0; i < 3; i++) { + write(table, row(10 + i, "n" + i)); + Snapshot snapshot = latest(table); + assertThat(snapshot.numFiles()) + .as("snapshot %s must stay unknown once the chain is broken", snapshot.id()) + .isNull(); + assertThat(snapshot.totalFileSizeInBytes()).isNull(); + } + + assertThat(readRows(table)) + .containsExactlyInAnyOrder("1:a", "2:b", "10:n0", "11:n1", "12:n2"); + + // a full scan still answers the question the statistics would have answered + assertThat(scannedNumFiles(table)).isGreaterThan(0L); + } + + @Test + public void testLegacyPrimaryKeyTableCompactsAndReadsCorrectly() throws Exception { + FileStoreTable table = createTable("legacy_pk", true); + for (int i = 0; i < 4; i++) { + write(table, row(i % 2, "v" + i)); + } + + table = degradeToLegacy("legacy_pk"); + + for (int i = 4; i < 16; i++) { + write(table, row(i % 2, "v" + i)); + assertThat(latest(table).numFiles()).isNull(); + } + + assertThat(commitKindCount(table, Snapshot.CommitKind.COMPACT)).isGreaterThan(0); + // the merge engine still resolves to the last value written per key + assertThat(readRows(table)).containsExactlyInAnyOrder("0:v14", "1:v15"); + } + + @Test + public void testOverwriteLegacyTable() throws Exception { + FileStoreTable table = createTable("legacy_overwrite", false); + write(table, row(1, "a"), row(2, "b")); + + table = degradeToLegacy("legacy_overwrite"); + + overwrite(table, row(9, "z")); + assertThat(latest(table).numFiles()).isNull(); + assertThat(readRows(table)).containsExactly("9:z"); + } + + @Test + public void testTagAndRollbackOnLegacyTable() throws Exception { + FileStoreTable table = createTable("legacy_tag", false); + write(table, row(1, "a")); + write(table, row(2, "b")); + + table = degradeToLegacy("legacy_tag"); + long targetId = table.snapshotManager().earliestSnapshotId(); + + assertThatCode(() -> reload("legacy_tag").createTag("legacy_t1")) + .doesNotThrowAnyException(); + Tag tag = table.tagManager().get("legacy_t1").orElseThrow(IllegalStateException::new); + assertThat(tag.numFiles()).isNull(); + assertThat(tag.totalFileSizeInBytes()).isNull(); + + table.rollbackTo(targetId); + table = reload("legacy_tag"); + assertThat(latest(table).numFiles()).isNull(); + assertThat(readRows(table)).containsExactly("1:a"); + } + + @Test + public void testSnapshotsSystemTableOnLegacyTable() throws Exception { + FileStoreTable table = createTable("legacy_sys", false); + write(table, row(1, "a")); + write(table, row(2, "b")); + + // a table that still carries the statistics reports them in the two new columns + Table systemTable = catalog.getTable(identifier("legacy_sys$snapshots")); + List rows = read(systemTable); + assertThat(rows).hasSize(2); + for (InternalRow row : rows) { + assertThat(row.isNullAt(16)).isFalse(); + assertThat(row.isNullAt(17)).isFalse(); + } + + degradeToLegacy("legacy_sys"); + + // after degrading, the same query must still work and report the columns as null + systemTable = catalog.getTable(identifier("legacy_sys$snapshots")); + rows = read(systemTable); + assertThat(rows).hasSize(2); + for (InternalRow row : rows) { + assertThat(row.isNullAt(16)).isTrue(); + assertThat(row.isNullAt(17)).isTrue(); + } + } + + @Test + public void testMixedLegacyAndCurrentSnapshotsAreReadable() throws Exception { + FileStoreTable table = createTable("legacy_mixed", false); + write(table, row(1, "a")); + write(table, row(2, "b")); + + // degrade only the first snapshot, as an upgraded table looks: old snapshots without the + // fields still in the retained history, newer ones with them + degradeSnapshot("legacy_mixed", table.snapshotManager().earliestSnapshotId()); + table = reload("legacy_mixed"); + + SnapshotManager manager = table.snapshotManager(); + assertThat(manager.snapshot(manager.earliestSnapshotId()).numFiles()).isNull(); + assertThat(manager.snapshot(manager.latestSnapshotId()).numFiles()).isNotNull(); + + // time travel into the degraded part of the history keeps working + assertThat( + readRows( + table.copy( + java.util.Collections.singletonMap( + CoreOptions.SCAN_SNAPSHOT_ID.key(), + String.valueOf(manager.earliestSnapshotId()))))) + .containsExactly("1:a"); + assertThat(readRows(table)).containsExactlyInAnyOrder("1:a", "2:b"); + } + + @Test + public void testManifestCompactionKeepsStats() throws Exception { + FileStoreTable table = createTable("manifest_compact", false); + for (int i = 0; i < 6; i++) { + write(table, row(i, "v" + i)); + } + Long filesBefore = latest(table).numFiles(); + Long sizeBefore = latest(table).totalFileSizeInBytes(); + + compactManifests(table); + table = reload("manifest_compact"); + + // compacting manifests only rewrites metadata, so the data file statistics carry over + assertThat(latest(table).numFiles()).isEqualTo(filesBefore); + assertThat(latest(table).totalFileSizeInBytes()).isEqualTo(sizeBefore); + assertStatsMatchManifests(table); + } + + @Test + public void testManifestCompactionOnLegacyTableStaysUnknown() throws Exception { + FileStoreTable table = createTable("manifest_compact_legacy", false); + for (int i = 0; i < 6; i++) { + write(table, row(i, "v" + i)); + } + + table = degradeToLegacy("manifest_compact_legacy"); + compactManifests(table); + table = reload("manifest_compact_legacy"); + + assertThat(latest(table).numFiles()).isNull(); + assertThat(readRows(table)) + .containsExactlyInAnyOrder("0:v0", "1:v1", "2:v2", "3:v3", "4:v4", "5:v5"); + } + + @Test + public void testExpireSnapshotsOnLegacyTable() throws Exception { + FileStoreTable table = createTable("legacy_expire", false); + for (int i = 0; i < 5; i++) { + write(table, row(i, "v" + i)); + } + + table = degradeToLegacy("legacy_expire"); + table.newExpireSnapshots() + .config( + ExpireConfig.builder() + .snapshotMaxDeletes(Integer.MAX_VALUE) + .snapshotRetainMax(2) + .snapshotRetainMin(1) + .build()) + .expire(); + table = reload("legacy_expire"); + + assertThat(latest(table).numFiles()).isNull(); + assertThat(readRows(table)) + .containsExactlyInAnyOrder("0:v0", "1:v1", "2:v2", "3:v3", "4:v4"); + + // writing after expiration on a legacy table still works + write(table, row(9, "z")); + assertThat(latest(table).numFiles()).isNull(); + assertThat(readRows(table)).hasSize(6); + } + + @Test + public void testReplaceManifestListReportsUnknown() throws Exception { + FileStoreTable table = createTable("replace_manifests", false); + write(table, row(1, "a"), row(2, "b")); + write(table, row(3, "c")); + List before = readRows(table); + assertThat(latest(table).numFiles()).isNotNull(); + + replaceManifestListWithSameLayout(table); + table = reload("replace_manifests"); + + // this path swaps the manifest layout wholesale, so the counters cannot be carried over. + // Reporting unknown is the only honest answer; reporting the previous numbers would be + // wrong for callers such as RemoveUnexistingManifestsAction, which drops files. + assertThat(latest(table).numFiles()).isNull(); + assertThat(latest(table).totalFileSizeInBytes()).isNull(); + + // the data itself is untouched, and the table keeps taking writes + assertThat(readRows(table)).containsExactlyInAnyOrderElementsOf(before); + write(table, row(4, "d")); + assertThat(latest(table).numFiles()).isNull(); + assertThat(readRows(table)).containsExactlyInAnyOrder("1:a", "2:b", "3:c", "4:d"); + } + + @Test + public void testReplaceManifestListThatDropsFiles() throws Exception { + // keep every commit in its own manifest so one of them can be dropped below + FileStoreTable table = + createTable( + "replace_manifests_drop", + false, + CoreOptions.MANIFEST_MERGE_MIN_COUNT.key(), + "100"); + write(table, row(1, "a")); + write(table, row(2, "b")); + write(table, row(3, "c")); + + long filesBefore = latest(table).numFiles(); + ManifestList manifestList = table.store().manifestListFactory().create(); + List manifests = manifestList.readDataManifests(latest(table)); + assertThat(manifests.size()).isGreaterThan(1); + + // drop one manifest, the way a repair operation removes manifests whose files are gone + Snapshot latest = latest(table); + List kept = new ArrayList<>(manifests.subList(1, manifests.size())); + Pair base = manifestList.write(kept); + Pair delta = manifestList.write(Collections.emptyList()); + try (FileStoreCommitImpl commit = + (FileStoreCommitImpl) table.store().newCommit(commitUser, table)) { + assertThat( + commit.replaceManifestList( + latest, + latest.totalRecordCount(), + base, + delta, + latest.indexManifest(), + latest.nextRowId())) + .isTrue(); + } + table = reload("replace_manifests_drop"); + + // the live file set really did shrink, so carrying the previous counters over would have + // published a number that is simply wrong; unknown is the correct answer + long filesAfter = scannedNumFiles(table); + assertThat(filesAfter).isLessThan(filesBefore); + assertThat(latest(table).numFiles()).isNull(); + assertThat(latest(table).totalFileSizeInBytes()).isNull(); + + // and the table stays usable afterwards + write(table, row(4, "d")); + assertThat(latest(table).numFiles()).isNull(); + assertThat(readRows(table)).contains("4:d"); + } + + @Test + public void testUnknownJsonFieldsAreIgnored() { + // an older reader must tolerate a snapshot written by a newer writer; the same mechanism + // that lets it ignore numFiles is exercised here with an unknown field + Snapshot parsed = + Snapshot.fromJson( + "{\n" + + " \"version\" : 3,\n" + + " \"id\" : 5,\n" + + " \"schemaId\" : 0,\n" + + " \"baseManifestList\" : \"base\",\n" + + " \"deltaManifestList\" : \"delta\",\n" + + " \"commitUser\" : \"user\",\n" + + " \"commitIdentifier\" : 0,\n" + + " \"commitKind\" : \"APPEND\",\n" + + " \"timeMillis\" : 1000,\n" + + " \"totalRecordCount\" : 10,\n" + + " \"deltaRecordCount\" : 10,\n" + + " \"numFiles\" : 7,\n" + + " \"totalFileSizeInBytes\" : 4096,\n" + + " \"someFieldFromTheFuture\" : \"whatever\"\n" + + "}"); + assertThat(parsed.numFiles()).isEqualTo(7L); + assertThat(parsed.totalFileSizeInBytes()).isEqualTo(4096L); + } + + // ------------------------------------------------------------------------------------------ + // helpers + // ------------------------------------------------------------------------------------------ + + private FileStoreTable createTable(String name, boolean primaryKey, String... options) + throws Exception { + Schema.Builder builder = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("v", DataTypes.STRING()) + .option(CoreOptions.BUCKET.key(), "1"); + for (int i = 0; i < options.length; i += 2) { + builder.option(options[i], options[i + 1]); + } + if (primaryKey) { + builder.primaryKey("k"); + } else { + builder.option(CoreOptions.BUCKET_KEY.key(), "k"); + } + Identifier identifier = identifier(name); + catalog.createTable(identifier, builder.build(), false); + return (FileStoreTable) catalog.getTable(identifier); + } + + private FileStoreTable reload(String name) throws Exception { + return (FileStoreTable) catalog.getTable(identifier(name)); + } + + private static GenericRow row(int k, String v) { + return GenericRow.of(k, BinaryString.fromString(v)); + } + + /** + * Rewrites the snapshot with exactly the manifest layout it already has, which is the shape + * metadata-repair operations commit through {@code replaceManifestList}. + */ + private void replaceManifestListWithSameLayout(FileStoreTable table) throws Exception { + Snapshot latest = latest(table); + ManifestList manifestList = table.store().manifestListFactory().create(); + Pair base = manifestList.write(manifestList.readDataManifests(latest)); + Pair delta = manifestList.write(Collections.emptyList()); + try (FileStoreCommitImpl commit = + (FileStoreCommitImpl) table.store().newCommit(commitUser, table)) { + assertThat( + commit.replaceManifestList( + latest, + latest.totalRecordCount(), + base, + delta, + latest.indexManifest(), + latest.nextRowId())) + .isTrue(); + } + } + + private void compactManifests(FileStoreTable table) throws Exception { + try (TableCommitImpl commit = table.newCommit(commitUser)) { + commit.compactManifests(); + } + } + + private void overwrite(FileStoreTable table, GenericRow... rows) throws Exception { + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder().withOverwrite(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + for (GenericRow row : rows) { + write.write(row); + } + commit.commit(write.prepareCommit()); + } + } + + private List readRows(Table table) throws Exception { + return read(table).stream() + .map(row -> row.getInt(0) + ":" + row.getString(1)) + .collect(Collectors.toList()); + } + + private static Snapshot latest(FileStoreTable table) { + Snapshot snapshot = table.snapshotManager().latestSnapshot(); + assertThat(snapshot).isNotNull(); + return snapshot; + } + + private static long scannedNumFiles(FileStoreTable table) { + return table.newScan().listPartitionEntries().stream() + .mapToLong(PartitionEntry::fileCount) + .sum(); + } + + private void assertStatsMatchManifests(FileStoreTable table) { + Snapshot snapshot = latest(table); + List entries = table.newScan().listPartitionEntries(); + long expectedFiles = entries.stream().mapToLong(PartitionEntry::fileCount).sum(); + long expectedSize = entries.stream().mapToLong(PartitionEntry::fileSizeInBytes).sum(); + + assertThat(snapshot.numFiles()) + .as("num files of snapshot %s", snapshot.id()) + .isEqualTo(expectedFiles); + assertThat(snapshot.totalFileSizeInBytes()) + .as("total file size of snapshot %s", snapshot.id()) + .isEqualTo(expectedSize); + } + + private int commitKindCount(FileStoreTable table, Snapshot.CommitKind kind) { + SnapshotManager manager = table.snapshotManager(); + int count = 0; + for (long id = manager.earliestSnapshotId(); id <= manager.latestSnapshotId(); id++) { + if (manager.snapshot(id).commitKind() == kind) { + count++; + } + } + return count; + } + + /** Rewrites every snapshot file the way a Paimon without these fields would have written it. */ + private FileStoreTable degradeToLegacy(String name) throws Exception { + FileStoreTable table = reload(name); + SnapshotManager manager = table.snapshotManager(); + List ids = new ArrayList<>(); + for (long id = manager.earliestSnapshotId(); id <= manager.latestSnapshotId(); id++) { + ids.add(id); + } + for (long id : ids) { + degradeSnapshot(name, id); + } + return reload(name); + } + + private void degradeSnapshot(String name, long snapshotId) throws Exception { + FileStoreTable table = reload(name); + Path path = table.snapshotManager().snapshotPath(snapshotId); + Snapshot snapshot = Snapshot.fromJson(table.fileIO().readFileUtf8(path)); + table.fileIO().overwriteFileUtf8(path, legacyJsonOf(snapshot)); + // the snapshot object and the table itself are cached, drop both so the rewritten file is + // the one that gets read back + table.snapshotManager().invalidateCache(); + catalog.invalidateTable(identifier(name)); + } + + /** + * Serializes through the constructor that predates the two fields, so the payload looks exactly + * like one written by an older version: the keys are absent, not null. + */ + private static String legacyJsonOf(Snapshot s) { + return new Snapshot( + s.version(), + s.uuid(), + s.id(), + s.schemaId(), + s.baseManifestList(), + s.baseManifestListSize(), + s.deltaManifestList(), + s.deltaManifestListSize(), + s.changelogManifestList(), + s.changelogManifestListSize(), + s.indexManifest(), + s.commitUser(), + s.writerVersion(), + s.commitIdentifier(), + s.commitKind(), + s.timeMillis(), + s.totalRecordCount(), + s.deltaRecordCount(), + s.changelogRecordCount(), + s.watermark(), + s.statistics(), + s.properties(), + s.nextRowId(), + s.operation()) + .toJson(); + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala index 27f98e71b615..c02c0b791a06 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala @@ -59,12 +59,18 @@ case class PaimonAnalyzeTableColumnCommand( } // compute stats - val totalSize = table - .newScan() - .listPartitionEntries() - .asScala - .map(_.fileSizeInBytes()) - .sum + // Prefer the total file size maintained incrementally on the snapshot. It is unknown for + // snapshots written before the field existed, or when a commit could not derive it, in which + // case fall back to folding the partition entries out of the manifests. + val totalSize = Option(currentSnapshot.totalFileSizeInBytes()) + .map(_.longValue()) + .getOrElse( + table + .newScan() + .listPartitionEntries() + .asScala + .map(_.fileSizeInBytes()) + .sum) val (mergedRecordCount, colStats) = PaimonStatsUtils.computeColumnStats(sparkSession, relation, attributes) From 0c11e6230a533c1bfd95e3bbe6a8d5f6377aa99e Mon Sep 17 00:00:00 2001 From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com> Date: Sun, 6 Sep 2026 15:05:17 +0800 Subject: [PATCH 2/3] [core] Recompute snapshot file statistics where they cannot be derived The previous commit left two gaps where numFiles and totalFileSizeInBytes stayed unknown for good. replaceManifestList reported unknown because it replaces the manifest layout wholesale and may drop files with it, so the previous counters do not apply. It now recomputes them from the manifests it is about to commit. That reads every entry, which is why it stays out of the regular commit path, but the callers are metadata repair operations - RemoveUnexistingManifestsAction and the data evolution row id reassigner - where a second pass is affordable and correctness by construction beats letting each caller supply its own numbers. Tables whose snapshots predate the fields had no way back at all: every later commit inherited the missing baseline. Manifest compaction now seeds them. It already rewrites the whole manifest set, and the compacted form is the cheapest one to fold, so compact_manifest becomes the way to recover the counters, after which the incremental chain continues on its own. To make that reliable the procedure now commits a snapshot when the counters are unknown even if the manifests need no merging; once they are known it returns early exactly as before. Both recompute paths share one helper that folds ADD and DELETE entries with opposite signs, so deletion entries cancel their adds rather than being counted. Tests cover: replacing a manifest layout with an equivalent one lands on the same numbers; replacing one that drops a manifest follows the shrink instead of reporting the stale count; seeding a legacy append table, a legacy primary key table with cancelling deletion entries, and a legacy table small enough that nothing needs merging; and that a seeded table keeps folding incrementally afterwards. RemoveUnexistingManifestsActionITCase and CompactManifestProcedureITCase exercise the same paths through Flink. --- docs/docs/concepts/system-tables.mdx | 13 +-- .../paimon/operation/FileStoreCommitImpl.java | 72 +++++++++++++-- .../paimon/table/SnapshotFileStatsTest.java | 90 +++++++++++++++---- 3 files changed, 144 insertions(+), 31 deletions(-) diff --git a/docs/docs/concepts/system-tables.mdx b/docs/docs/concepts/system-tables.mdx index fd952d46a85e..969d0ea62850 100644 --- a/docs/docs/concepts/system-tables.mdx +++ b/docs/docs/concepts/system-tables.mdx @@ -76,11 +76,14 @@ costs nothing extra, and they cover data files only: changelog files, index file vectors are not counted. Both columns are nullable, and `NULL` means unknown rather than zero. It appears for snapshots -written before these columns existed, and for commits that could not derive the values from the -previous snapshot, such as metadata repair operations that replace the manifest layout wholesale. -Once a snapshot reports unknown, the snapshots committed after it report unknown as well, until the -values are recomputed. Fall back to scanning the manifests when you read `NULL`; never read it as -zero. +written before these columns existed, and once a snapshot reports unknown the snapshots committed +after it report unknown as well, because there is no baseline to fold the next commit onto. Fall +back to scanning the manifests when you read `NULL`; never read it as zero. + +Run the `compact_manifest` procedure to recover the values on such a table. It rewrites the whole +manifest set anyway, so it folds the counters from the compacted manifests and commits them, after +which the incremental chain continues on its own. The procedure commits a snapshot for this even +when the manifests themselves need no merging. ### Schemas Table diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index d2d1b23e12af..24e8bb4578c6 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -1412,6 +1412,17 @@ public boolean replaceManifestList( Pair deltaManifestList, @Nullable String indexManifest, @Nullable Long nextRowId) { + // The manifest layout is replaced wholesale and callers may drop files along with it, so + // the file statistics cannot be carried over from the previous snapshot. Recompute them + // from the manifests actually being committed: this reads every entry, which is acceptable + // because this path only serves metadata repair, never regular commits. + List committedManifests = new ArrayList<>(); + committedManifests.addAll( + manifestList.read(baseManifestList.getLeft(), baseManifestList.getRight())); + committedManifests.addAll( + manifestList.read(deltaManifestList.getLeft(), deltaManifestList.getRight())); + FileStats fileStats = computeFileStats(committedManifests); + Snapshot newSnapshot = new Snapshot( latest.id() + 1, @@ -1437,10 +1448,8 @@ public boolean replaceManifestList( latest.properties(), nextRowId, null, - // the manifest layout is rewritten wholesale here, so the incremental file - // statistics cannot be carried over; leave them unknown - null, - null); + fileStats.numFiles, + fileStats.totalFileSizeInBytes); return commitSnapshotImpl(latest, newSnapshot, emptyList()); } @@ -1646,7 +1655,10 @@ private boolean compactManifestOnce() { manifestCompactionOptions(options, mergeBeforeManifests, partitionType), ioManager); - if (new HashSet<>(mergeBeforeManifests).equals(new HashSet<>(mergeAfterManifests))) { + boolean statsUnknown = + latestSnapshot.numFiles() == null || latestSnapshot.totalFileSizeInBytes() == null; + if (new HashSet<>(mergeBeforeManifests).equals(new HashSet<>(mergeAfterManifests)) + && !statsUnknown) { // no need to commit this snapshot, because no compact were happened return true; } @@ -1654,6 +1666,18 @@ private boolean compactManifestOnce() { Pair baseManifestList = manifestList.write(mergeAfterManifests); Pair deltaManifestList = manifestList.write(emptyList()); + // Manifest compaction only rewrites metadata, so known statistics carry over untouched. + // When they are unknown - a table whose snapshots predate the fields, or one that went + // through a metadata repair - this is the place to seed them: the manifest set has just + // been rewritten, and the compacted form is the cheapest one to fold. + Long numFiles = latestSnapshot.numFiles(); + Long totalFileSizeInBytes = latestSnapshot.totalFileSizeInBytes(); + if (statsUnknown) { + FileStats fileStats = computeFileStats(mergeAfterManifests); + numFiles = fileStats.numFiles; + totalFileSizeInBytes = fileStats.totalFileSizeInBytes; + } + // prepare snapshot file Snapshot newSnapshot = new Snapshot( @@ -1679,9 +1703,8 @@ private boolean compactManifestOnce() { latestSnapshot.properties(), latestSnapshot.nextRowId(), null, - // manifest compaction only rewrites metadata, the data files are untouched - latestSnapshot.numFiles(), - latestSnapshot.totalFileSizeInBytes()); + numFiles, + totalFileSizeInBytes); return commitSnapshotImpl(latestSnapshot, newSnapshot, emptyList()); } @@ -1700,6 +1723,39 @@ static CoreOptions manifestCompactionOptions( return new CoreOptions(compactOptions); } + /** + * Folds the live data file count and total size out of a manifest set by reading every entry. + * + *

This is the expensive way to obtain what {@link Snapshot#numFiles()} and {@link + * Snapshot#totalFileSizeInBytes()} normally maintain incrementally, so it belongs to + * maintenance operations only and must never run on the regular commit path. + */ + private FileStats computeFileStats(List manifests) { + long numFiles = 0L; + long totalFileSizeInBytes = 0L; + for (ManifestFileMeta manifest : manifests) { + for (ManifestEntry entry : + manifestFile.read(manifest.fileName(), manifest.fileSize())) { + long sign = entry.kind() == FileKind.ADD ? 1L : -1L; + numFiles += sign; + totalFileSizeInBytes += sign * entry.file().fileSize(); + } + } + return new FileStats(numFiles, totalFileSizeInBytes); + } + + /** Live data file count and total size of a snapshot. */ + private static class FileStats { + + private final long numFiles; + private final long totalFileSizeInBytes; + + private FileStats(long numFiles, long totalFileSizeInBytes) { + this.numFiles = numFiles; + this.totalFileSizeInBytes = totalFileSizeInBytes; + } + } + private boolean commitSnapshotImpl( @Nullable Snapshot baseSnapshot, Snapshot newSnapshot, diff --git a/paimon-core/src/test/java/org/apache/paimon/table/SnapshotFileStatsTest.java b/paimon-core/src/test/java/org/apache/paimon/table/SnapshotFileStatsTest.java index dbce22dbe1b6..025f421c7ab5 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/SnapshotFileStatsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/SnapshotFileStatsTest.java @@ -181,8 +181,12 @@ public void testWriteToLegacyTableStaysUnknownAndNeverWrong() throws Exception { assertThat(readRows(table)) .containsExactlyInAnyOrder("1:a", "2:b", "10:n0", "11:n1", "12:n2"); - // a full scan still answers the question the statistics would have answered + // a full scan still answers the question the statistics would have answered, and + // compacting the manifests is what brings the counters back assertThat(scannedNumFiles(table)).isGreaterThan(0L); + compactManifests(table); + table = reload("legacy_write"); + assertStatsMatchManifests(table); } @Test @@ -309,19 +313,70 @@ public void testManifestCompactionKeepsStats() throws Exception { } @Test - public void testManifestCompactionOnLegacyTableStaysUnknown() throws Exception { + public void testManifestCompactionSeedsLegacyTable() throws Exception { FileStoreTable table = createTable("manifest_compact_legacy", false); for (int i = 0; i < 6; i++) { write(table, row(i, "v" + i)); } + long expectedFiles = scannedNumFiles(table); table = degradeToLegacy("manifest_compact_legacy"); + assertThat(latest(table).numFiles()).isNull(); + + // manifest compaction rewrites the whole manifest set anyway, so it seeds the counters compactManifests(table); table = reload("manifest_compact_legacy"); - assertThat(latest(table).numFiles()).isNull(); + assertThat(latest(table).numFiles()).isEqualTo(expectedFiles); + assertStatsMatchManifests(table); assertThat(readRows(table)) .containsExactlyInAnyOrder("0:v0", "1:v1", "2:v2", "3:v3", "4:v4", "5:v5"); + + // once seeded, the incremental chain picks up again + write(table, row(9, "z")); + assertStatsMatchManifests(table); + } + + @Test + public void testManifestCompactionSeedsEvenWhenNothingToMerge() throws Exception { + // a legacy table small enough that the manifests need no merging must still get seeded, + // otherwise compact_manifest would be an unreliable way to recover the counters + FileStoreTable table = createTable("manifest_compact_noop", false); + write(table, row(1, "a")); + long expectedFiles = scannedNumFiles(table); + + table = degradeToLegacy("manifest_compact_noop"); + long snapshotsBefore = latest(table).id(); + + compactManifests(table); + table = reload("manifest_compact_noop"); + + assertThat(latest(table).id()).isGreaterThan(snapshotsBefore); + assertThat(latest(table).numFiles()).isEqualTo(expectedFiles); + assertStatsMatchManifests(table); + + // and a second run is a no-op now that the counters are known + long seededId = latest(table).id(); + compactManifests(table); + table = reload("manifest_compact_noop"); + assertThat(latest(table).id()).isEqualTo(seededId); + } + + @Test + public void testManifestCompactionSeedsLegacyPrimaryKeyTable() throws Exception { + FileStoreTable table = createTable("manifest_compact_legacy_pk", true); + for (int i = 0; i < 8; i++) { + write(table, row(i % 3, "v" + i)); + } + long expectedFiles = scannedNumFiles(table); + + table = degradeToLegacy("manifest_compact_legacy_pk"); + compactManifests(table); + table = reload("manifest_compact_legacy_pk"); + + // deletion entries left behind by compaction must cancel their adds, not be counted + assertThat(latest(table).numFiles()).isEqualTo(expectedFiles); + assertStatsMatchManifests(table); } @Test @@ -353,26 +408,25 @@ public void testExpireSnapshotsOnLegacyTable() throws Exception { } @Test - public void testReplaceManifestListReportsUnknown() throws Exception { + public void testReplaceManifestListRecomputes() throws Exception { FileStoreTable table = createTable("replace_manifests", false); write(table, row(1, "a"), row(2, "b")); write(table, row(3, "c")); List before = readRows(table); - assertThat(latest(table).numFiles()).isNotNull(); + Long filesBefore = latest(table).numFiles(); + assertThat(filesBefore).isNotNull(); replaceManifestListWithSameLayout(table); table = reload("replace_manifests"); - // this path swaps the manifest layout wholesale, so the counters cannot be carried over. - // Reporting unknown is the only honest answer; reporting the previous numbers would be - // wrong for callers such as RemoveUnexistingManifestsAction, which drops files. - assertThat(latest(table).numFiles()).isNull(); - assertThat(latest(table).totalFileSizeInBytes()).isNull(); + // the layout was replaced with an equivalent one, so recomputing lands on the same numbers + assertStatsMatchManifests(table); + assertThat(latest(table).numFiles()).isEqualTo(filesBefore); - // the data itself is untouched, and the table keeps taking writes + // the data itself is untouched, and the chain continues from the recomputed values assertThat(readRows(table)).containsExactlyInAnyOrderElementsOf(before); write(table, row(4, "d")); - assertThat(latest(table).numFiles()).isNull(); + assertStatsMatchManifests(table); assertThat(readRows(table)).containsExactlyInAnyOrder("1:a", "2:b", "3:c", "4:d"); } @@ -413,16 +467,16 @@ public void testReplaceManifestListThatDropsFiles() throws Exception { } table = reload("replace_manifests_drop"); - // the live file set really did shrink, so carrying the previous counters over would have - // published a number that is simply wrong; unknown is the correct answer + // the live file set really did shrink; carrying the previous counters over would have + // published a number that no longer matches it, so the values are recomputed instead long filesAfter = scannedNumFiles(table); assertThat(filesAfter).isLessThan(filesBefore); - assertThat(latest(table).numFiles()).isNull(); - assertThat(latest(table).totalFileSizeInBytes()).isNull(); + assertThat(latest(table).numFiles()).isEqualTo(filesAfter); + assertStatsMatchManifests(table); - // and the table stays usable afterwards + // and the table stays usable afterwards, with the chain continuing from the new values write(table, row(4, "d")); - assertThat(latest(table).numFiles()).isNull(); + assertStatsMatchManifests(table); assertThat(readRows(table)).contains("4:d"); } From 2bffd476fcdafafe0810cf17cff775a25c14334d Mon Sep 17 00:00:00 2001 From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com> Date: Sun, 6 Sep 2026 16:42:12 +0800 Subject: [PATCH 3/3] [core] Fix SnapshotsTableTest for the two new snapshot columns SnapshotsTableTest compares whole rows, so it has to list every column of the snapshots system table. Add num_files and total_file_size_in_bytes to the expected rows. --- .../org/apache/paimon/table/system/SnapshotsTableTest.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/paimon-core/src/test/java/org/apache/paimon/table/system/SnapshotsTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/system/SnapshotsTableTest.java index 648344b2b889..56c3377e5672 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/system/SnapshotsTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/system/SnapshotsTableTest.java @@ -162,7 +162,9 @@ private List getExpectedResult(long[] snapshotIds) { snapshot.operation() == null ? null : BinaryString.fromString(snapshot.operation().toString()), - BinaryString.fromString(snapshot.writerVersion()))); + BinaryString.fromString(snapshot.writerVersion()), + snapshot.numFiles(), + snapshot.totalFileSizeInBytes())); } return expectedRow;