diff --git a/docs/docs/concepts/system-tables.mdx b/docs/docs/concepts/system-tables.mdx index 0bb6d2629cc4..969d0ea62850 100644 --- a/docs/docs/concepts/system-tables.mdx +++ b/docs/docs/concepts/system-tables.mdx @@ -70,6 +70,21 @@ 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 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 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..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 @@ -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( @@ -1385,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, @@ -1409,7 +1447,9 @@ public boolean replaceManifestList( // if empty properties, just set to null latest.properties(), nextRowId, - null); + null, + fileStats.numFiles, + fileStats.totalFileSizeInBytes); return commitSnapshotImpl(latest, newSnapshot, emptyList()); } @@ -1495,7 +1535,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 @@ -1612,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; } @@ -1620,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( @@ -1644,7 +1702,9 @@ private boolean compactManifestOnce() { latestSnapshot.statistics(), latestSnapshot.properties(), latestSnapshot.nextRowId(), - null); + null, + numFiles, + totalFileSizeInBytes); return commitSnapshotImpl(latestSnapshot, newSnapshot, emptyList()); } @@ -1663,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/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..025f421c7ab5 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/SnapshotFileStatsTest.java @@ -0,0 +1,681 @@ +/* + * 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, and + // compacting the manifests is what brings the counters back + assertThat(scannedNumFiles(table)).isGreaterThan(0L); + compactManifests(table); + table = reload("legacy_write"); + assertStatsMatchManifests(table); + } + + @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 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()).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 + 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 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); + Long filesBefore = latest(table).numFiles(); + assertThat(filesBefore).isNotNull(); + + replaceManifestListWithSameLayout(table); + table = reload("replace_manifests"); + + // 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 chain continues from the recomputed values + assertThat(readRows(table)).containsExactlyInAnyOrderElementsOf(before); + write(table, row(4, "d")); + assertStatsMatchManifests(table); + 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; 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()).isEqualTo(filesAfter); + assertStatsMatchManifests(table); + + // and the table stays usable afterwards, with the chain continuing from the new values + write(table, row(4, "d")); + assertStatsMatchManifests(table); + 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-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; 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)