badalprasadsingh commented on code in PR #2394: URL: https://github.com/apache/iceberg-go/pull/2394#discussion_r4452196594
########## table/snapshot_file_diff.go: ########## @@ -0,0 +1,273 @@ +// 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 table + +import ( + "cmp" + "context" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// SnapshotFileDiff describes the physical file changes between two snapshots. +// Data and delete files are reported separately; position-delete files and +// deletion vectors are both included in the delete-file slices. +// +// A rewrite that replaces data files may change these slices even when the +// logical table rows are unchanged. This is a file-level diff, not a row diff. +type SnapshotFileDiff struct { + FromSnapshotID int64 + ToSnapshotID int64 + + AddedDataFiles []iceberg.DataFile + RemovedDataFiles []iceberg.DataFile + AddedDeleteFiles []iceberg.DataFile + RemovedDeleteFiles []iceberg.DataFile +} + +// HasChanges reports whether either snapshot references physical files the +// other does not. +func (d SnapshotFileDiff) HasChanges() bool { + return len(d.AddedDataFiles)+len(d.RemovedDataFiles)+ + len(d.AddedDeleteFiles)+len(d.RemovedDeleteFiles) != 0 +} + +// DiffSnapshotFiles compares the live files referenced by any two snapshots +// still present in the table's metadata. The snapshots need not be ancestors; +// for example, they may belong to different branches. The order of the +// arguments determines the direction of additions and removals. +// +// Iceberg manifests are immutable. Shared manifests therefore contain +// identical live entries and can be skipped without opening their Avro files. +// For the remaining manifests, ADDED and EXISTING entries are live; DELETED +// entries are tombstones and must not be included. Data-file paths identify +// immutable content, so a file rewritten into a different manifest does not +// count as added or removed. In v3, deletion vectors use the Puffin +// location, blob offset, and blob length as their identity. +// +// This works for v1 snapshots that embed manifest paths, as well as v1-v3 +// snapshots with manifest lists. It reads only metadata and never reads the +// Parquet, Avro, or ORC data files. +func (t Table) DiffSnapshotFiles(ctx context.Context, fromSnapshotID, toSnapshotID int64) (SnapshotFileDiff, error) { Review Comment: #2062 and #2083 move read-only services off Table before v1. Better: InspectTable or free function. ########## table/snapshot_file_diff.go: ########## @@ -0,0 +1,273 @@ +// 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 table + +import ( + "cmp" + "context" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// SnapshotFileDiff describes the physical file changes between two snapshots. +// Data and delete files are reported separately; position-delete files and +// deletion vectors are both included in the delete-file slices. +// +// A rewrite that replaces data files may change these slices even when the +// logical table rows are unchanged. This is a file-level diff, not a row diff. +type SnapshotFileDiff struct { + FromSnapshotID int64 + ToSnapshotID int64 + + AddedDataFiles []iceberg.DataFile + RemovedDataFiles []iceberg.DataFile + AddedDeleteFiles []iceberg.DataFile + RemovedDeleteFiles []iceberg.DataFile +} + +// HasChanges reports whether either snapshot references physical files the +// other does not. +func (d SnapshotFileDiff) HasChanges() bool { + return len(d.AddedDataFiles)+len(d.RemovedDataFiles)+ + len(d.AddedDeleteFiles)+len(d.RemovedDeleteFiles) != 0 +} + +// DiffSnapshotFiles compares the live files referenced by any two snapshots +// still present in the table's metadata. The snapshots need not be ancestors; +// for example, they may belong to different branches. The order of the +// arguments determines the direction of additions and removals. +// +// Iceberg manifests are immutable. Shared manifests therefore contain +// identical live entries and can be skipped without opening their Avro files. +// For the remaining manifests, ADDED and EXISTING entries are live; DELETED +// entries are tombstones and must not be included. Data-file paths identify +// immutable content, so a file rewritten into a different manifest does not +// count as added or removed. In v3, deletion vectors use the Puffin +// location, blob offset, and blob length as their identity. +// +// This works for v1 snapshots that embed manifest paths, as well as v1-v3 +// snapshots with manifest lists. It reads only metadata and never reads the +// Parquet, Avro, or ORC data files. +func (t Table) DiffSnapshotFiles(ctx context.Context, fromSnapshotID, toSnapshotID int64) (SnapshotFileDiff, error) { + diff := SnapshotFileDiff{ + FromSnapshotID: fromSnapshotID, + ToSnapshotID: toSnapshotID, + } + if err := ctx.Err(); err != nil { + return diff, err + } + if t.metadata == nil { + return diff, fmt.Errorf("%w: table metadata is not available", ErrInvalidMetadata) + } + + from := t.metadata.SnapshotByID(fromSnapshotID) + if from == nil { + return diff, fmt.Errorf("%w: from snapshot %d", ErrSnapshotNotFound, fromSnapshotID) + } + to := t.metadata.SnapshotByID(toSnapshotID) + if to == nil { + return diff, fmt.Errorf("%w: to snapshot %d", ErrSnapshotNotFound, toSnapshotID) + } + if fromSnapshotID == toSnapshotID { + return diff, nil + } + if t.fsF == nil { + return diff, fmt.Errorf("%w: table file IO is not configured", ErrInvalidOperation) + } + + fs, err := t.fsF(ctx) + if err != nil { + return diff, fmt.Errorf("resolve table file IO: %w", err) + } + if fs == nil { + return diff, fmt.Errorf("%w: table file IO is nil", ErrInvalidOperation) + } + // Share the resolved FileIO between the two manifest-list loads and the + // changed manifest entries. Keep the normal bounded manifest-list cache. + fsF := func(context.Context) (iceio.IO, error) { return fs, nil } + + fromSet, err := t.manifestSetWithFSF(ctx, *from, fsF) + if err != nil { + return diff, fmt.Errorf("read from snapshot %d manifests: %w", fromSnapshotID, err) + } + toSet, err := t.manifestSetWithFSF(ctx, *to, fsF) + if err != nil { + return diff, fmt.Errorf("read to snapshot %d manifests: %w", toSnapshotID, err) + } + + fromOnly, toOnly := uniqueSnapshotManifests(fromSet.borrowAllManifests(), toSet.borrowAllManifests()) + if len(fromOnly) == 0 && len(toOnly) == 0 { + return diff, nil + } + + fromFiles, err := snapshotDiffLiveFiles(ctx, fs, fromOnly) + if err != nil { + return diff, fmt.Errorf("read from snapshot %d entries: %w", fromSnapshotID, err) + } + toFiles, err := snapshotDiffLiveFiles(ctx, fs, toOnly) + if err != nil { + return diff, fmt.Errorf("read to snapshot %d entries: %w", toSnapshotID, err) + } + + // A manifest rewrite can move the same data file into a different + // manifest without modifying table contents. Cancel those paths out. + for key, fromFile := range fromFiles { + if toFile, ok := toFiles[key]; ok { + if fromFile.ContentType() != toFile.ContentType() { + return diff, fmt.Errorf("%w: content file %q changed type between snapshots", + ErrInvalidMetadata, key.path) + } + delete(fromFiles, key) + delete(toFiles, key) + } + } + + for _, file := range fromFiles { + if file.ContentType() == iceberg.EntryContentData { + diff.RemovedDataFiles = append(diff.RemovedDataFiles, file) + } else { + diff.RemovedDeleteFiles = append(diff.RemovedDeleteFiles, file) + } + } + for _, file := range toFiles { + if file.ContentType() == iceberg.EntryContentData { + diff.AddedDataFiles = append(diff.AddedDataFiles, file) + } else { + diff.AddedDeleteFiles = append(diff.AddedDeleteFiles, file) + } + } + + // Maps have unstable iteration order. Stable paths make the API useful + // for reproducible audit logs and change-detection checkpoints. + sortSnapshotDiffFiles(diff.AddedDataFiles) + sortSnapshotDiffFiles(diff.RemovedDataFiles) + sortSnapshotDiffFiles(diff.AddedDeleteFiles) + sortSnapshotDiffFiles(diff.RemovedDeleteFiles) + + return diff, nil +} + +// uniqueSnapshotManifests skips immutable manifests that are present in both +// snapshots. No ancestry walk is needed, and different branches work. +func uniqueSnapshotManifests(from, to []iceberg.ManifestFile) (fromOnly, toOnly []iceberg.ManifestFile) { + fromPaths := make(map[string]struct{}, len(from)) + toPaths := make(map[string]struct{}, len(to)) + for _, manifest := range from { + fromPaths[manifest.FilePath()] = struct{}{} + } + for _, manifest := range to { + toPaths[manifest.FilePath()] = struct{}{} + if _, shared := fromPaths[manifest.FilePath()]; !shared { + toOnly = append(toOnly, manifest) + } + } + for _, manifest := range from { + if _, shared := toPaths[manifest.FilePath()]; !shared { + fromOnly = append(fromOnly, manifest) + } + } + + return fromOnly, toOnly +} + +// snapshotDiffFileKey uses the spec's identity for content files. A v3 Puffin +// file may hold several different deletion vector blobs, so its path alone +// is not enough to identify the vector. +type snapshotDiffFileKey struct { + path string + vector bool + offset int64 + length int64 +} + +func snapshotFileDiffKey(file iceberg.DataFile) (snapshotDiffFileKey, error) { + key := snapshotDiffFileKey{path: file.FilePath()} + if key.path == "" { + return key, fmt.Errorf("%w: content file path is empty", ErrInvalidMetadata) + } + if IsDeletionVector(file) { + offset := file.ContentOffset() + length := file.ContentSizeInBytes() + if offset == nil || length == nil || *offset < 0 || *length <= 0 { + return key, fmt.Errorf("%w: deletion vector %q has an invalid offset or length", + ErrInvalidMetadata, file.FilePath()) + } + key.vector = true + key.offset = *offset + key.length = *length + } + + return key, nil +} + +func snapshotDiffLiveFiles(ctx context.Context, fs iceio.IO, manifests []iceberg.ManifestFile) (map[snapshotDiffFileKey]iceberg.DataFile, error) { + files := make(map[snapshotDiffFileKey]iceberg.DataFile) + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + return nil, err + } + for entry, err := range manifest.Entries(fs, true) { Review Comment: some manifests have only DELETED entries, so they can't hold a live file, and this code still opens them, and each open is a wasted read. add this check, plus a test, unknown v1 counts return true, so old tables stay safe: ```go if !manifest.HasAddedFiles() && !manifest.HasExistingFiles() { continue } ``` ########## table/snapshot_file_diff.go: ########## @@ -0,0 +1,273 @@ +// 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 table + +import ( + "cmp" + "context" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// SnapshotFileDiff describes the physical file changes between two snapshots. +// Data and delete files are reported separately; position-delete files and +// deletion vectors are both included in the delete-file slices. +// +// A rewrite that replaces data files may change these slices even when the +// logical table rows are unchanged. This is a file-level diff, not a row diff. +type SnapshotFileDiff struct { + FromSnapshotID int64 + ToSnapshotID int64 + + AddedDataFiles []iceberg.DataFile + RemovedDataFiles []iceberg.DataFile + AddedDeleteFiles []iceberg.DataFile + RemovedDeleteFiles []iceberg.DataFile +} + +// HasChanges reports whether either snapshot references physical files the +// other does not. +func (d SnapshotFileDiff) HasChanges() bool { + return len(d.AddedDataFiles)+len(d.RemovedDataFiles)+ + len(d.AddedDeleteFiles)+len(d.RemovedDeleteFiles) != 0 +} + +// DiffSnapshotFiles compares the live files referenced by any two snapshots +// still present in the table's metadata. The snapshots need not be ancestors; +// for example, they may belong to different branches. The order of the +// arguments determines the direction of additions and removals. +// +// Iceberg manifests are immutable. Shared manifests therefore contain +// identical live entries and can be skipped without opening their Avro files. +// For the remaining manifests, ADDED and EXISTING entries are live; DELETED +// entries are tombstones and must not be included. Data-file paths identify +// immutable content, so a file rewritten into a different manifest does not +// count as added or removed. In v3, deletion vectors use the Puffin +// location, blob offset, and blob length as their identity. +// +// This works for v1 snapshots that embed manifest paths, as well as v1-v3 +// snapshots with manifest lists. It reads only metadata and never reads the +// Parquet, Avro, or ORC data files. +func (t Table) DiffSnapshotFiles(ctx context.Context, fromSnapshotID, toSnapshotID int64) (SnapshotFileDiff, error) { + diff := SnapshotFileDiff{ + FromSnapshotID: fromSnapshotID, + ToSnapshotID: toSnapshotID, + } + if err := ctx.Err(); err != nil { + return diff, err + } + if t.metadata == nil { + return diff, fmt.Errorf("%w: table metadata is not available", ErrInvalidMetadata) + } + + from := t.metadata.SnapshotByID(fromSnapshotID) + if from == nil { + return diff, fmt.Errorf("%w: from snapshot %d", ErrSnapshotNotFound, fromSnapshotID) + } + to := t.metadata.SnapshotByID(toSnapshotID) + if to == nil { + return diff, fmt.Errorf("%w: to snapshot %d", ErrSnapshotNotFound, toSnapshotID) + } + if fromSnapshotID == toSnapshotID { + return diff, nil + } + if t.fsF == nil { + return diff, fmt.Errorf("%w: table file IO is not configured", ErrInvalidOperation) + } + + fs, err := t.fsF(ctx) + if err != nil { + return diff, fmt.Errorf("resolve table file IO: %w", err) + } + if fs == nil { + return diff, fmt.Errorf("%w: table file IO is nil", ErrInvalidOperation) + } + // Share the resolved FileIO between the two manifest-list loads and the + // changed manifest entries. Keep the normal bounded manifest-list cache. + fsF := func(context.Context) (iceio.IO, error) { return fs, nil } Review Comment: can we reuse the function `sharedSnapshotManifestFSF` here? it will remove this duplicate fs setup here ########## table/snapshot_file_diff.go: ########## @@ -0,0 +1,273 @@ +// 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 table + +import ( + "cmp" + "context" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// SnapshotFileDiff describes the physical file changes between two snapshots. +// Data and delete files are reported separately; position-delete files and +// deletion vectors are both included in the delete-file slices. +// +// A rewrite that replaces data files may change these slices even when the +// logical table rows are unchanged. This is a file-level diff, not a row diff. +type SnapshotFileDiff struct { + FromSnapshotID int64 + ToSnapshotID int64 + + AddedDataFiles []iceberg.DataFile + RemovedDataFiles []iceberg.DataFile + AddedDeleteFiles []iceberg.DataFile + RemovedDeleteFiles []iceberg.DataFile +} + +// HasChanges reports whether either snapshot references physical files the +// other does not. +func (d SnapshotFileDiff) HasChanges() bool { + return len(d.AddedDataFiles)+len(d.RemovedDataFiles)+ + len(d.AddedDeleteFiles)+len(d.RemovedDeleteFiles) != 0 +} + +// DiffSnapshotFiles compares the live files referenced by any two snapshots +// still present in the table's metadata. The snapshots need not be ancestors; +// for example, they may belong to different branches. The order of the +// arguments determines the direction of additions and removals. +// +// Iceberg manifests are immutable. Shared manifests therefore contain +// identical live entries and can be skipped without opening their Avro files. +// For the remaining manifests, ADDED and EXISTING entries are live; DELETED +// entries are tombstones and must not be included. Data-file paths identify +// immutable content, so a file rewritten into a different manifest does not +// count as added or removed. In v3, deletion vectors use the Puffin +// location, blob offset, and blob length as their identity. +// +// This works for v1 snapshots that embed manifest paths, as well as v1-v3 +// snapshots with manifest lists. It reads only metadata and never reads the +// Parquet, Avro, or ORC data files. +func (t Table) DiffSnapshotFiles(ctx context.Context, fromSnapshotID, toSnapshotID int64) (SnapshotFileDiff, error) { + diff := SnapshotFileDiff{ + FromSnapshotID: fromSnapshotID, + ToSnapshotID: toSnapshotID, + } + if err := ctx.Err(); err != nil { + return diff, err + } + if t.metadata == nil { + return diff, fmt.Errorf("%w: table metadata is not available", ErrInvalidMetadata) + } + + from := t.metadata.SnapshotByID(fromSnapshotID) + if from == nil { + return diff, fmt.Errorf("%w: from snapshot %d", ErrSnapshotNotFound, fromSnapshotID) + } + to := t.metadata.SnapshotByID(toSnapshotID) + if to == nil { + return diff, fmt.Errorf("%w: to snapshot %d", ErrSnapshotNotFound, toSnapshotID) + } + if fromSnapshotID == toSnapshotID { + return diff, nil + } + if t.fsF == nil { + return diff, fmt.Errorf("%w: table file IO is not configured", ErrInvalidOperation) + } + + fs, err := t.fsF(ctx) + if err != nil { + return diff, fmt.Errorf("resolve table file IO: %w", err) + } + if fs == nil { + return diff, fmt.Errorf("%w: table file IO is nil", ErrInvalidOperation) + } + // Share the resolved FileIO between the two manifest-list loads and the + // changed manifest entries. Keep the normal bounded manifest-list cache. + fsF := func(context.Context) (iceio.IO, error) { return fs, nil } + + fromSet, err := t.manifestSetWithFSF(ctx, *from, fsF) + if err != nil { + return diff, fmt.Errorf("read from snapshot %d manifests: %w", fromSnapshotID, err) + } + toSet, err := t.manifestSetWithFSF(ctx, *to, fsF) + if err != nil { + return diff, fmt.Errorf("read to snapshot %d manifests: %w", toSnapshotID, err) + } + + fromOnly, toOnly := uniqueSnapshotManifests(fromSet.borrowAllManifests(), toSet.borrowAllManifests()) + if len(fromOnly) == 0 && len(toOnly) == 0 { + return diff, nil + } + + fromFiles, err := snapshotDiffLiveFiles(ctx, fs, fromOnly) + if err != nil { + return diff, fmt.Errorf("read from snapshot %d entries: %w", fromSnapshotID, err) + } + toFiles, err := snapshotDiffLiveFiles(ctx, fs, toOnly) + if err != nil { + return diff, fmt.Errorf("read to snapshot %d entries: %w", toSnapshotID, err) + } + + // A manifest rewrite can move the same data file into a different + // manifest without modifying table contents. Cancel those paths out. + for key, fromFile := range fromFiles { + if toFile, ok := toFiles[key]; ok { + if fromFile.ContentType() != toFile.ContentType() { + return diff, fmt.Errorf("%w: content file %q changed type between snapshots", + ErrInvalidMetadata, key.path) + } + delete(fromFiles, key) + delete(toFiles, key) + } + } + + for _, file := range fromFiles { + if file.ContentType() == iceberg.EntryContentData { + diff.RemovedDataFiles = append(diff.RemovedDataFiles, file) + } else { + diff.RemovedDeleteFiles = append(diff.RemovedDeleteFiles, file) + } + } + for _, file := range toFiles { + if file.ContentType() == iceberg.EntryContentData { + diff.AddedDataFiles = append(diff.AddedDataFiles, file) + } else { + diff.AddedDeleteFiles = append(diff.AddedDeleteFiles, file) + } + } + + // Maps have unstable iteration order. Stable paths make the API useful + // for reproducible audit logs and change-detection checkpoints. + sortSnapshotDiffFiles(diff.AddedDataFiles) + sortSnapshotDiffFiles(diff.RemovedDataFiles) + sortSnapshotDiffFiles(diff.AddedDeleteFiles) + sortSnapshotDiffFiles(diff.RemovedDeleteFiles) + + return diff, nil +} + +// uniqueSnapshotManifests skips immutable manifests that are present in both +// snapshots. No ancestry walk is needed, and different branches work. +func uniqueSnapshotManifests(from, to []iceberg.ManifestFile) (fromOnly, toOnly []iceberg.ManifestFile) { + fromPaths := make(map[string]struct{}, len(from)) + toPaths := make(map[string]struct{}, len(to)) + for _, manifest := range from { + fromPaths[manifest.FilePath()] = struct{}{} + } + for _, manifest := range to { + toPaths[manifest.FilePath()] = struct{}{} + if _, shared := fromPaths[manifest.FilePath()]; !shared { + toOnly = append(toOnly, manifest) + } + } + for _, manifest := range from { + if _, shared := toPaths[manifest.FilePath()]; !shared { + fromOnly = append(fromOnly, manifest) + } + } + + return fromOnly, toOnly +} + +// snapshotDiffFileKey uses the spec's identity for content files. A v3 Puffin +// file may hold several different deletion vector blobs, so its path alone +// is not enough to identify the vector. +type snapshotDiffFileKey struct { + path string + vector bool + offset int64 + length int64 +} + +func snapshotFileDiffKey(file iceberg.DataFile) (snapshotDiffFileKey, error) { + key := snapshotDiffFileKey{path: file.FilePath()} + if key.path == "" { + return key, fmt.Errorf("%w: content file path is empty", ErrInvalidMetadata) + } + if IsDeletionVector(file) { + offset := file.ContentOffset() + length := file.ContentSizeInBytes() + if offset == nil || length == nil || *offset < 0 || *length <= 0 { + return key, fmt.Errorf("%w: deletion vector %q has an invalid offset or length", + ErrInvalidMetadata, file.FilePath()) + } + key.vector = true + key.offset = *offset + key.length = *length + } + + return key, nil +} + +func snapshotDiffLiveFiles(ctx context.Context, fs iceio.IO, manifests []iceberg.ManifestFile) (map[snapshotDiffFileKey]iceberg.DataFile, error) { + files := make(map[snapshotDiffFileKey]iceberg.DataFile) + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + return nil, err + } + for entry, err := range manifest.Entries(fs, true) { Review Comment: This code reads manifests one at a time. After a manifest rewrite, no manifest is shared, so every one gets read. On a big table with slow storage, this can take a long time. The scanner (`collectManifestEntries`) already reads in parallel with errgroup, so can we do the same? Merge results in manifest order, so the duplicate error stays the same on every run. Should this take options now, like WithDeleteConcurrency? apidiff flags adding them later. ########## README.md: ########## @@ -119,6 +119,34 @@ Tests requiring Spark 4 (currently the variant and unknown-type tests) automatic | Create New Manifests | X | | Plan Scan | Local + Remote | | Plan Scan for Snapshot | Local + Remote | +| Diff Snapshot Files | X | + +### Snapshot file diffs (v1-v3) Review Comment: we should move this usage section to `website/src/api.md`, under "Snapshots and refs", and keep only the table row here. #2397 documents its API in that file too, so readers can find both in one place. `website/src/feature-status.md` says its table stays in sync with this README, so please add the new row there as well. Otherwise the two pages won't match. This is a preference, not a hard rule, because the README has other usage sections too. ########## table/snapshot_file_diff.go: ########## @@ -0,0 +1,273 @@ +// 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 table + +import ( + "cmp" + "context" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// SnapshotFileDiff describes the physical file changes between two snapshots. +// Data and delete files are reported separately; position-delete files and +// deletion vectors are both included in the delete-file slices. Review Comment: why omit equality deletes? ########## table/snapshot_file_diff_internal_test.go: ########## @@ -0,0 +1,461 @@ +// 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 table + +import ( + "bytes" + "context" + "errors" + "fmt" + "strconv" + "testing" + "time" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + "github.com/stretchr/testify/require" +) + +func diffTestFile(t *testing.T, path string, content iceberg.ManifestEntryContent) iceberg.DataFile { + t.Helper() + builder, err := iceberg.NewDataFileBuilder( + *iceberg.UnpartitionedSpec, content, path, iceberg.ParquetFile, + nil, nil, nil, 10, 1024) + require.NoError(t, err) + + return builder.Build() +} + +func writeDiffTestManifest( + t *testing.T, fs iceio.WriteFileIO, version int, path string, + content iceberg.ManifestContent, snapshotID int64, + added, existing, deleted []iceberg.DataFile, +) iceberg.ManifestFile { + t.Helper() + var buf bytes.Buffer + wr, err := iceberg.NewManifestWriter( + version, &buf, *iceberg.UnpartitionedSpec, simpleSchema(), snapshotID, + iceberg.WithManifestWriterContent(content)) + require.NoError(t, err) + + seq := int64(1) + for _, file := range added { + require.NoError(t, wr.Add(iceberg.NewManifestEntry( + iceberg.EntryStatusADDED, &snapshotID, nil, nil, file))) + } + for _, file := range existing { + var dataSeq, fileSeq *int64 + if version > 1 { + dataSeq, fileSeq = &seq, &seq + } + require.NoError(t, wr.Existing(iceberg.NewManifestEntry( + iceberg.EntryStatusEXISTING, &snapshotID, dataSeq, fileSeq, file))) + } + for _, file := range deleted { + var dataSeq, fileSeq *int64 + if version > 1 { + dataSeq, fileSeq = &seq, &seq + } + require.NoError(t, wr.Delete(iceberg.NewManifestEntry( + iceberg.EntryStatusDELETED, &snapshotID, dataSeq, fileSeq, file))) + } + require.NoError(t, wr.Close()) + require.NoError(t, fs.WriteFile(context.Background(), path, buf.Bytes())) + manifest, err := wr.ToManifestFile( + path, int64(buf.Len()), iceberg.WithManifestFileContent(content)) + require.NoError(t, err) + + return manifest +} + +func writeDiffTestManifestList( + t *testing.T, fs iceio.WriteFileIO, path string, + snapshotID, sequenceNumber int64, manifests []iceberg.ManifestFile, +) string { + t.Helper() + out, err := fs.Create(context.Background(), path) + require.NoError(t, err) + require.NoError(t, iceberg.WriteManifestList(2, out, snapshotID, nil, + &sequenceNumber, 0, manifests)) + require.NoError(t, out.Close()) + + return path +} + +func diffTestTable( + t *testing.T, version int, fs iceio.IO, snapshots ...Snapshot, +) *Table { + t.Helper() + spec := *iceberg.UnpartitionedSpec + meta, err := NewMetadata(simpleSchema(), &spec, UnsortedSortOrder, + "mem://diff/table", iceberg.Properties{PropertyFormatVersion: strconv.Itoa(version)}) + require.NoError(t, err) + builder, err := MetadataBuilderFromBase(meta, "") + require.NoError(t, err) + + for i := range snapshots { + require.NoError(t, builder.AddSnapshot(&snapshots[i])) + } + require.NotEmpty(t, snapshots) + require.NoError(t, builder.SetSnapshotRef( + MainBranch, snapshots[len(snapshots)-1].SnapshotID, BranchRef)) + meta, err = builder.Build() + require.NoError(t, err) + + return New(Identifier{"db", "diff"}, meta, "mem://diff/table/metadata.json", + func(context.Context) (iceio.IO, error) { return fs, nil }, nil) +} + +func diffFilePaths(files []iceberg.DataFile) []string { + paths := make([]string, 0, len(files)) + for _, file := range files { + paths = append(paths, file.FilePath()) + } + + return paths +} + +func TestDiffSnapshotFilesV1EmbeddedManifests(t *testing.T) { + fs := iceio.NewMemFS() + prefix := "mem://diff/v1/" + a := diffTestFile(t, prefix+"a.parquet", iceberg.EntryContentData) + b := diffTestFile(t, prefix+"b.parquet", iceberg.EntryContentData) + c := diffTestFile(t, prefix+"c.parquet", iceberg.EntryContentData) + ghost := diffTestFile(t, prefix+"ghost.parquet", iceberg.EntryContentData) + + shared := writeDiffTestManifest(t, fs, 1, prefix+"shared.avro", + iceberg.ManifestContentData, 1, []iceberg.DataFile{a}, nil, nil) + before := writeDiffTestManifest(t, fs, 1, prefix+"before.avro", + iceberg.ManifestContentData, 1, []iceberg.DataFile{b}, nil, nil) + after := writeDiffTestManifest(t, fs, 1, prefix+"after.avro", + iceberg.ManifestContentData, 2, []iceberg.DataFile{c}, nil, []iceberg.DataFile{ghost}) + + now := time.Now().UnixMilli() + first := Snapshot{ + SnapshotID: 1, TimestampMs: now, + ManifestLocations: []string{shared.FilePath(), before.FilePath()}, + } + firstID := first.SnapshotID + second := Snapshot{ + SnapshotID: 2, TimestampMs: now + 1, ParentSnapshotID: &firstID, + ManifestLocations: []string{shared.FilePath(), after.FilePath()}, + } + tbl := diffTestTable(t, 1, fs, first, second) + + diff, err := tbl.DiffSnapshotFiles(context.Background(), 1, 2) + require.NoError(t, err) + require.True(t, diff.HasChanges()) + require.Equal(t, int64(1), diff.FromSnapshotID) + require.Equal(t, int64(2), diff.ToSnapshotID) + require.Equal(t, []string{c.FilePath()}, diffFilePaths(diff.AddedDataFiles)) + require.Equal(t, []string{b.FilePath()}, diffFilePaths(diff.RemovedDataFiles)) + require.Empty(t, diff.AddedDeleteFiles) + require.Empty(t, diff.RemovedDeleteFiles) + + reverse, err := tbl.DiffSnapshotFiles(context.Background(), 2, 1) + require.NoError(t, err) + require.Equal(t, []string{b.FilePath()}, diffFilePaths(reverse.AddedDataFiles)) + require.Equal(t, []string{c.FilePath()}, diffFilePaths(reverse.RemovedDataFiles)) +} + +func writeDiffTestManifestListV1( + t *testing.T, fs iceio.WriteFileIO, path string, + snapshotID int64, manifests []iceberg.ManifestFile, +) string { + t.Helper() + out, err := fs.Create(context.Background(), path) + require.NoError(t, err) + require.NoError(t, iceberg.WriteManifestList(1, out, snapshotID, nil, nil, 0, manifests)) + require.NoError(t, out.Close()) + + return path +} + +func TestDiffSnapshotFilesV1ManifestLists(t *testing.T) { + fs := iceio.NewMemFS() + prefix := "mem://diff/v1-lists/" + reused := diffTestFile(t, prefix+"reused.parquet", iceberg.EntryContentData) + before := diffTestFile(t, prefix+"before.parquet", iceberg.EntryContentData) + after := diffTestFile(t, prefix+"after.parquet", iceberg.EntryContentData) + + shared := writeDiffTestManifest(t, fs, 1, prefix+"shared.avro", + iceberg.ManifestContentData, 1, []iceberg.DataFile{reused}, nil, nil) + oldManifest := writeDiffTestManifest(t, fs, 1, prefix+"old.avro", + iceberg.ManifestContentData, 1, []iceberg.DataFile{before}, nil, nil) + newManifest := writeDiffTestManifest(t, fs, 1, prefix+"new.avro", + iceberg.ManifestContentData, 2, nil, []iceberg.DataFile{after}, nil) + fromList := writeDiffTestManifestListV1(t, fs, prefix+"before-list.avro", 1, + []iceberg.ManifestFile{shared, oldManifest}) + toList := writeDiffTestManifestListV1(t, fs, prefix+"after-list.avro", 2, + []iceberg.ManifestFile{shared, newManifest}) + now := time.Now().UnixMilli() + parent := int64(1) + tbl := diffTestTable(t, 1, fs, + Snapshot{SnapshotID: 1, TimestampMs: now, ManifestList: fromList}, + Snapshot{SnapshotID: 2, TimestampMs: now + 1, ParentSnapshotID: &parent, ManifestList: toList}) + + diff, err := tbl.DiffSnapshotFiles(context.Background(), 1, 2) + require.NoError(t, err) + require.Equal(t, []string{after.FilePath()}, diffFilePaths(diff.AddedDataFiles)) + require.Equal(t, []string{before.FilePath()}, diffFilePaths(diff.RemovedDataFiles)) + require.Empty(t, diff.AddedDeleteFiles) + require.Empty(t, diff.RemovedDeleteFiles) +} + +func TestDiffSnapshotFilesV2ManifestRewriteAndDeleteFiles(t *testing.T) { + fs := iceio.NewMemFS() + prefix := "mem://diff/v2/" + moved := diffTestFile(t, prefix+"moved.parquet", iceberg.EntryContentData) + added := diffTestFile(t, prefix+"added.parquet", iceberg.EntryContentData) + removedDelete := diffTestFile(t, prefix+"removed-delete.parquet", iceberg.EntryContentPosDeletes) + addedDelete := diffTestFile(t, prefix+"added-delete.parquet", iceberg.EntryContentEqDeletes) + + // Deliberately leave the shared manifest's Avro file absent. Its descriptor + // exists in both lists, so reading it would be an unnecessary I/O error. + shared := iceberg.NewManifestBuilder( + 2, prefix+"unchanged-manifest.avro", 100, 0, 1).SequenceNumber(1, 1).AddedFiles(1).Build() + oldData := writeDiffTestManifest(t, fs, 2, prefix+"old-data.avro", + iceberg.ManifestContentData, 1, []iceberg.DataFile{moved}, nil, nil) + newData := writeDiffTestManifest(t, fs, 2, prefix+"new-data.avro", + iceberg.ManifestContentData, 2, []iceberg.DataFile{added}, + []iceberg.DataFile{moved}, nil) + oldDeletes := writeDiffTestManifest(t, fs, 2, prefix+"old-deletes.avro", + iceberg.ManifestContentDeletes, 1, + []iceberg.DataFile{removedDelete}, nil, nil) + newDeletes := writeDiffTestManifest(t, fs, 2, prefix+"new-deletes.avro", + iceberg.ManifestContentDeletes, 2, + []iceberg.DataFile{addedDelete}, nil, nil) + + // A shared manifest that cannot be opened must not be read during diff. + fromList := writeDiffTestManifestList(t, fs, prefix+"list1.avro", 1, 1, + []iceberg.ManifestFile{shared, oldData, oldDeletes}) + toList := writeDiffTestManifestList(t, fs, prefix+"list2.avro", 2, 2, + []iceberg.ManifestFile{newDeletes, shared, newData}) + now := time.Now().UnixMilli() + fromID := int64(1) + tbl := diffTestTable(t, 2, fs, + Snapshot{SnapshotID: 1, SequenceNumber: 1, TimestampMs: now, ManifestList: fromList}, + Snapshot{ + SnapshotID: 2, SequenceNumber: 2, TimestampMs: now + 1, + ParentSnapshotID: &fromID, ManifestList: toList, + }) + + diff, err := tbl.DiffSnapshotFiles(context.Background(), 1, 2) + require.NoError(t, err) + require.Equal(t, []string{added.FilePath()}, diffFilePaths(diff.AddedDataFiles)) + require.Empty(t, diff.RemovedDataFiles, "manifest-only rewrite does not remove a data file") + require.Equal(t, []string{addedDelete.FilePath()}, diffFilePaths(diff.AddedDeleteFiles)) + require.Equal(t, []string{removedDelete.FilePath()}, diffFilePaths(diff.RemovedDeleteFiles)) +} + +func writeDiffTestManifestListV3( + t *testing.T, fs iceio.WriteFileIO, path string, + snapshotID, sequenceNumber int64, manifests []iceberg.ManifestFile, +) string { + t.Helper() + out, err := fs.Create(context.Background(), path) + require.NoError(t, err) + require.NoError(t, iceberg.WriteManifestList(3, out, snapshotID, nil, + &sequenceNumber, 0, manifests)) + require.NoError(t, out.Close()) + + return path +} + +func diffTestDeletionVector( + t *testing.T, puffinPath, dataPath string, offset int64, +) iceberg.DataFile { + t.Helper() + builder, err := iceberg.NewDataFileBuilder( + *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes, + puffinPath, iceberg.PuffinFile, nil, nil, nil, 10, 1024) + require.NoError(t, err) + + return builder. + ReferencedDataFile(dataPath). + ContentOffset(offset). + ContentSizeInBytes(32). + Build() +} + +func TestDiffSnapshotFilesV3DeletionVectorsSharePuffinFile(t *testing.T) { + fs := iceio.NewMemFS() + prefix := "mem://diff/v3/" + puffin := prefix + "deletes.puffin" + oldA := diffTestDeletionVector(t, puffin, prefix+"a.parquet", 100) + newA := diffTestDeletionVector(t, puffin, prefix+"a.parquet", 200) + sameB := diffTestDeletionVector(t, puffin, prefix+"b.parquet", 300) + + // Two DVs can live in one Puffin file. A replacement must not be hidden + // by path-only deduplication, nor should the retained DV be reported. + oldManifest := writeDiffTestManifest(t, fs, 3, prefix+"old.avro", + iceberg.ManifestContentDeletes, 1, []iceberg.DataFile{oldA, sameB}, nil, nil) + newManifest := writeDiffTestManifest(t, fs, 3, prefix+"new.avro", + iceberg.ManifestContentDeletes, 2, []iceberg.DataFile{newA, sameB}, nil, nil) + + fromList := writeDiffTestManifestListV3(t, fs, prefix+"from-list.avro", 1, 1, + []iceberg.ManifestFile{oldManifest}) + toList := writeDiffTestManifestListV3(t, fs, prefix+"to-list.avro", 2, 2, + []iceberg.ManifestFile{newManifest}) + now := time.Now().UnixMilli() + parent, zero := int64(1), int64(0) + tbl := diffTestTable(t, 3, fs, + Snapshot{ + SnapshotID: 1, SequenceNumber: 1, TimestampMs: now, ManifestList: fromList, + FirstRowID: &zero, AddedRows: &zero, + }, + Snapshot{ + SnapshotID: 2, SequenceNumber: 2, TimestampMs: now + 1, + ParentSnapshotID: &parent, ManifestList: toList, FirstRowID: &zero, AddedRows: &zero, + }) + + diff, err := tbl.DiffSnapshotFiles(context.Background(), 1, 2) + require.NoError(t, err) + require.True(t, diff.HasChanges()) + require.Empty(t, diff.AddedDataFiles) + require.Empty(t, diff.RemovedDataFiles) + require.Len(t, diff.AddedDeleteFiles, 1) + require.Len(t, diff.RemovedDeleteFiles, 1) + require.Equal(t, puffin, diff.AddedDeleteFiles[0].FilePath()) + require.Equal(t, int64(200), *diff.AddedDeleteFiles[0].ContentOffset()) + require.Equal(t, puffin, diff.RemovedDeleteFiles[0].FilePath()) + require.Equal(t, int64(100), *diff.RemovedDeleteFiles[0].ContentOffset()) +} + +func TestDiffSnapshotFilesDifferentBranchesAndSorting(t *testing.T) { Review Comment: This test has no branches. `diffTestTable` only sets main, so these are just two unrelated snapshots. All manifests in this file are hand-written too, so nothing tests what real commits produce. Branches and manifest rewrites are the main use case of this PR. I'd rename this test, and add one with real commits: `NewTransactionOnBranch`, `Append`, `Delete`, `RewriteManifests`. Then check the diff between the branch and main. That test will also guard against future breakage. ########## table/snapshot_file_diff_internal_test.go: ########## @@ -0,0 +1,461 @@ +// 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 table + +import ( + "bytes" + "context" + "errors" + "fmt" + "strconv" + "testing" + "time" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + "github.com/stretchr/testify/require" +) + +func diffTestFile(t *testing.T, path string, content iceberg.ManifestEntryContent) iceberg.DataFile { + t.Helper() + builder, err := iceberg.NewDataFileBuilder( + *iceberg.UnpartitionedSpec, content, path, iceberg.ParquetFile, + nil, nil, nil, 10, 1024) + require.NoError(t, err) + + return builder.Build() +} + +func writeDiffTestManifest( + t *testing.T, fs iceio.WriteFileIO, version int, path string, + content iceberg.ManifestContent, snapshotID int64, + added, existing, deleted []iceberg.DataFile, +) iceberg.ManifestFile { + t.Helper() + var buf bytes.Buffer + wr, err := iceberg.NewManifestWriter( + version, &buf, *iceberg.UnpartitionedSpec, simpleSchema(), snapshotID, + iceberg.WithManifestWriterContent(content)) + require.NoError(t, err) + + seq := int64(1) + for _, file := range added { + require.NoError(t, wr.Add(iceberg.NewManifestEntry( + iceberg.EntryStatusADDED, &snapshotID, nil, nil, file))) + } + for _, file := range existing { + var dataSeq, fileSeq *int64 + if version > 1 { + dataSeq, fileSeq = &seq, &seq + } + require.NoError(t, wr.Existing(iceberg.NewManifestEntry( + iceberg.EntryStatusEXISTING, &snapshotID, dataSeq, fileSeq, file))) + } + for _, file := range deleted { + var dataSeq, fileSeq *int64 + if version > 1 { + dataSeq, fileSeq = &seq, &seq + } + require.NoError(t, wr.Delete(iceberg.NewManifestEntry( + iceberg.EntryStatusDELETED, &snapshotID, dataSeq, fileSeq, file))) + } + require.NoError(t, wr.Close()) + require.NoError(t, fs.WriteFile(context.Background(), path, buf.Bytes())) + manifest, err := wr.ToManifestFile( + path, int64(buf.Len()), iceberg.WithManifestFileContent(content)) + require.NoError(t, err) + + return manifest +} + +func writeDiffTestManifestList( Review Comment: Please merge these helpers into one, they are just copies of one another: `writeDiffTestManifestList`, `writeDiffTestManifestListV1`, `writeDiffTestManifestListV3` ########## table/snapshot_file_diff.go: ########## @@ -0,0 +1,273 @@ +// 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 table + +import ( + "cmp" + "context" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// SnapshotFileDiff describes the physical file changes between two snapshots. +// Data and delete files are reported separately; position-delete files and +// deletion vectors are both included in the delete-file slices. +// +// A rewrite that replaces data files may change these slices even when the +// logical table rows are unchanged. This is a file-level diff, not a row diff. +type SnapshotFileDiff struct { + FromSnapshotID int64 + ToSnapshotID int64 + + AddedDataFiles []iceberg.DataFile + RemovedDataFiles []iceberg.DataFile + AddedDeleteFiles []iceberg.DataFile + RemovedDeleteFiles []iceberg.DataFile +} + +// HasChanges reports whether either snapshot references physical files the +// other does not. +func (d SnapshotFileDiff) HasChanges() bool { + return len(d.AddedDataFiles)+len(d.RemovedDataFiles)+ + len(d.AddedDeleteFiles)+len(d.RemovedDeleteFiles) != 0 +} + +// DiffSnapshotFiles compares the live files referenced by any two snapshots +// still present in the table's metadata. The snapshots need not be ancestors; +// for example, they may belong to different branches. The order of the +// arguments determines the direction of additions and removals. +// +// Iceberg manifests are immutable. Shared manifests therefore contain +// identical live entries and can be skipped without opening their Avro files. +// For the remaining manifests, ADDED and EXISTING entries are live; DELETED +// entries are tombstones and must not be included. Data-file paths identify +// immutable content, so a file rewritten into a different manifest does not +// count as added or removed. In v3, deletion vectors use the Puffin +// location, blob offset, and blob length as their identity. +// +// This works for v1 snapshots that embed manifest paths, as well as v1-v3 +// snapshots with manifest lists. It reads only metadata and never reads the +// Parquet, Avro, or ORC data files. +func (t Table) DiffSnapshotFiles(ctx context.Context, fromSnapshotID, toSnapshotID int64) (SnapshotFileDiff, error) { + diff := SnapshotFileDiff{ + FromSnapshotID: fromSnapshotID, + ToSnapshotID: toSnapshotID, + } + if err := ctx.Err(); err != nil { + return diff, err + } + if t.metadata == nil { + return diff, fmt.Errorf("%w: table metadata is not available", ErrInvalidMetadata) + } + + from := t.metadata.SnapshotByID(fromSnapshotID) + if from == nil { + return diff, fmt.Errorf("%w: from snapshot %d", ErrSnapshotNotFound, fromSnapshotID) + } + to := t.metadata.SnapshotByID(toSnapshotID) + if to == nil { + return diff, fmt.Errorf("%w: to snapshot %d", ErrSnapshotNotFound, toSnapshotID) + } + if fromSnapshotID == toSnapshotID { + return diff, nil + } + if t.fsF == nil { + return diff, fmt.Errorf("%w: table file IO is not configured", ErrInvalidOperation) + } + + fs, err := t.fsF(ctx) + if err != nil { + return diff, fmt.Errorf("resolve table file IO: %w", err) + } + if fs == nil { + return diff, fmt.Errorf("%w: table file IO is nil", ErrInvalidOperation) + } + // Share the resolved FileIO between the two manifest-list loads and the + // changed manifest entries. Keep the normal bounded manifest-list cache. + fsF := func(context.Context) (iceio.IO, error) { return fs, nil } + + fromSet, err := t.manifestSetWithFSF(ctx, *from, fsF) + if err != nil { + return diff, fmt.Errorf("read from snapshot %d manifests: %w", fromSnapshotID, err) + } + toSet, err := t.manifestSetWithFSF(ctx, *to, fsF) + if err != nil { + return diff, fmt.Errorf("read to snapshot %d manifests: %w", toSnapshotID, err) + } + + fromOnly, toOnly := uniqueSnapshotManifests(fromSet.borrowAllManifests(), toSet.borrowAllManifests()) + if len(fromOnly) == 0 && len(toOnly) == 0 { + return diff, nil + } + + fromFiles, err := snapshotDiffLiveFiles(ctx, fs, fromOnly) + if err != nil { + return diff, fmt.Errorf("read from snapshot %d entries: %w", fromSnapshotID, err) + } + toFiles, err := snapshotDiffLiveFiles(ctx, fs, toOnly) + if err != nil { + return diff, fmt.Errorf("read to snapshot %d entries: %w", toSnapshotID, err) + } + + // A manifest rewrite can move the same data file into a different + // manifest without modifying table contents. Cancel those paths out. + for key, fromFile := range fromFiles { + if toFile, ok := toFiles[key]; ok { + if fromFile.ContentType() != toFile.ContentType() { + return diff, fmt.Errorf("%w: content file %q changed type between snapshots", + ErrInvalidMetadata, key.path) + } + delete(fromFiles, key) + delete(toFiles, key) + } + } + + for _, file := range fromFiles { + if file.ContentType() == iceberg.EntryContentData { + diff.RemovedDataFiles = append(diff.RemovedDataFiles, file) + } else { + diff.RemovedDeleteFiles = append(diff.RemovedDeleteFiles, file) + } + } + for _, file := range toFiles { + if file.ContentType() == iceberg.EntryContentData { + diff.AddedDataFiles = append(diff.AddedDataFiles, file) + } else { + diff.AddedDeleteFiles = append(diff.AddedDeleteFiles, file) + } + } + + // Maps have unstable iteration order. Stable paths make the API useful + // for reproducible audit logs and change-detection checkpoints. + sortSnapshotDiffFiles(diff.AddedDataFiles) + sortSnapshotDiffFiles(diff.RemovedDataFiles) + sortSnapshotDiffFiles(diff.AddedDeleteFiles) + sortSnapshotDiffFiles(diff.RemovedDeleteFiles) + + return diff, nil +} + +// uniqueSnapshotManifests skips immutable manifests that are present in both +// snapshots. No ancestry walk is needed, and different branches work. +func uniqueSnapshotManifests(from, to []iceberg.ManifestFile) (fromOnly, toOnly []iceberg.ManifestFile) { + fromPaths := make(map[string]struct{}, len(from)) + toPaths := make(map[string]struct{}, len(to)) + for _, manifest := range from { + fromPaths[manifest.FilePath()] = struct{}{} + } + for _, manifest := range to { + toPaths[manifest.FilePath()] = struct{}{} + if _, shared := fromPaths[manifest.FilePath()]; !shared { + toOnly = append(toOnly, manifest) + } + } + for _, manifest := range from { + if _, shared := toPaths[manifest.FilePath()]; !shared { + fromOnly = append(fromOnly, manifest) + } + } + + return fromOnly, toOnly +} + +// snapshotDiffFileKey uses the spec's identity for content files. A v3 Puffin +// file may hold several different deletion vector blobs, so its path alone +// is not enough to identify the vector. +type snapshotDiffFileKey struct { + path string + vector bool + offset int64 + length int64 +} + +func snapshotFileDiffKey(file iceberg.DataFile) (snapshotDiffFileKey, error) { + key := snapshotDiffFileKey{path: file.FilePath()} + if key.path == "" { + return key, fmt.Errorf("%w: content file path is empty", ErrInvalidMetadata) + } + if IsDeletionVector(file) { + offset := file.ContentOffset() + length := file.ContentSizeInBytes() + if offset == nil || length == nil || *offset < 0 || *length <= 0 { + return key, fmt.Errorf("%w: deletion vector %q has an invalid offset or length", + ErrInvalidMetadata, file.FilePath()) + } + key.vector = true + key.offset = *offset + key.length = *length + } + + return key, nil +} + +func snapshotDiffLiveFiles(ctx context.Context, fs iceio.IO, manifests []iceberg.ManifestFile) (map[snapshotDiffFileKey]iceberg.DataFile, error) { + files := make(map[snapshotDiffFileKey]iceberg.DataFile) + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + return nil, err + } + for entry, err := range manifest.Entries(fs, true) { Review Comment: Also, this keeps all column stats in memory for both snapshots, but the diff only needs path, offset and length. On a big table, a manifest rewrite can use many GB of memory just to return an empty diff. `iceberg.EntriesWithProjection(fs, manifest, true, iceberg.ManifestEntryProjection{})` skips the stats. If callers don't need them, I'd use that. Note that files without stats can't go to a ManifestWriter, so the doc should say that. If callers do need stats, let's document the memory cost. ########## table/snapshot_file_diff.go: ########## @@ -0,0 +1,273 @@ +// 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 table + +import ( + "cmp" + "context" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// SnapshotFileDiff describes the physical file changes between two snapshots. +// Data and delete files are reported separately; position-delete files and +// deletion vectors are both included in the delete-file slices. +// +// A rewrite that replaces data files may change these slices even when the +// logical table rows are unchanged. This is a file-level diff, not a row diff. +type SnapshotFileDiff struct { + FromSnapshotID int64 + ToSnapshotID int64 + + AddedDataFiles []iceberg.DataFile + RemovedDataFiles []iceberg.DataFile + AddedDeleteFiles []iceberg.DataFile + RemovedDeleteFiles []iceberg.DataFile +} + +// HasChanges reports whether either snapshot references physical files the +// other does not. +func (d SnapshotFileDiff) HasChanges() bool { + return len(d.AddedDataFiles)+len(d.RemovedDataFiles)+ + len(d.AddedDeleteFiles)+len(d.RemovedDeleteFiles) != 0 +} + +// DiffSnapshotFiles compares the live files referenced by any two snapshots +// still present in the table's metadata. The snapshots need not be ancestors; +// for example, they may belong to different branches. The order of the +// arguments determines the direction of additions and removals. +// +// Iceberg manifests are immutable. Shared manifests therefore contain +// identical live entries and can be skipped without opening their Avro files. +// For the remaining manifests, ADDED and EXISTING entries are live; DELETED +// entries are tombstones and must not be included. Data-file paths identify +// immutable content, so a file rewritten into a different manifest does not +// count as added or removed. In v3, deletion vectors use the Puffin +// location, blob offset, and blob length as their identity. +// +// This works for v1 snapshots that embed manifest paths, as well as v1-v3 +// snapshots with manifest lists. It reads only metadata and never reads the +// Parquet, Avro, or ORC data files. +func (t Table) DiffSnapshotFiles(ctx context.Context, fromSnapshotID, toSnapshotID int64) (SnapshotFileDiff, error) { + diff := SnapshotFileDiff{ + FromSnapshotID: fromSnapshotID, + ToSnapshotID: toSnapshotID, + } + if err := ctx.Err(); err != nil { + return diff, err + } + if t.metadata == nil { + return diff, fmt.Errorf("%w: table metadata is not available", ErrInvalidMetadata) + } + + from := t.metadata.SnapshotByID(fromSnapshotID) + if from == nil { + return diff, fmt.Errorf("%w: from snapshot %d", ErrSnapshotNotFound, fromSnapshotID) + } + to := t.metadata.SnapshotByID(toSnapshotID) + if to == nil { + return diff, fmt.Errorf("%w: to snapshot %d", ErrSnapshotNotFound, toSnapshotID) + } + if fromSnapshotID == toSnapshotID { + return diff, nil + } + if t.fsF == nil { + return diff, fmt.Errorf("%w: table file IO is not configured", ErrInvalidOperation) + } + + fs, err := t.fsF(ctx) + if err != nil { + return diff, fmt.Errorf("resolve table file IO: %w", err) + } + if fs == nil { + return diff, fmt.Errorf("%w: table file IO is nil", ErrInvalidOperation) + } + // Share the resolved FileIO between the two manifest-list loads and the + // changed manifest entries. Keep the normal bounded manifest-list cache. + fsF := func(context.Context) (iceio.IO, error) { return fs, nil } + + fromSet, err := t.manifestSetWithFSF(ctx, *from, fsF) + if err != nil { + return diff, fmt.Errorf("read from snapshot %d manifests: %w", fromSnapshotID, err) + } + toSet, err := t.manifestSetWithFSF(ctx, *to, fsF) + if err != nil { + return diff, fmt.Errorf("read to snapshot %d manifests: %w", toSnapshotID, err) + } + + fromOnly, toOnly := uniqueSnapshotManifests(fromSet.borrowAllManifests(), toSet.borrowAllManifests()) + if len(fromOnly) == 0 && len(toOnly) == 0 { + return diff, nil + } + + fromFiles, err := snapshotDiffLiveFiles(ctx, fs, fromOnly) + if err != nil { + return diff, fmt.Errorf("read from snapshot %d entries: %w", fromSnapshotID, err) + } + toFiles, err := snapshotDiffLiveFiles(ctx, fs, toOnly) + if err != nil { + return diff, fmt.Errorf("read to snapshot %d entries: %w", toSnapshotID, err) + } + + // A manifest rewrite can move the same data file into a different + // manifest without modifying table contents. Cancel those paths out. + for key, fromFile := range fromFiles { + if toFile, ok := toFiles[key]; ok { + if fromFile.ContentType() != toFile.ContentType() { + return diff, fmt.Errorf("%w: content file %q changed type between snapshots", + ErrInvalidMetadata, key.path) + } + delete(fromFiles, key) + delete(toFiles, key) + } + } + + for _, file := range fromFiles { + if file.ContentType() == iceberg.EntryContentData { + diff.RemovedDataFiles = append(diff.RemovedDataFiles, file) + } else { + diff.RemovedDeleteFiles = append(diff.RemovedDeleteFiles, file) + } + } + for _, file := range toFiles { + if file.ContentType() == iceberg.EntryContentData { + diff.AddedDataFiles = append(diff.AddedDataFiles, file) + } else { + diff.AddedDeleteFiles = append(diff.AddedDeleteFiles, file) + } + } + + // Maps have unstable iteration order. Stable paths make the API useful + // for reproducible audit logs and change-detection checkpoints. + sortSnapshotDiffFiles(diff.AddedDataFiles) + sortSnapshotDiffFiles(diff.RemovedDataFiles) + sortSnapshotDiffFiles(diff.AddedDeleteFiles) + sortSnapshotDiffFiles(diff.RemovedDeleteFiles) + + return diff, nil +} + +// uniqueSnapshotManifests skips immutable manifests that are present in both +// snapshots. No ancestry walk is needed, and different branches work. +func uniqueSnapshotManifests(from, to []iceberg.ManifestFile) (fromOnly, toOnly []iceberg.ManifestFile) { + fromPaths := make(map[string]struct{}, len(from)) + toPaths := make(map[string]struct{}, len(to)) + for _, manifest := range from { + fromPaths[manifest.FilePath()] = struct{}{} + } + for _, manifest := range to { + toPaths[manifest.FilePath()] = struct{}{} + if _, shared := fromPaths[manifest.FilePath()]; !shared { + toOnly = append(toOnly, manifest) + } + } + for _, manifest := range from { + if _, shared := toPaths[manifest.FilePath()]; !shared { + fromOnly = append(fromOnly, manifest) + } + } + + return fromOnly, toOnly +} + +// snapshotDiffFileKey uses the spec's identity for content files. A v3 Puffin +// file may hold several different deletion vector blobs, so its path alone +// is not enough to identify the vector. +type snapshotDiffFileKey struct { + path string + vector bool + offset int64 + length int64 +} + +func snapshotFileDiffKey(file iceberg.DataFile) (snapshotDiffFileKey, error) { + key := snapshotDiffFileKey{path: file.FilePath()} + if key.path == "" { + return key, fmt.Errorf("%w: content file path is empty", ErrInvalidMetadata) + } + if IsDeletionVector(file) { + offset := file.ContentOffset() + length := file.ContentSizeInBytes() + if offset == nil || length == nil || *offset < 0 || *length <= 0 { + return key, fmt.Errorf("%w: deletion vector %q has an invalid offset or length", + ErrInvalidMetadata, file.FilePath()) + } + key.vector = true + key.offset = *offset + key.length = *length + } + + return key, nil +} + +func snapshotDiffLiveFiles(ctx context.Context, fs iceio.IO, manifests []iceberg.ManifestFile) (map[snapshotDiffFileKey]iceberg.DataFile, error) { + files := make(map[snapshotDiffFileKey]iceberg.DataFile) + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + return nil, err + } + for entry, err := range manifest.Entries(fs, true) { + if err != nil { + return nil, fmt.Errorf("manifest %q: %w", manifest.FilePath(), err) + } + if err := ctx.Err(); err != nil { + return nil, err + } + if entry == nil || entry.DataFile() == nil { + return nil, fmt.Errorf("%w: manifest %q contains a nil data file", ErrInvalidMetadata, manifest.FilePath()) + } + file := entry.DataFile() Review Comment: are we sure that these checks can fire? the manifest iterator never gives a nil entry without an error, and ReadEntry already rejects unknown content types, so the default branch never runs. Better: delete the nil check and the switch, and keep the code inside the case. Two more spots can go too: 1. the t.metadata == nil check at L77, which AllManifests doesn't have 2. the empty check at L117, which gives the same result as continuing. ########## table/snapshot_file_diff.go: ########## @@ -0,0 +1,273 @@ +// 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 table + +import ( + "cmp" + "context" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// SnapshotFileDiff describes the physical file changes between two snapshots. +// Data and delete files are reported separately; position-delete files and +// deletion vectors are both included in the delete-file slices. +// +// A rewrite that replaces data files may change these slices even when the +// logical table rows are unchanged. This is a file-level diff, not a row diff. +type SnapshotFileDiff struct { + FromSnapshotID int64 + ToSnapshotID int64 + + AddedDataFiles []iceberg.DataFile + RemovedDataFiles []iceberg.DataFile + AddedDeleteFiles []iceberg.DataFile + RemovedDeleteFiles []iceberg.DataFile +} + +// HasChanges reports whether either snapshot references physical files the +// other does not. +func (d SnapshotFileDiff) HasChanges() bool { + return len(d.AddedDataFiles)+len(d.RemovedDataFiles)+ + len(d.AddedDeleteFiles)+len(d.RemovedDeleteFiles) != 0 +} + +// DiffSnapshotFiles compares the live files referenced by any two snapshots +// still present in the table's metadata. The snapshots need not be ancestors; +// for example, they may belong to different branches. The order of the +// arguments determines the direction of additions and removals. +// +// Iceberg manifests are immutable. Shared manifests therefore contain +// identical live entries and can be skipped without opening their Avro files. +// For the remaining manifests, ADDED and EXISTING entries are live; DELETED +// entries are tombstones and must not be included. Data-file paths identify +// immutable content, so a file rewritten into a different manifest does not +// count as added or removed. In v3, deletion vectors use the Puffin +// location, blob offset, and blob length as their identity. +// +// This works for v1 snapshots that embed manifest paths, as well as v1-v3 +// snapshots with manifest lists. It reads only metadata and never reads the +// Parquet, Avro, or ORC data files. +func (t Table) DiffSnapshotFiles(ctx context.Context, fromSnapshotID, toSnapshotID int64) (SnapshotFileDiff, error) { + diff := SnapshotFileDiff{ + FromSnapshotID: fromSnapshotID, + ToSnapshotID: toSnapshotID, + } + if err := ctx.Err(); err != nil { + return diff, err + } + if t.metadata == nil { + return diff, fmt.Errorf("%w: table metadata is not available", ErrInvalidMetadata) + } + + from := t.metadata.SnapshotByID(fromSnapshotID) + if from == nil { + return diff, fmt.Errorf("%w: from snapshot %d", ErrSnapshotNotFound, fromSnapshotID) + } + to := t.metadata.SnapshotByID(toSnapshotID) + if to == nil { + return diff, fmt.Errorf("%w: to snapshot %d", ErrSnapshotNotFound, toSnapshotID) + } + if fromSnapshotID == toSnapshotID { + return diff, nil + } + if t.fsF == nil { + return diff, fmt.Errorf("%w: table file IO is not configured", ErrInvalidOperation) + } + + fs, err := t.fsF(ctx) + if err != nil { + return diff, fmt.Errorf("resolve table file IO: %w", err) + } + if fs == nil { + return diff, fmt.Errorf("%w: table file IO is nil", ErrInvalidOperation) + } + // Share the resolved FileIO between the two manifest-list loads and the + // changed manifest entries. Keep the normal bounded manifest-list cache. + fsF := func(context.Context) (iceio.IO, error) { return fs, nil } + + fromSet, err := t.manifestSetWithFSF(ctx, *from, fsF) + if err != nil { + return diff, fmt.Errorf("read from snapshot %d manifests: %w", fromSnapshotID, err) + } + toSet, err := t.manifestSetWithFSF(ctx, *to, fsF) + if err != nil { + return diff, fmt.Errorf("read to snapshot %d manifests: %w", toSnapshotID, err) + } + + fromOnly, toOnly := uniqueSnapshotManifests(fromSet.borrowAllManifests(), toSet.borrowAllManifests()) + if len(fromOnly) == 0 && len(toOnly) == 0 { + return diff, nil + } + + fromFiles, err := snapshotDiffLiveFiles(ctx, fs, fromOnly) + if err != nil { + return diff, fmt.Errorf("read from snapshot %d entries: %w", fromSnapshotID, err) + } + toFiles, err := snapshotDiffLiveFiles(ctx, fs, toOnly) + if err != nil { + return diff, fmt.Errorf("read to snapshot %d entries: %w", toSnapshotID, err) + } + + // A manifest rewrite can move the same data file into a different + // manifest without modifying table contents. Cancel those paths out. + for key, fromFile := range fromFiles { + if toFile, ok := toFiles[key]; ok { + if fromFile.ContentType() != toFile.ContentType() { + return diff, fmt.Errorf("%w: content file %q changed type between snapshots", + ErrInvalidMetadata, key.path) + } + delete(fromFiles, key) + delete(toFiles, key) + } + } + + for _, file := range fromFiles { + if file.ContentType() == iceberg.EntryContentData { + diff.RemovedDataFiles = append(diff.RemovedDataFiles, file) + } else { + diff.RemovedDeleteFiles = append(diff.RemovedDeleteFiles, file) + } + } + for _, file := range toFiles { + if file.ContentType() == iceberg.EntryContentData { + diff.AddedDataFiles = append(diff.AddedDataFiles, file) + } else { + diff.AddedDeleteFiles = append(diff.AddedDeleteFiles, file) + } + } + + // Maps have unstable iteration order. Stable paths make the API useful + // for reproducible audit logs and change-detection checkpoints. + sortSnapshotDiffFiles(diff.AddedDataFiles) + sortSnapshotDiffFiles(diff.RemovedDataFiles) + sortSnapshotDiffFiles(diff.AddedDeleteFiles) + sortSnapshotDiffFiles(diff.RemovedDeleteFiles) + + return diff, nil +} + +// uniqueSnapshotManifests skips immutable manifests that are present in both +// snapshots. No ancestry walk is needed, and different branches work. +func uniqueSnapshotManifests(from, to []iceberg.ManifestFile) (fromOnly, toOnly []iceberg.ManifestFile) { + fromPaths := make(map[string]struct{}, len(from)) + toPaths := make(map[string]struct{}, len(to)) + for _, manifest := range from { + fromPaths[manifest.FilePath()] = struct{}{} + } + for _, manifest := range to { + toPaths[manifest.FilePath()] = struct{}{} + if _, shared := fromPaths[manifest.FilePath()]; !shared { + toOnly = append(toOnly, manifest) + } + } + for _, manifest := range from { + if _, shared := toPaths[manifest.FilePath()]; !shared { + fromOnly = append(fromOnly, manifest) + } + } + + return fromOnly, toOnly +} + +// snapshotDiffFileKey uses the spec's identity for content files. A v3 Puffin +// file may hold several different deletion vector blobs, so its path alone +// is not enough to identify the vector. +type snapshotDiffFileKey struct { + path string + vector bool + offset int64 + length int64 +} Review Comment: vector isn't needed. A valid dv always has length above 0, and every other file has length 0, so two keys can't clash without it. The ContentType step in the sort can go for the same reason. the struct then has the same fields as deletionVectorBlobKey in transaction.go, so I'd reuse that type, and rename it if the name reads wrong for data files. #2397 adds another copy of this key, so let's keep just one. ########## table/snapshot_file_diff_internal_test.go: ########## @@ -0,0 +1,461 @@ +// 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 table + +import ( + "bytes" + "context" + "errors" + "fmt" + "strconv" + "testing" + "time" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + "github.com/stretchr/testify/require" +) + +func diffTestFile(t *testing.T, path string, content iceberg.ManifestEntryContent) iceberg.DataFile { + t.Helper() + builder, err := iceberg.NewDataFileBuilder( + *iceberg.UnpartitionedSpec, content, path, iceberg.ParquetFile, + nil, nil, nil, 10, 1024) + require.NoError(t, err) + + return builder.Build() +} + +func writeDiffTestManifest( + t *testing.T, fs iceio.WriteFileIO, version int, path string, + content iceberg.ManifestContent, snapshotID int64, + added, existing, deleted []iceberg.DataFile, +) iceberg.ManifestFile { + t.Helper() + var buf bytes.Buffer + wr, err := iceberg.NewManifestWriter( + version, &buf, *iceberg.UnpartitionedSpec, simpleSchema(), snapshotID, + iceberg.WithManifestWriterContent(content)) + require.NoError(t, err) + + seq := int64(1) + for _, file := range added { + require.NoError(t, wr.Add(iceberg.NewManifestEntry( + iceberg.EntryStatusADDED, &snapshotID, nil, nil, file))) + } + for _, file := range existing { + var dataSeq, fileSeq *int64 + if version > 1 { + dataSeq, fileSeq = &seq, &seq + } + require.NoError(t, wr.Existing(iceberg.NewManifestEntry( + iceberg.EntryStatusEXISTING, &snapshotID, dataSeq, fileSeq, file))) + } + for _, file := range deleted { + var dataSeq, fileSeq *int64 + if version > 1 { + dataSeq, fileSeq = &seq, &seq + } + require.NoError(t, wr.Delete(iceberg.NewManifestEntry( + iceberg.EntryStatusDELETED, &snapshotID, dataSeq, fileSeq, file))) + } + require.NoError(t, wr.Close()) + require.NoError(t, fs.WriteFile(context.Background(), path, buf.Bytes())) + manifest, err := wr.ToManifestFile( + path, int64(buf.Len()), iceberg.WithManifestFileContent(content)) + require.NoError(t, err) + + return manifest +} + +func writeDiffTestManifestList( + t *testing.T, fs iceio.WriteFileIO, path string, + snapshotID, sequenceNumber int64, manifests []iceberg.ManifestFile, +) string { + t.Helper() + out, err := fs.Create(context.Background(), path) + require.NoError(t, err) + require.NoError(t, iceberg.WriteManifestList(2, out, snapshotID, nil, + &sequenceNumber, 0, manifests)) + require.NoError(t, out.Close()) + + return path +} + +func diffTestTable( + t *testing.T, version int, fs iceio.IO, snapshots ...Snapshot, +) *Table { + t.Helper() + spec := *iceberg.UnpartitionedSpec + meta, err := NewMetadata(simpleSchema(), &spec, UnsortedSortOrder, + "mem://diff/table", iceberg.Properties{PropertyFormatVersion: strconv.Itoa(version)}) + require.NoError(t, err) + builder, err := MetadataBuilderFromBase(meta, "") + require.NoError(t, err) + + for i := range snapshots { + require.NoError(t, builder.AddSnapshot(&snapshots[i])) + } + require.NotEmpty(t, snapshots) + require.NoError(t, builder.SetSnapshotRef( + MainBranch, snapshots[len(snapshots)-1].SnapshotID, BranchRef)) + meta, err = builder.Build() + require.NoError(t, err) + + return New(Identifier{"db", "diff"}, meta, "mem://diff/table/metadata.json", + func(context.Context) (iceio.IO, error) { return fs, nil }, nil) +} + +func diffFilePaths(files []iceberg.DataFile) []string { + paths := make([]string, 0, len(files)) + for _, file := range files { + paths = append(paths, file.FilePath()) + } + + return paths +} + +func TestDiffSnapshotFilesV1EmbeddedManifests(t *testing.T) { + fs := iceio.NewMemFS() + prefix := "mem://diff/v1/" + a := diffTestFile(t, prefix+"a.parquet", iceberg.EntryContentData) + b := diffTestFile(t, prefix+"b.parquet", iceberg.EntryContentData) + c := diffTestFile(t, prefix+"c.parquet", iceberg.EntryContentData) + ghost := diffTestFile(t, prefix+"ghost.parquet", iceberg.EntryContentData) + + shared := writeDiffTestManifest(t, fs, 1, prefix+"shared.avro", + iceberg.ManifestContentData, 1, []iceberg.DataFile{a}, nil, nil) + before := writeDiffTestManifest(t, fs, 1, prefix+"before.avro", + iceberg.ManifestContentData, 1, []iceberg.DataFile{b}, nil, nil) + after := writeDiffTestManifest(t, fs, 1, prefix+"after.avro", + iceberg.ManifestContentData, 2, []iceberg.DataFile{c}, nil, []iceberg.DataFile{ghost}) + + now := time.Now().UnixMilli() + first := Snapshot{ + SnapshotID: 1, TimestampMs: now, + ManifestLocations: []string{shared.FilePath(), before.FilePath()}, + } + firstID := first.SnapshotID + second := Snapshot{ + SnapshotID: 2, TimestampMs: now + 1, ParentSnapshotID: &firstID, + ManifestLocations: []string{shared.FilePath(), after.FilePath()}, + } + tbl := diffTestTable(t, 1, fs, first, second) + + diff, err := tbl.DiffSnapshotFiles(context.Background(), 1, 2) + require.NoError(t, err) + require.True(t, diff.HasChanges()) + require.Equal(t, int64(1), diff.FromSnapshotID) + require.Equal(t, int64(2), diff.ToSnapshotID) + require.Equal(t, []string{c.FilePath()}, diffFilePaths(diff.AddedDataFiles)) + require.Equal(t, []string{b.FilePath()}, diffFilePaths(diff.RemovedDataFiles)) + require.Empty(t, diff.AddedDeleteFiles) + require.Empty(t, diff.RemovedDeleteFiles) + + reverse, err := tbl.DiffSnapshotFiles(context.Background(), 2, 1) + require.NoError(t, err) + require.Equal(t, []string{b.FilePath()}, diffFilePaths(reverse.AddedDataFiles)) + require.Equal(t, []string{c.FilePath()}, diffFilePaths(reverse.RemovedDataFiles)) +} + +func writeDiffTestManifestListV1( + t *testing.T, fs iceio.WriteFileIO, path string, + snapshotID int64, manifests []iceberg.ManifestFile, +) string { + t.Helper() + out, err := fs.Create(context.Background(), path) + require.NoError(t, err) + require.NoError(t, iceberg.WriteManifestList(1, out, snapshotID, nil, nil, 0, manifests)) + require.NoError(t, out.Close()) + + return path +} + +func TestDiffSnapshotFilesV1ManifestLists(t *testing.T) { + fs := iceio.NewMemFS() + prefix := "mem://diff/v1-lists/" + reused := diffTestFile(t, prefix+"reused.parquet", iceberg.EntryContentData) + before := diffTestFile(t, prefix+"before.parquet", iceberg.EntryContentData) + after := diffTestFile(t, prefix+"after.parquet", iceberg.EntryContentData) + + shared := writeDiffTestManifest(t, fs, 1, prefix+"shared.avro", + iceberg.ManifestContentData, 1, []iceberg.DataFile{reused}, nil, nil) + oldManifest := writeDiffTestManifest(t, fs, 1, prefix+"old.avro", + iceberg.ManifestContentData, 1, []iceberg.DataFile{before}, nil, nil) + newManifest := writeDiffTestManifest(t, fs, 1, prefix+"new.avro", + iceberg.ManifestContentData, 2, nil, []iceberg.DataFile{after}, nil) + fromList := writeDiffTestManifestListV1(t, fs, prefix+"before-list.avro", 1, + []iceberg.ManifestFile{shared, oldManifest}) + toList := writeDiffTestManifestListV1(t, fs, prefix+"after-list.avro", 2, + []iceberg.ManifestFile{shared, newManifest}) + now := time.Now().UnixMilli() + parent := int64(1) + tbl := diffTestTable(t, 1, fs, + Snapshot{SnapshotID: 1, TimestampMs: now, ManifestList: fromList}, + Snapshot{SnapshotID: 2, TimestampMs: now + 1, ParentSnapshotID: &parent, ManifestList: toList}) + + diff, err := tbl.DiffSnapshotFiles(context.Background(), 1, 2) + require.NoError(t, err) + require.Equal(t, []string{after.FilePath()}, diffFilePaths(diff.AddedDataFiles)) + require.Equal(t, []string{before.FilePath()}, diffFilePaths(diff.RemovedDataFiles)) + require.Empty(t, diff.AddedDeleteFiles) + require.Empty(t, diff.RemovedDeleteFiles) +} + +func TestDiffSnapshotFilesV2ManifestRewriteAndDeleteFiles(t *testing.T) { + fs := iceio.NewMemFS() + prefix := "mem://diff/v2/" + moved := diffTestFile(t, prefix+"moved.parquet", iceberg.EntryContentData) + added := diffTestFile(t, prefix+"added.parquet", iceberg.EntryContentData) + removedDelete := diffTestFile(t, prefix+"removed-delete.parquet", iceberg.EntryContentPosDeletes) + addedDelete := diffTestFile(t, prefix+"added-delete.parquet", iceberg.EntryContentEqDeletes) + + // Deliberately leave the shared manifest's Avro file absent. Its descriptor + // exists in both lists, so reading it would be an unnecessary I/O error. + shared := iceberg.NewManifestBuilder( + 2, prefix+"unchanged-manifest.avro", 100, 0, 1).SequenceNumber(1, 1).AddedFiles(1).Build() + oldData := writeDiffTestManifest(t, fs, 2, prefix+"old-data.avro", + iceberg.ManifestContentData, 1, []iceberg.DataFile{moved}, nil, nil) + newData := writeDiffTestManifest(t, fs, 2, prefix+"new-data.avro", + iceberg.ManifestContentData, 2, []iceberg.DataFile{added}, + []iceberg.DataFile{moved}, nil) + oldDeletes := writeDiffTestManifest(t, fs, 2, prefix+"old-deletes.avro", + iceberg.ManifestContentDeletes, 1, + []iceberg.DataFile{removedDelete}, nil, nil) + newDeletes := writeDiffTestManifest(t, fs, 2, prefix+"new-deletes.avro", + iceberg.ManifestContentDeletes, 2, + []iceberg.DataFile{addedDelete}, nil, nil) + + // A shared manifest that cannot be opened must not be read during diff. + fromList := writeDiffTestManifestList(t, fs, prefix+"list1.avro", 1, 1, + []iceberg.ManifestFile{shared, oldData, oldDeletes}) + toList := writeDiffTestManifestList(t, fs, prefix+"list2.avro", 2, 2, + []iceberg.ManifestFile{newDeletes, shared, newData}) + now := time.Now().UnixMilli() + fromID := int64(1) + tbl := diffTestTable(t, 2, fs, + Snapshot{SnapshotID: 1, SequenceNumber: 1, TimestampMs: now, ManifestList: fromList}, + Snapshot{ + SnapshotID: 2, SequenceNumber: 2, TimestampMs: now + 1, + ParentSnapshotID: &fromID, ManifestList: toList, + }) + + diff, err := tbl.DiffSnapshotFiles(context.Background(), 1, 2) + require.NoError(t, err) + require.Equal(t, []string{added.FilePath()}, diffFilePaths(diff.AddedDataFiles)) + require.Empty(t, diff.RemovedDataFiles, "manifest-only rewrite does not remove a data file") + require.Equal(t, []string{addedDelete.FilePath()}, diffFilePaths(diff.AddedDeleteFiles)) + require.Equal(t, []string{removedDelete.FilePath()}, diffFilePaths(diff.RemovedDeleteFiles)) +} + +func writeDiffTestManifestListV3( + t *testing.T, fs iceio.WriteFileIO, path string, + snapshotID, sequenceNumber int64, manifests []iceberg.ManifestFile, +) string { + t.Helper() + out, err := fs.Create(context.Background(), path) + require.NoError(t, err) + require.NoError(t, iceberg.WriteManifestList(3, out, snapshotID, nil, + &sequenceNumber, 0, manifests)) + require.NoError(t, out.Close()) + + return path +} + +func diffTestDeletionVector( + t *testing.T, puffinPath, dataPath string, offset int64, +) iceberg.DataFile { + t.Helper() + builder, err := iceberg.NewDataFileBuilder( + *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes, + puffinPath, iceberg.PuffinFile, nil, nil, nil, 10, 1024) + require.NoError(t, err) + + return builder. + ReferencedDataFile(dataPath). + ContentOffset(offset). + ContentSizeInBytes(32). + Build() +} + +func TestDiffSnapshotFilesV3DeletionVectorsSharePuffinFile(t *testing.T) { Review Comment: No test checks the order of dvs in one Puffin file, though one file can hold many dvs and the comment promises stable output. Here, each side has only one changed dv, so there is nothing to order. A later cleanup could break the sort without anyone noticing. I'd add a snapshot with three dvs in one puffin file, written out of order, and check that the offsets come back as 100, 200, 300. ########## README.md: ########## @@ -119,6 +119,34 @@ Tests requiring Spark 4 (currently the variant and unknown-type tests) automatic | Create New Manifests | X | | Plan Scan | Local + Remote | | Plan Scan for Snapshot | Local + Remote | +| Diff Snapshot Files | X | + +### Snapshot file diffs (v1-v3) + +Use `Table.DiffSnapshotFiles` to compare the live physical files referenced by +any two snapshots, including snapshots on different branches: + +```go +diff, err := tbl.DiffSnapshotFiles(ctx, fromSnapshotID, toSnapshotID) +if err != nil { + return err +} +for _, file := range diff.AddedDataFiles { + fmt.Println("added:", file.FilePath()) +} +for _, file := range diff.RemovedDataFiles { + fmt.Println("removed:", file.FilePath()) +} +``` + +The result also includes `AddedDeleteFiles` and `RemovedDeleteFiles` for +v2/v3 tables. It reads manifest metadata only, skips immutable manifests +referenced by both snapshots, and excludes deleted manifest entries. In v1, +snapshots may store either an embedded `manifests` array or a `manifest-list`. +This reports changes in physical files, **not** logical row changes from +compaction or merge-on-read deletes. See the +[Iceberg snapshot specification](https://iceberg.apache.org/spec/#snapshots) +and [manifest entry specification](https://iceberg.apache.org/spec/#manifest-files). Review Comment: broken anchor, `#manifest-files` absent on live spec page, use `#manifests` ########## table/snapshot_file_diff.go: ########## @@ -0,0 +1,273 @@ +// 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 table + +import ( + "cmp" + "context" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// SnapshotFileDiff describes the physical file changes between two snapshots. +// Data and delete files are reported separately; position-delete files and +// deletion vectors are both included in the delete-file slices. +// +// A rewrite that replaces data files may change these slices even when the +// logical table rows are unchanged. This is a file-level diff, not a row diff. +type SnapshotFileDiff struct { + FromSnapshotID int64 + ToSnapshotID int64 + + AddedDataFiles []iceberg.DataFile + RemovedDataFiles []iceberg.DataFile + AddedDeleteFiles []iceberg.DataFile + RemovedDeleteFiles []iceberg.DataFile +} + +// HasChanges reports whether either snapshot references physical files the +// other does not. +func (d SnapshotFileDiff) HasChanges() bool { + return len(d.AddedDataFiles)+len(d.RemovedDataFiles)+ + len(d.AddedDeleteFiles)+len(d.RemovedDeleteFiles) != 0 +} + +// DiffSnapshotFiles compares the live files referenced by any two snapshots +// still present in the table's metadata. The snapshots need not be ancestors; +// for example, they may belong to different branches. The order of the +// arguments determines the direction of additions and removals. +// +// Iceberg manifests are immutable. Shared manifests therefore contain +// identical live entries and can be skipped without opening their Avro files. +// For the remaining manifests, ADDED and EXISTING entries are live; DELETED +// entries are tombstones and must not be included. Data-file paths identify +// immutable content, so a file rewritten into a different manifest does not +// count as added or removed. In v3, deletion vectors use the Puffin +// location, blob offset, and blob length as their identity. +// +// This works for v1 snapshots that embed manifest paths, as well as v1-v3 +// snapshots with manifest lists. It reads only metadata and never reads the +// Parquet, Avro, or ORC data files. +func (t Table) DiffSnapshotFiles(ctx context.Context, fromSnapshotID, toSnapshotID int64) (SnapshotFileDiff, error) { + diff := SnapshotFileDiff{ + FromSnapshotID: fromSnapshotID, + ToSnapshotID: toSnapshotID, + } + if err := ctx.Err(); err != nil { + return diff, err + } + if t.metadata == nil { + return diff, fmt.Errorf("%w: table metadata is not available", ErrInvalidMetadata) + } + + from := t.metadata.SnapshotByID(fromSnapshotID) + if from == nil { + return diff, fmt.Errorf("%w: from snapshot %d", ErrSnapshotNotFound, fromSnapshotID) + } + to := t.metadata.SnapshotByID(toSnapshotID) + if to == nil { + return diff, fmt.Errorf("%w: to snapshot %d", ErrSnapshotNotFound, toSnapshotID) + } + if fromSnapshotID == toSnapshotID { + return diff, nil + } + if t.fsF == nil { + return diff, fmt.Errorf("%w: table file IO is not configured", ErrInvalidOperation) + } + + fs, err := t.fsF(ctx) + if err != nil { + return diff, fmt.Errorf("resolve table file IO: %w", err) + } + if fs == nil { + return diff, fmt.Errorf("%w: table file IO is nil", ErrInvalidOperation) + } + // Share the resolved FileIO between the two manifest-list loads and the + // changed manifest entries. Keep the normal bounded manifest-list cache. + fsF := func(context.Context) (iceio.IO, error) { return fs, nil } + + fromSet, err := t.manifestSetWithFSF(ctx, *from, fsF) + if err != nil { + return diff, fmt.Errorf("read from snapshot %d manifests: %w", fromSnapshotID, err) + } + toSet, err := t.manifestSetWithFSF(ctx, *to, fsF) + if err != nil { + return diff, fmt.Errorf("read to snapshot %d manifests: %w", toSnapshotID, err) + } + + fromOnly, toOnly := uniqueSnapshotManifests(fromSet.borrowAllManifests(), toSet.borrowAllManifests()) + if len(fromOnly) == 0 && len(toOnly) == 0 { + return diff, nil + } + + fromFiles, err := snapshotDiffLiveFiles(ctx, fs, fromOnly) + if err != nil { + return diff, fmt.Errorf("read from snapshot %d entries: %w", fromSnapshotID, err) + } + toFiles, err := snapshotDiffLiveFiles(ctx, fs, toOnly) + if err != nil { + return diff, fmt.Errorf("read to snapshot %d entries: %w", toSnapshotID, err) + } + + // A manifest rewrite can move the same data file into a different + // manifest without modifying table contents. Cancel those paths out. + for key, fromFile := range fromFiles { + if toFile, ok := toFiles[key]; ok { + if fromFile.ContentType() != toFile.ContentType() { + return diff, fmt.Errorf("%w: content file %q changed type between snapshots", + ErrInvalidMetadata, key.path) + } + delete(fromFiles, key) + delete(toFiles, key) + } + } + + for _, file := range fromFiles { + if file.ContentType() == iceberg.EntryContentData { + diff.RemovedDataFiles = append(diff.RemovedDataFiles, file) + } else { + diff.RemovedDeleteFiles = append(diff.RemovedDeleteFiles, file) + } + } + for _, file := range toFiles { + if file.ContentType() == iceberg.EntryContentData { + diff.AddedDataFiles = append(diff.AddedDataFiles, file) + } else { + diff.AddedDeleteFiles = append(diff.AddedDeleteFiles, file) + } + } + + // Maps have unstable iteration order. Stable paths make the API useful + // for reproducible audit logs and change-detection checkpoints. + sortSnapshotDiffFiles(diff.AddedDataFiles) + sortSnapshotDiffFiles(diff.RemovedDataFiles) + sortSnapshotDiffFiles(diff.AddedDeleteFiles) + sortSnapshotDiffFiles(diff.RemovedDeleteFiles) + + return diff, nil +} + +// uniqueSnapshotManifests skips immutable manifests that are present in both +// snapshots. No ancestry walk is needed, and different branches work. +func uniqueSnapshotManifests(from, to []iceberg.ManifestFile) (fromOnly, toOnly []iceberg.ManifestFile) { + fromPaths := make(map[string]struct{}, len(from)) + toPaths := make(map[string]struct{}, len(to)) + for _, manifest := range from { + fromPaths[manifest.FilePath()] = struct{}{} + } + for _, manifest := range to { + toPaths[manifest.FilePath()] = struct{}{} + if _, shared := fromPaths[manifest.FilePath()]; !shared { + toOnly = append(toOnly, manifest) + } + } + for _, manifest := range from { + if _, shared := toPaths[manifest.FilePath()]; !shared { + fromOnly = append(fromOnly, manifest) + } + } + + return fromOnly, toOnly +} + +// snapshotDiffFileKey uses the spec's identity for content files. A v3 Puffin +// file may hold several different deletion vector blobs, so its path alone +// is not enough to identify the vector. +type snapshotDiffFileKey struct { + path string + vector bool + offset int64 + length int64 +} + +func snapshotFileDiffKey(file iceberg.DataFile) (snapshotDiffFileKey, error) { + key := snapshotDiffFileKey{path: file.FilePath()} + if key.path == "" { + return key, fmt.Errorf("%w: content file path is empty", ErrInvalidMetadata) + } + if IsDeletionVector(file) { + offset := file.ContentOffset() + length := file.ContentSizeInBytes() + if offset == nil || length == nil || *offset < 0 || *length <= 0 { + return key, fmt.Errorf("%w: deletion vector %q has an invalid offset or length", + ErrInvalidMetadata, file.FilePath()) + } Review Comment: No test reaches this branch, so someone could break it without noticing. Without the nil check, bad metadata would cause a panic. This is easy to test, because the manifest writer accepts such files: build a Puffin position delete with NewDataFileBuilder, and expect ErrInvalidMetadata. I'd add four cases: missing offset, missing length, negative offset, and zero length. The test can reuse the helpers already in this file, so it needs no new setup. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
