Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
zhuxiangyi wants to merge 3 commits into from
Closed

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b59b9e5
[core][spark] Maintain num files and total file size on snapshots
zhuxiangyi Sep 6, 2026
0c11e62
[core] Recompute snapshot file statistics where they cannot be derived
zhuxiangyi Sep 6, 2026
2bffd47
[core] Fix SnapshotsTableTest for the two new snapshot columns
zhuxiangyi Sep 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions docs/docs/concepts/system-tables.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
166 changes: 162 additions & 4 deletions paimon-api/src/main/java/org/apache/paimon/Snapshot.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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,
Expand All @@ -226,6 +241,58 @@ public Snapshot(
@Nullable Map<String, String> 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<String, String> properties,
@Nullable Long nextRowId,
@Nullable Operation operation,
@Nullable Long numFiles,
@Nullable Long totalFileSizeInBytes) {
this(
CURRENT_VERSION,
UUID.randomUUID().toString(),
Expand All @@ -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<String, String> 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
Expand Down Expand Up @@ -279,7 +406,9 @@ public Snapshot(
@JsonProperty(FIELD_STATISTICS) @Nullable String statistics,
@JsonProperty(FIELD_PROPERTIES) @Nullable Map<String, String> 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;
Expand All @@ -304,6 +433,8 @@ public Snapshot(
this.properties = properties;
this.nextRowId = nextRowId;
this.operation = operation;
this.numFiles = numFiles;
this.totalFileSizeInBytes = totalFileSizeInBytes;
}

@JsonGetter(FIELD_VERSION)
Expand Down Expand Up @@ -439,6 +570,29 @@ public Operation operation() {
return operation;
}

/**
* Number of live data files in this snapshot, maintained incrementally at commit time.
*
* <p>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);
}
Expand Down Expand Up @@ -469,7 +623,9 @@ public int hashCode() {
statistics,
properties,
nextRowId,
operation);
operation,
numFiles,
totalFileSizeInBytes);
}

@Override
Expand Down Expand Up @@ -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. */
Expand Down
12 changes: 9 additions & 3 deletions paimon-core/src/main/java/org/apache/paimon/Changelog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,9 @@ public Changelog(Snapshot snapshot) {
snapshot.statistics(),
snapshot.properties,
snapshot.nextRowId,
snapshot.operation);
snapshot.operation,
snapshot.numFiles,
snapshot.totalFileSizeInBytes);
}

@JsonCreator
Expand Down Expand Up @@ -95,7 +97,9 @@ public Changelog(
@JsonProperty(FIELD_STATISTICS) @Nullable String statistics,
@JsonProperty(FIELD_PROPERTIES) Map<String, String> 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,
Expand All @@ -120,7 +124,9 @@ public Changelog(
statistics,
properties,
nextRowId,
operation);
operation,
numFiles,
totalFileSizeInBytes);
}

public static Changelog fromJson(String json) {
Expand Down
Loading
Loading