This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new cbd8b7059 fix(table): clean up expired files from planning metadata
(#2351)
cbd8b7059 is described below
commit cbd8b70593f81198717979b4f9e6301732e427da
Author: Harrison Crosse <[email protected]>
AuthorDate: Fri Oct 30 11:47:32 2026 -0400
fix(table): clean up expired files from planning metadata (#2351)
* fix(table): clean up expired files from planning metadata
Cleanup uses the metadata that expiration was planned against to find
snapshots and statistics staged in the same transaction. For multiple
expirations, cleanup collects files before deleting any. Cleanup keeps
files for snapshots that remain in the committed table.
A retry that rebuilds an expired snapshot can still cause a cleanup error
after the commit succeeds. I will replan expiration in a follow-up PR.
Part of #2339.
Signed-off-by: Harrison Crosse <[email protected]>
* refactor(table): retain only expired snapshots for cleanup
I replaced the metadata copy per expiry with copies of the expired
snapshots and their statistics file paths. `ExpireSnapshots` no longer
clones or builds metadata. I added tests for mixed cleanup and
statistics cleanup within a transaction.
Part of #2339.
Signed-off-by: Harrison Crosse <[email protected]>
---------
Signed-off-by: Harrison Crosse <[email protected]>
---
table/metadata.go | 11 +-
table/transaction.go | 13 ++-
table/transaction_internal_test.go | 213 +++++++++++++++++++++++++++++++++++++
table/updates.go | 111 +++++++++++++------
table/updates_test.go | 97 +++++++++++++++++
5 files changed, 411 insertions(+), 34 deletions(-)
diff --git a/table/metadata.go b/table/metadata.go
index 6c15b229c..1ba78b3bf 100644
--- a/table/metadata.go
+++ b/table/metadata.go
@@ -1065,8 +1065,13 @@ func (b *MetadataBuilder) NextRowID() int64 {
}
func (b *MetadataBuilder) RemoveSnapshots(snapshotIds []int64, postCommit
bool) error {
- removedIDs := make(map[int64]struct{}, len(snapshotIds))
- for _, snapshotID := range snapshotIds {
+ return b.removeSnapshotsInternal(NewRemoveSnapshotsUpdate(snapshotIds,
postCommit))
+}
+
+// Stage u itself because it carries the expired snapshots that cleanup reads.
+func (b *MetadataBuilder) removeSnapshotsInternal(u *removeSnapshotsUpdate)
error {
+ removedIDs := make(map[int64]struct{}, len(u.SnapshotIDs))
+ for _, snapshotID := range u.SnapshotIDs {
removedIDs[snapshotID] = struct{}{}
}
@@ -1116,7 +1121,7 @@ func (b *MetadataBuilder) RemoveSnapshots(snapshotIds
[]int64, postCommit bool)
return ok
})
- b.updates = append(b.updates, NewRemoveSnapshotsUpdate(snapshotIds,
postCommit))
+ b.updates = append(b.updates, u)
return nil
}
diff --git a/table/transaction.go b/table/transaction.go
index 54321735a..67ffbb08c 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -909,7 +909,9 @@ func (t *Transaction) ExpireSnapshots(opts
...ExpireSnapshotsOpt) error {
// Only add the update if there are actually snapshots to delete
if len(snapsToDelete) > 0 {
- updates = append(updates,
NewRemoveSnapshotsUpdate(snapsToDelete, cfg.postCommit))
+ removal := NewRemoveSnapshotsUpdate(snapsToDelete,
cfg.postCommit)
+ removal.expired = captureExpired(snapsToDelete, lookup,
slices.Values(meta.statisticsList), slices.Values(meta.partitionStatsList))
+ updates = append(updates, removal)
}
// A no-op expiry must not pin the commit branch, which would stop the
@@ -2632,11 +2634,20 @@ func (t *Transaction) Commit(ctx context.Context)
(*Table, error) {
// otherwise a retry would re-run the catalog commit.
t.committed = true
+ var removals []*removeSnapshotsUpdate
for _, u := range meta.updates {
+ if removal, ok := u.(*removeSnapshotsUpdate); ok {
+ removals = append(removals, removal)
+
+ continue
+ }
if perr := u.PostCommit(ctx, t.tbl, tbl); perr != nil {
err = errors.Join(err, perr)
}
}
+ if perr := deleteExpiredFiles(ctx, t.tbl, tbl, removals...);
perr != nil {
+ err = errors.Join(err, perr)
+ }
return tbl, err
}
diff --git a/table/transaction_internal_test.go
b/table/transaction_internal_test.go
index ecc074488..8a4a6270d 100644
--- a/table/transaction_internal_test.go
+++ b/table/transaction_internal_test.go
@@ -20,6 +20,8 @@ package table
import (
"context"
"fmt"
+ "io/fs"
+ "maps"
"slices"
"testing"
"time"
@@ -614,6 +616,217 @@ func TestExpireSnapshotsMaterializesDeferredHistory(t
*testing.T) {
assert.Contains(t, removal.SnapshotIDs, historicalID)
}
+func TestExpireSnapshotsCleansUpSnapshotStagedInSameTransaction(t *testing.T) {
+ ctx := context.Background()
+ spec := iceberg.NewPartitionSpec()
+ wfs, meta := newMemIOWithRetryMeta(t, spec)
+ cat := &sequentialCatalog{metadata: meta}
+ txn := newOCCTable(t, meta, wfs, cat).NewTransaction()
+
+ // Snapshots are dated before the table's last update, so
WithOlderThan(0)
+ // selects s1 however quickly the test runs.
+ stageAppend := func(path string, timestampMs int64) *Snapshot {
+ sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+ sp.appendDataFile(newTestDataFile(t, spec, path, nil))
+ updates, reqs, err := sp.commit(ctx)
+ require.NoError(t, err)
+ snap := updates[0].(*addSnapshotUpdate).Snapshot
+ snap.TimestampMs = timestampMs
+ require.NoError(t, txn.apply(updates, reqs))
+
+ return snap
+ }
+ s1 := stageAppend("mem://default/table-location/data/s1.parquet",
meta.LastUpdatedMillis()-30_000)
+ s2 := stageAppend("mem://default/table-location/data/s2.parquet",
meta.LastUpdatedMillis()-20_000)
+
+ require.NoError(t, txn.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(1)))
+
+ before := slices.Collect(maps.Keys(wfs.files))
+ committed, err := txn.Commit(ctx)
+ require.NoError(t, err)
+
+ assert.Nil(t, committed.Metadata().SnapshotByID(s1.SnapshotID))
+ require.NotNil(t, committed.Metadata().SnapshotByID(s2.SnapshotID))
+
+ var removed []string
+ for _, path := range before {
+ if _, ok := wfs.files[path]; !ok {
+ removed = append(removed, path)
+ }
+ }
+ // s2 still references s1's manifest.
+ assert.Equal(t, []string{s1.ManifestList}, removed)
+}
+
+func TestExpireSnapshotsTwiceInTransactionDeletesSharedFilesOnce(t *testing.T)
{
+ const (
+ list1 = "s3://bucket/meta/snap-1.avro"
+ list2 = "s3://bucket/meta/snap-2.avro"
+ list3 = "s3://bucket/meta/snap-3.avro"
+ manifest1 = "s3://bucket/meta/manifest-1.avro"
+ manifest2 = "s3://bucket/meta/manifest-2.avro"
+ manifest3 = "s3://bucket/meta/manifest-3.avro"
+ data1 = "s3://bucket/data/file-1.parquet"
+ data2 = "s3://bucket/data/file-2.parquet"
+ data3 = "s3://bucket/data/file-3.parquet"
+ )
+
+ // Snapshot 3 uses separate manifests to model an overwrite.
+ tio := newTrackingCallsIO()
+ m1 := writeManifest(t, tio.trackingIO, 1, 1, manifest1, data1)
+ m2 := writeManifest(t, tio.trackingIO, 2, 2, manifest2, data2)
+ m3 := writeManifest(t, tio.trackingIO, 3, 3, manifest3, data3)
+ writeManifestList(t, tio.trackingIO, 1, list1,
[]iceberg.ManifestFile{m1})
+ writeManifestList(t, tio.trackingIO, 2, list2,
[]iceberg.ManifestFile{m1, m2})
+ writeManifestList(t, tio.trackingIO, 3, list3,
[]iceberg.ManifestFile{m3})
+ for _, path := range []string{data1, data2, data3} {
+ tio.files[path] = []byte("data")
+ }
+
+ txn := stageTwoExpirations(t, tio, list1, list2, list3)
+ committed, err := txn.Commit(context.Background())
+ require.NoError(t, err)
+ require.Len(t, committed.Metadata().Snapshots(), 1)
+
+ for _, path := range []string{list1, list2, manifest1, manifest2,
data1, data2} {
+ assert.NotContains(t, tio.files, path)
+ assert.Equal(t, 1, tio.removeCount[path], "%s must be deleted
exactly once", path)
+ }
+ for _, path := range []string{list3, manifest3, data3} {
+ assert.Contains(t, tio.files, path)
+ }
+}
+
+func TestExpireSnapshotsTwiceInTransactionDeletesNothingWhenCleanupFails(t
*testing.T) {
+ const (
+ list1 = "s3://bucket/meta/snap-1.avro"
+ list2 = "s3://bucket/meta/snap-2.avro"
+ list3 = "s3://bucket/meta/snap-3.avro"
+ manifest1 = "s3://bucket/meta/manifest-1.avro"
+ manifest3 = "s3://bucket/meta/manifest-3.avro"
+ data1 = "s3://bucket/data/file-1.parquet"
+ data3 = "s3://bucket/data/file-3.parquet"
+ )
+
+ // Snapshot 2's manifest list is never written, so the second removal
+ // cannot read it.
+ tio := newTrackingCallsIO()
+ m1 := writeManifest(t, tio.trackingIO, 1, 1, manifest1, data1)
+ m3 := writeManifest(t, tio.trackingIO, 3, 3, manifest3, data3)
+ writeManifestList(t, tio.trackingIO, 1, list1,
[]iceberg.ManifestFile{m1})
+ writeManifestList(t, tio.trackingIO, 3, list3,
[]iceberg.ManifestFile{m3})
+ for _, path := range []string{data1, data3} {
+ tio.files[path] = []byte("data")
+ }
+
+ txn := stageTwoExpirations(t, tio, list1, list2, list3)
+ committed, err := txn.Commit(context.Background())
+ require.ErrorIs(t, err, fs.ErrNotExist)
+ require.NotNil(t, committed, "the catalog commit landed before cleanup
failed")
+ require.Len(t, committed.Metadata().Snapshots(), 1)
+
+ assert.Empty(t, tio.removeCount)
+}
+
+func TestExpireSnapshotsWithAndWithoutCleanupInTransaction(t *testing.T) {
+ const (
+ list1 = "s3://bucket/meta/snap-1.avro"
+ list2 = "s3://bucket/meta/snap-2.avro"
+ list3 = "s3://bucket/meta/snap-3.avro"
+ manifest1 = "s3://bucket/meta/manifest-1.avro"
+ manifest2 = "s3://bucket/meta/manifest-2.avro"
+ manifest3 = "s3://bucket/meta/manifest-3.avro"
+ data1 = "s3://bucket/data/file-1.parquet"
+ data2 = "s3://bucket/data/file-2.parquet"
+ data3 = "s3://bucket/data/file-3.parquet"
+ )
+
+ tio := newTrackingIO()
+ m1 := writeManifest(t, tio, 1, 1, manifest1, data1)
+ m2 := writeManifest(t, tio, 2, 2, manifest2, data2)
+ m3 := writeManifest(t, tio, 3, 3, manifest3, data3)
+ writeManifestList(t, tio, 1, list1, []iceberg.ManifestFile{m1})
+ writeManifestList(t, tio, 2, list2, []iceberg.ManifestFile{m1, m2})
+ writeManifestList(t, tio, 3, list3, []iceberg.ManifestFile{m3})
+ for _, path := range []string{data1, data2, data3} {
+ tio.files[path] = []byte("data")
+ }
+
+ txn := newThreeSnapshotTxn(t, tio, list1, list2, list3)
+ require.NoError(t, txn.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(2)))
+ require.NoError(t, txn.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(1), WithPostCommit(false)))
+ committed, err := txn.Commit(context.Background())
+ require.NoError(t, err)
+ require.Len(t, committed.Metadata().Snapshots(), 1)
+
+ // Snapshot 2 shares manifest 1 with snapshot 1, whose expiry cleans up.
+ assert.ElementsMatch(t,
+ []string{list2, manifest2, data2, list3, manifest3, data3},
+ slices.Collect(maps.Keys(tio.files)))
+}
+
+func TestExpireSnapshotsCleansUpStatisticsStagedInSameTransaction(t
*testing.T) {
+ const (
+ list1 = "s3://bucket/meta/snap-1.avro"
+ list2 = "s3://bucket/meta/snap-2.avro"
+ list3 = "s3://bucket/meta/snap-3.avro"
+ statsPath = "s3://bucket/stats/snap1.puffin"
+ partStatsPath = "s3://bucket/stats/snap1-part.puffin"
+ )
+
+ tio := newTrackingIO()
+ for i, list := range []string{list1, list2, list3} {
+ writeManifestList(t, tio, int64(i+1), list, nil)
+ }
+ tio.files[statsPath] = []byte("puffin")
+ tio.files[partStatsPath] = []byte("puffin")
+
+ // The base table has no statistics, so only the transaction lists them.
+ txn := newThreeSnapshotTxn(t, tio, list1, list2, list3)
+ require.NoError(t, txn.apply([]Update{
+ NewSetStatisticsUpdate(StatisticsFile{SnapshotID: 1,
StatisticsPath: statsPath}),
+
NewSetPartitionStatisticsUpdate(PartitionStatisticsFile{SnapshotID: 1,
StatisticsPath: partStatsPath}),
+ }, nil))
+ require.NoError(t, txn.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(2)))
+ committed, err := txn.Commit(context.Background())
+ require.NoError(t, err)
+ require.Nil(t, committed.Metadata().SnapshotByID(1))
+
+ assert.ElementsMatch(t, []string{list2, list3},
slices.Collect(maps.Keys(tio.files)))
+}
+
+func stageTwoExpirations(t *testing.T, fio iceio.IO, list1, list2, list3
string) *Transaction {
+ t.Helper()
+
+ txn := newThreeSnapshotTxn(t, fio, list1, list2, list3)
+ require.NoError(t, txn.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(2)))
+ require.NoError(t, txn.ExpireSnapshots(WithOlderThan(0),
WithRetainLast(1)))
+
+ return txn
+}
+
+func newThreeSnapshotTxn(t *testing.T, fio iceio.IO, list1, list2, list3
string) *Transaction {
+ t.Helper()
+
+ base, err := ParseMetadataString(buildMetaJSON(metaJSONOpts{
+ snapshots: fmt.Sprintf(
+
`{"snapshot-id":1,"timestamp-ms":1000,"manifest-list":%q},`+
+
`{"snapshot-id":2,"parent-snapshot-id":1,"timestamp-ms":2000,"manifest-list":%q},`+
+
`{"snapshot-id":3,"parent-snapshot-id":2,"timestamp-ms":3000,"manifest-list":%q}`,
+ list1, list2, list3),
+ }))
+ require.NoError(t, err)
+ builder, err := MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ require.NoError(t, builder.SetSnapshotRef(MainBranch, 3, BranchRef))
+ meta, err := builder.Build()
+ require.NoError(t, err)
+
+ cat := &sequentialCatalog{metadata: meta}
+
+ return New(Identifier{"ns", "tbl"}, meta, "metadata.json",
testFSF(fio), cat).NewTransaction()
+}
+
func TestExpireSnapshotsUsesStandardRetentionPropertyNames(t *testing.T) {
tests := []struct {
name string
diff --git a/table/updates.go b/table/updates.go
index 511f556d5..b6e4b50e9 100644
--- a/table/updates.go
+++ b/table/updates.go
@@ -23,6 +23,7 @@ import (
"encoding/json"
"errors"
"fmt"
+ "iter"
"maps"
"slices"
"strings"
@@ -636,6 +637,16 @@ type removeSnapshotsUpdate struct {
baseUpdate
SnapshotIDs []int64 `json:"snapshot-ids"`
postCommit bool
+
+ // Includes snapshots and statistics staged before this removal.
+ // Nil falls back to preTable.
+ expired map[int64]*expiredSnapshot
+}
+
+type expiredSnapshot struct {
+ Snapshot
+ // Both statistics and partition-statistics files.
+ statisticsPaths []string
}
// NewRemoveSnapshotsUpdate creates a new update that removes all snapshots
from
@@ -649,11 +660,17 @@ func NewRemoveSnapshotsUpdate(ids []int64, postCommit
bool) *removeSnapshotsUpda
}
func (u *removeSnapshotsUpdate) Apply(builder *MetadataBuilder) error {
- return builder.RemoveSnapshots(u.SnapshotIDs, u.postCommit)
+ return builder.removeSnapshotsInternal(u)
}
func (u *removeSnapshotsUpdate) PostCommit(ctx context.Context, preTable
*Table, postTable *Table) error {
- if !u.postCommit || postTable == nil ||
!IsGCEnabled(postTable.Properties()) {
+ return deleteExpiredFiles(ctx, preTable, postTable, u)
+}
+
+// Removals can share files. Collect all candidates before deleting any files.
+func deleteExpiredFiles(ctx context.Context, preTable, postTable *Table,
removals ...*removeSnapshotsUpdate) error {
+ cleansUp := func(u *removeSnapshotsUpdate) bool { return u.postCommit }
+ if !slices.ContainsFunc(removals, cleansUp) || postTable == nil ||
!IsGCEnabled(postTable.Properties()) {
return nil
}
@@ -664,30 +681,16 @@ func (u *removeSnapshotsUpdate) PostCommit(ctx
context.Context, preTable *Table,
filesToDelete := make(map[string]struct{})
- for _, snapId := range u.SnapshotIDs {
- snap := preTable.Metadata().SnapshotByID(snapId)
- if snap == nil {
- return errors.New("snapshot should never be nil")
+ var expired []*Snapshot
+ for _, u := range removals {
+ if !u.postCommit {
+ continue
}
-
- filesToDelete[snap.ManifestList] = struct{}{}
- }
-
- expiredIDs := make(map[int64]struct{}, len(u.SnapshotIDs))
- for _, id := range u.SnapshotIDs {
- expiredIDs[id] = struct{}{}
- }
-
- for sf := range preTable.Metadata().Statistics() {
- if _, ok := expiredIDs[sf.SnapshotID]; ok && sf.StatisticsPath
!= "" {
- filesToDelete[sf.StatisticsPath] = struct{}{}
- }
- }
-
- for psf := range preTable.Metadata().PartitionStatistics() {
- if _, ok := expiredIDs[psf.SnapshotID]; ok &&
psf.StatisticsPath != "" {
- filesToDelete[psf.StatisticsPath] = struct{}{}
+ snaps, err := u.expiredSnapshots(preTable, postTable,
filesToDelete)
+ if err != nil {
+ return err
}
+ expired = append(expired, snaps...)
}
// Preload retained manifest lists once so the live-data-file pass
below can walk them without re-fetching.
@@ -713,12 +716,7 @@ func (u *removeSnapshotsUpdate) PostCommit(ctx
context.Context, preTable *Table,
// snapshots that share manifests by reference.
visitedManifests := make(map[string]struct{})
- for _, snapId := range u.SnapshotIDs {
- snap := preTable.SnapshotByID(snapId)
- if snap == nil {
- return fmt.Errorf("missing snapshot %d", snapId)
- }
-
+ for _, snap := range expired {
mans, err := snap.Manifests(prefs)
if err != nil {
return err
@@ -807,6 +805,59 @@ func (u *removeSnapshotsUpdate) PostCommit(ctx
context.Context, preTable *Table,
return res
}
+// expiredSnapshots also adds the expired snapshots' manifest lists and
+// statistics paths to files.
+func (u *removeSnapshotsUpdate) expiredSnapshots(preTable, postTable *Table,
files map[string]struct{}) ([]*Snapshot, error) {
+ planned := u.expired
+ if planned == nil {
+ pre := preTable.Metadata()
+ planned = captureExpired(u.SnapshotIDs, pre.SnapshotByID,
pre.Statistics(), pre.PartitionStatistics())
+ }
+
+ snaps := make([]*Snapshot, 0, len(u.SnapshotIDs))
+ for _, id := range u.SnapshotIDs {
+ if postTable.Metadata().SnapshotByID(id) != nil {
+ continue
+ }
+ snap, ok := planned[id]
+ if !ok {
+ return nil, fmt.Errorf("missing snapshot %d", id)
+ }
+ files[snap.ManifestList] = struct{}{}
+ for _, path := range snap.statisticsPaths {
+ files[path] = struct{}{}
+ }
+ snaps = append(snaps, &snap.Snapshot)
+ }
+
+ return snaps, nil
+}
+
+// captureExpired copies values so the result does not keep the source
+// metadata alive. It leaves out IDs that snapshotByID cannot find.
+func captureExpired(ids []int64, snapshotByID func(int64) *Snapshot, stats
iter.Seq[StatisticsFile], partitionStats iter.Seq[PartitionStatisticsFile])
map[int64]*expiredSnapshot {
+ expired := make(map[int64]*expiredSnapshot, len(ids))
+ for _, id := range ids {
+ if snap := snapshotByID(id); snap != nil {
+ expired[id] = &expiredSnapshot{Snapshot:
cloneSnapshot(*snap)}
+ }
+ }
+
+ addPath := func(id int64, path string) {
+ if snap, ok := expired[id]; ok && path != "" {
+ snap.statisticsPaths = append(snap.statisticsPaths,
path)
+ }
+ }
+ for sf := range stats {
+ addPath(sf.SnapshotID, sf.StatisticsPath)
+ }
+ for psf := range partitionStats {
+ addPath(psf.SnapshotID, psf.StatisticsPath)
+ }
+
+ return expired
+}
+
type removeSnapshotRefUpdate struct {
baseUpdate
RefName string `json:"ref-name"`
diff --git a/table/updates_test.go b/table/updates_test.go
index d640654cb..f6b85af01 100644
--- a/table/updates_test.go
+++ b/table/updates_test.go
@@ -228,6 +228,45 @@ func TestRemoveSnapshotsPostCommitDeletesStatisticsFiles(t
*testing.T) {
assert.NotContains(t, tio.files, "s3://bucket/stats/snap1-part.puffin")
}
+func TestRemoveSnapshotsPostCommitDeletesEveryStatisticsFileOfSnapshot(t
*testing.T) {
+ paths := []string{
+ "s3://bucket/stats/snap1-a.puffin",
+ "s3://bucket/stats/snap1-b.puffin",
+ "s3://bucket/stats/snap1-part-a.puffin",
+ "s3://bucket/stats/snap1-part-b.puffin",
+ }
+ stat := func(path string) string {
+ return `{"snapshot-id":1,"statistics-path":"` + path +
`","file-size-in-bytes":100,"file-footer-size-in-bytes":10,"blob-metadata":[]}`
+ }
+ partStat := func(path string) string {
+ return `{"snapshot-id":1,"statistics-path":"` + path +
`","file-size-in-bytes":50}`
+ }
+ pre, err := ParseMetadataString(buildMetaJSON(metaJSONOpts{
+ snapshots: `{"snapshot-id":1,"timestamp-ms":1000}`,
+ statistics: stat(paths[0]) + "," + stat(paths[1]),
+ partitionStatistics: partStat(paths[2]) + "," +
partStat(paths[3]),
+ }))
+ require.NoError(t, err)
+ post, err := ParseMetadataString(buildMetaJSON(metaJSONOpts{}))
+ require.NoError(t, err)
+
+ tio := newTrackingIO()
+ for _, path := range paths {
+ tio.files[path] = []byte("puffin")
+ }
+
+ fsF := testFSF(tio)
+ preTable := New(Identifier{"ns", "tbl"}, pre, "metadata.json", fsF, nil)
+ postTable := New(Identifier{"ns", "tbl"}, post, "metadata.json", fsF,
nil)
+
+ update := NewRemoveSnapshotsUpdate([]int64{1}, true)
+ require.NoError(t, update.PostCommit(context.Background(), preTable,
postTable))
+
+ for _, path := range paths {
+ assert.NotContains(t, tio.files, path)
+ }
+}
+
func TestRemoveSnapshotsPostCommitPreservesStatisticsOfSurvivingSnapshots(t
*testing.T) {
// pre: snapshots 1 and 2, statistics for both.
// post: snapshot 1 expired, snapshot 2 kept with its statistics still
present.
@@ -324,6 +363,64 @@ func TestRemoveSnapshotsPostCommitSharedManifestRetained(t
*testing.T) {
assert.Contains(t, tio.files, manifestList2)
}
+func TestRemoveSnapshotsPostCommitKeepsFilesOfSnapshotStillInPostTable(t
*testing.T) {
+ const (
+ dataPath = "s3://bucket/data/file-1.parquet"
+ manifestPath = "s3://bucket/meta/manifest-1.avro"
+ manifestList = "s3://bucket/meta/snap-1.avro"
+ )
+
+ tio := newTrackingIO()
+ mf := writeManifest(t, tio, 1, 1, manifestPath, dataPath)
+ writeManifestList(t, tio, 1, manifestList, []iceberg.ManifestFile{mf})
+ tio.files[dataPath] = []byte("data")
+
+ meta, err := ParseMetadataString(buildMetaJSON(metaJSONOpts{
+ snapshots:
fmt.Sprintf(`{"snapshot-id":1,"timestamp-ms":1000,"manifest-list":%q}`,
manifestList),
+ }))
+ require.NoError(t, err)
+ tbl := New(Identifier{"ns", "tbl"}, meta, "metadata.json",
testFSF(tio), nil)
+
+ update := NewRemoveSnapshotsUpdate([]int64{1}, true)
+ require.NoError(t, update.PostCommit(context.Background(), tbl, tbl))
+
+ assert.Contains(t, tio.files, manifestList)
+ assert.Contains(t, tio.files, manifestPath)
+ assert.Contains(t, tio.files, dataPath)
+}
+
+func TestRemoveSnapshotsPostCommitKeepsStatisticsOfSnapshotStillInPostTable(t
*testing.T) {
+ // Omit statistics from postTable to verify that retaining the snapshot
is sufficient.
+ const (
+ statsPath = "s3://bucket/stats/snap1.puffin"
+ partStatsPath = "s3://bucket/stats/snap1-part.puffin"
+ )
+ pre, err := ParseMetadataString(buildMetaJSON(metaJSONOpts{
+ snapshots: `{"snapshot-id":1,"timestamp-ms":1000}`,
+ statistics: `{"snapshot-id":1,"statistics-path":"` +
statsPath +
`","file-size-in-bytes":100,"file-footer-size-in-bytes":10,"blob-metadata":[]}`,
+ partitionStatistics: `{"snapshot-id":1,"statistics-path":"` +
partStatsPath + `","file-size-in-bytes":50}`,
+ }))
+ require.NoError(t, err)
+ post, err := ParseMetadataString(buildMetaJSON(metaJSONOpts{
+ snapshots: `{"snapshot-id":1,"timestamp-ms":1000}`,
+ }))
+ require.NoError(t, err)
+
+ tio := newTrackingIO()
+ tio.files[statsPath] = []byte("puffin")
+ tio.files[partStatsPath] = []byte("puffin")
+
+ fsF := testFSF(tio)
+ preTable := New(Identifier{"ns", "tbl"}, pre, "metadata.json", fsF, nil)
+ postTable := New(Identifier{"ns", "tbl"}, post, "metadata.json", fsF,
nil)
+
+ update := NewRemoveSnapshotsUpdate([]int64{1}, true)
+ require.NoError(t, update.PostCommit(context.Background(), preTable,
postTable))
+
+ assert.Contains(t, tio.files, statsPath)
+ assert.Contains(t, tio.files, partStatsPath)
+}
+
func TestRemoveSnapshotsPostCommitSharedManifestExpiredOnce(t *testing.T) {
// Manifest M is referenced by two expired snapshots (1 and 2) and no
// retained snapshot. PostCommit must delete M and its data file, and
