This is an automated email from the ASF dual-hosted git repository.
laskoviymishka 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 9cbeca458 perf(table): keep deferred snapshots lazy in metadata
builders (#2186)
9cbeca458 is described below
commit 9cbeca458a2e5b734f5f162fc81af267750015e5
Author: Roli Bhardwaj <[email protected]>
AuthorDate: Sat Oct 17 17:40:02 2026 -0400
perf(table): keep deferred snapshots lazy in metadata builders (#2186)
---
catalog/rest/lazy_snapshot_bench_test.go | 81 +++++++-
table/deferred_snapshots.go | 72 +++++++
table/deferred_snapshots_test.go | 324 ++++++++++++++++++++++++++++++-
table/metadata.go | 206 ++++++++++++++------
table/metadata_builder_internal_test.go | 1 +
table/transaction.go | 8 +-
table/transaction_internal_test.go | 32 +++
7 files changed, 650 insertions(+), 74 deletions(-)
diff --git a/catalog/rest/lazy_snapshot_bench_test.go
b/catalog/rest/lazy_snapshot_bench_test.go
index a18b8d2dc..a815c81ea 100644
--- a/catalog/rest/lazy_snapshot_bench_test.go
+++ b/catalog/rest/lazy_snapshot_bench_test.go
@@ -141,9 +141,8 @@ func BenchmarkStageDeferredAllSnapshots(b *testing.B) {
}
// BenchmarkStageParseThenMetadataBuilder measures the path taken when a caller
-// starts another transaction from the Table returned by Commit. Builder
-// creation requires complete snapshot history, so this exposes the cost of
-// indexing deferred snapshots and then fully materializing them.
+// starts another transaction from the Table returned by Commit. Deferred
+// builder creation should retain the indexed raw history without decoding it.
func BenchmarkStageParseThenMetadataBuilder(b *testing.B) {
for _, n := range benchSnapshotCounts {
body := makeTableResponseWithSnapshots(n)
@@ -178,6 +177,82 @@ func BenchmarkStageParseThenMetadataBuilder(b *testing.B) {
}
}
+// BenchmarkStageParseBuilderMutation measures representative transaction work
+// after parsing a commit response. Add/ref and remove/build should preserve
the
+// deferred history, while serialization intentionally materializes it.
+func BenchmarkStageParseBuilderMutation(b *testing.B) {
+ for _, n := range benchSnapshotCounts {
+ body := makeTableResponseWithSnapshots(n)
+ rawMeta := extractMetadata(b, body)
+ for _, mode := range []struct {
+ name string
+ parse func([]byte) (table.Metadata, error)
+ }{
+ {name: "eager", parse: table.ParseMetadataBytes},
+ {name: "deferred", parse:
table.ParseMetadataBytesDeferredSnapshots},
+ } {
+ for _, operation := range []struct {
+ name string
+ remove bool
+ serialize bool
+ }{
+ {name: "add-ref-build"},
+ {name: "remove-build", remove: true},
+ {name: "add-ref-build-serialize", serialize:
true},
+ } {
+
b.Run(metadataBenchmarkName(rawMeta)+"/"+mode.name+"/"+operation.name, func(b
*testing.B) {
+ b.ReportAllocs()
+ b.ReportMetric(float64(len(rawMeta)),
"metadata-bytes")
+ b.ReportMetric(float64(n), "snapshots")
+ for range b.N {
+ meta, err := mode.parse(rawMeta)
+ if err != nil {
+ b.Fatal(err)
+ }
+ builder, err :=
table.MetadataBuilderFromBase(meta, "")
+ if err != nil {
+ b.Fatal(err)
+ }
+ if operation.remove {
+ if err :=
builder.RemoveSnapshots([]int64{0}, false); err != nil {
+ b.Fatal(err)
+ }
+ } else {
+ current :=
meta.CurrentSnapshot()
+ if current == nil {
+
b.Fatal("missing current snapshot")
+ }
+ added := *current
+ added.SnapshotID = n
+ added.ParentSnapshotID
= ¤t.SnapshotID
+ added.TimestampMs =
current.TimestampMs + 1
+ if meta.Version() > 1 {
+
added.SequenceNumber = meta.LastSequenceNumber() + 1
+ }
+ if err :=
builder.AddSnapshot(&added); err != nil {
+ b.Fatal(err)
+ }
+ if err :=
builder.SetSnapshotRef(table.MainBranch, added.SnapshotID, table.BranchRef);
err != nil {
+ b.Fatal(err)
+ }
+ }
+
+ built, err := builder.Build()
+ if err != nil {
+ b.Fatal(err)
+ }
+ if operation.serialize {
+ if _, err :=
json.Marshal(built); err != nil {
+ b.Fatal(err)
+ }
+ }
+ }
+ })
+ }
+ }
+ }
+}
+
// BenchmarkStageDeferredRetainedHeap measures the live heap held by one
// deferred metadata object before and after full snapshot materialization.
// Run with -benchtime=1x: retained-heap measurements describe one object and
diff --git a/table/deferred_snapshots.go b/table/deferred_snapshots.go
index 510efd297..f1954edab 100644
--- a/table/deferred_snapshots.go
+++ b/table/deferred_snapshots.go
@@ -22,6 +22,7 @@ import (
"encoding/json"
"errors"
"fmt"
+ "maps"
"strconv"
"sync"
)
@@ -41,6 +42,56 @@ type deferredSnapshotState struct {
err error
}
+// deferredSnapshotOverlay records mutations over an immutable deferred
+// snapshot state. It belongs to immutable built metadata, so the fully applied
+// collection can be cached after the first collection access or marshal.
+type deferredSnapshotOverlay struct {
+ additions []Snapshot
+ removed map[int64]struct{}
+
+ applyOnce sync.Once
+ applied []Snapshot
+}
+
+func (o *deferredSnapshotOverlay) clone() *deferredSnapshotOverlay {
+ if o == nil {
+ return nil
+ }
+
+ return &deferredSnapshotOverlay{
+ additions: cloneSnapshots(o.additions),
+ removed: maps.Clone(o.removed),
+ }
+}
+
+func (o *deferredSnapshotOverlay) isRemoved(id int64) bool {
+ if o == nil {
+ return false
+ }
+
+ _, ok := o.removed[id]
+
+ return ok
+}
+
+func (o *deferredSnapshotOverlay) apply(base []Snapshot) []Snapshot {
+ if o == nil || (len(o.additions) == 0 && len(o.removed) == 0) {
+ return base
+ }
+
+ o.applyOnce.Do(func() {
+ o.applied = make([]Snapshot, 0, len(base)+len(o.additions))
+ for _, snapshot := range base {
+ if !o.isRemoved(snapshot.SnapshotID) {
+ o.applied = append(o.applied, snapshot)
+ }
+ }
+ o.applied = append(o.applied, o.additions...)
+ })
+
+ return o.applied
+}
+
type deferredSnapshotEntry struct {
start int
end int
@@ -134,6 +185,27 @@ func (s *deferredSnapshotState) snapshotByID(id int64)
(*Snapshot, error) {
return cloneSnapshotPtr(&entry.snapshot), nil
}
+func (s *deferredSnapshotState) containsID(id int64) bool {
+ s.mu.RLock()
+ defer s.mu.RUnlock()
+
+ if s.snapshots != nil {
+ _, ok := snapshotIndexPosition(s.index, s.snapshots, id)
+
+ return ok
+ }
+ _, ok := s.byID[id]
+
+ return ok
+}
+
+func (s *deferredSnapshotState) isMaterialized() bool {
+ s.mu.RLock()
+ defer s.mu.RUnlock()
+
+ return s.raw == nil
+}
+
type deferredSnapshotFields struct {
SnapshotID *int64 `json:"snapshot-id"`
ParentSnapshotID *int64 `json:"parent-snapshot-id,omitempty"`
diff --git a/table/deferred_snapshots_test.go b/table/deferred_snapshots_test.go
index 87dda88d1..9a178a5e2 100644
--- a/table/deferred_snapshots_test.go
+++ b/table/deferred_snapshots_test.go
@@ -22,6 +22,7 @@ import (
"sync"
"testing"
+ "github.com/apache/iceberg-go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -82,19 +83,19 @@ func TestParseMetadataBytesDeferredSnapshots(t *testing.T) {
common := commonMetadataOf(meta)
require.NotNil(t, common.deferredSnapshots)
require.Len(t, common.SnapshotList, 1)
- assert.Nil(t, common.deferredSnapshots.snapshots)
+ assert.False(t, common.deferredSnapshots.isMaterialized())
assert.Equal(t, eager.CurrentSnapshot(), meta.CurrentSnapshot())
assert.Equal(t, eager.SnapshotByName(MainBranch),
meta.SnapshotByName(MainBranch))
- assert.Nil(t, common.deferredSnapshots.snapshots, "referenced snapshot
access must remain eager")
+ assert.False(t, common.deferredSnapshots.isMaterialized(), "referenced
snapshot access must remain eager")
assert.Equal(t, eager.SnapshotByID(historicalID),
meta.SnapshotByID(historicalID))
- assert.Nil(t, common.deferredSnapshots.snapshots, "historical lookup
must decode only the requested snapshot")
+ assert.False(t, common.deferredSnapshots.isMaterialized(), "historical
lookup must decode only the requested snapshot")
entry :=
&common.deferredSnapshots.entries[common.deferredSnapshots.byID[historicalID]]
assert.Equal(t, historicalID, entry.snapshot.SnapshotID)
assert.Len(t, meta.Snapshots(), 2)
- assert.Len(t, common.deferredSnapshots.snapshots, 2)
+ assert.True(t, common.deferredSnapshots.isMaterialized())
assert.True(t, eager.Equals(meta))
}
@@ -103,11 +104,11 @@ func
TestDeferredSnapshotsMaterializeForCollectionAndSerialization(t *testing.T)
meta, err := ParseMetadataBytesDeferredSnapshots(data)
require.NoError(t, err)
deferredState := commonMetadataOf(meta).deferredSnapshots
- require.NotEmpty(t, deferredState.raw)
+ require.False(t, deferredState.isMaterialized())
serialized, err := json.Marshal(meta)
require.NoError(t, err)
- assert.Nil(t, deferredState.raw)
+ assert.True(t, deferredState.isMaterialized())
assert.Nil(t, deferredState.entries)
assert.Nil(t, deferredState.byID)
reparsed, err := ParseMetadataBytes(serialized)
@@ -131,6 +132,313 @@ func
TestDeferredSnapshotsMaterializeForCollectionAndSerialization(t *testing.T)
assert.Equal(t, eagerRebuilt.Snapshots(), rebuilt.Snapshots())
}
+func TestMetadataBuilderFromDeferredBaseRemainsLazyThroughBuild(t *testing.T) {
+ data := metadataWithUnreferencedSnapshot(t)
+ meta, err := ParseMetadataBytesDeferredSnapshots(data)
+ require.NoError(t, err)
+ state := commonMetadataOf(meta).deferredSnapshots
+ require.NotNil(t, state)
+ require.False(t, state.isMaterialized())
+
+ builder, err := MetadataBuilderFromBase(meta, "")
+ require.NoError(t, err)
+ assert.False(t, state.isMaterialized(), "builder construction must not
materialize deferred history")
+
+ rebuilt, err := builder.Build()
+ require.NoError(t, err)
+ assert.False(t, state.isMaterialized(), "building unchanged metadata
must remain lazy")
+ assert.Equal(t, meta.CurrentSnapshot(), rebuilt.CurrentSnapshot())
+ assert.False(t, state.isMaterialized(), "referenced snapshot access
must remain eager")
+
+ assert.Len(t, rebuilt.Snapshots(), 2)
+ assert.True(t, state.isMaterialized(), "complete collection access must
materialize history")
+}
+
+func TestDeferredBuilderAddsSnapshotWithoutMaterializingHistory(t *testing.T) {
+ data := metadataWithUnreferencedSnapshot(t)
+ meta, err := ParseMetadataBytesDeferredSnapshots(data)
+ require.NoError(t, err)
+ state := commonMetadataOf(meta).deferredSnapshots
+ require.NotNil(t, state)
+
+ builder, err := MetadataBuilderFromBase(meta, "")
+ require.NoError(t, err)
+ current := meta.CurrentSnapshot()
+ require.NotNil(t, current)
+ added := *current
+ added.SnapshotID = 987654321
+ added.ParentSnapshotID = ¤t.SnapshotID
+ added.SequenceNumber = meta.LastSequenceNumber() + 1
+ added.TimestampMs = max(meta.LastUpdatedMillis(), current.TimestampMs)
+ 1
+ require.NoError(t, builder.AddSnapshot(&added))
+ require.NoError(t, builder.SetSnapshotRef(MainBranch, added.SnapshotID,
BranchRef))
+ assert.False(t, state.isMaterialized(), "snapshot addition and ref
update must remain lazy")
+
+ rebuilt, err := builder.Build()
+ require.NoError(t, err)
+ assert.False(t, state.isMaterialized(), "building metadata with an
addition must remain lazy")
+ assert.Equal(t, &added, rebuilt.SnapshotByID(added.SnapshotID))
+ assert.False(t, state.isMaterialized())
+ rebuiltCommon := commonMetadataOf(rebuilt)
+ first, err := rebuiltCommon.snapshotsForMarshal()
+ require.NoError(t, err)
+ second, err := rebuiltCommon.snapshotsForMarshal()
+ require.NoError(t, err)
+ require.NotEmpty(t, first)
+ require.True(t, &first[0] == &second[0], "immutable built metadata must
cache its applied overlay")
+
+ serialized, err := json.Marshal(rebuilt)
+ require.NoError(t, err)
+ reparsed, err := ParseMetadataBytes(serialized)
+ require.NoError(t, err)
+ assert.Len(t, reparsed.Snapshots(), 3)
+ assert.Equal(t, &added, reparsed.SnapshotByID(added.SnapshotID))
+}
+
+func TestDeferredBuilderRemovesSnapshotWithoutResurrection(t *testing.T) {
+ data := metadataWithUnreferencedSnapshot(t)
+ eager, err := ParseMetadataBytes(data)
+ require.NoError(t, err)
+ require.Len(t, eager.Snapshots(), 2)
+ historicalID := eager.Snapshots()[0].SnapshotID
+
+ meta, err := ParseMetadataBytesDeferredSnapshots(data)
+ require.NoError(t, err)
+ state := commonMetadataOf(meta).deferredSnapshots
+ require.NotNil(t, state)
+ builder, err := MetadataBuilderFromBase(meta, "")
+ require.NoError(t, err)
+ require.NoError(t, builder.RemoveSnapshots([]int64{historicalID},
false))
+ assert.False(t, state.isMaterialized(), "snapshot removal must use the
deferred ID index")
+
+ rebuilt, err := builder.Build()
+ require.NoError(t, err)
+ assert.False(t, state.isMaterialized(), "building metadata with a
removal must remain lazy")
+ assert.Nil(t, rebuilt.SnapshotByID(historicalID))
+ assert.False(t, state.isMaterialized(), "removed lookup must not
materialize history")
+
+ serialized, err := json.Marshal(rebuilt)
+ require.NoError(t, err)
+ reparsed, err := ParseMetadataBytes(serialized)
+ require.NoError(t, err)
+ assert.Len(t, reparsed.Snapshots(), 1)
+ assert.Nil(t, reparsed.SnapshotByID(historicalID))
+}
+
+func TestDeferredBuilderCanReplaceRemovedSnapshotID(t *testing.T) {
+ data := metadataWithUnreferencedSnapshot(t)
+ meta, err := ParseMetadataBytesDeferredSnapshots(data)
+ require.NoError(t, err)
+ historical := meta.SnapshotByID(3051729675574597004)
+ require.NotNil(t, historical)
+ state := commonMetadataOf(meta).deferredSnapshots
+ require.NotNil(t, state)
+
+ builder, err := MetadataBuilderFromBase(meta, "")
+ require.NoError(t, err)
+ require.NoError(t,
builder.RemoveSnapshots([]int64{historical.SnapshotID}, false))
+ replacement := *historical
+ replacement.ManifestList = "replacement.avro"
+ current := meta.CurrentSnapshot()
+ require.NotNil(t, current)
+ replacement.ParentSnapshotID = ¤t.SnapshotID
+ replacement.SequenceNumber = meta.LastSequenceNumber() + 1
+ replacement.TimestampMs = meta.LastUpdatedMillis() + 1
+ require.NoError(t, builder.AddSnapshot(&replacement))
+ assert.False(t, state.isMaterialized())
+
+ rebuilt, err := builder.Build()
+ require.NoError(t, err)
+ assert.Equal(t, "replacement.avro",
rebuilt.SnapshotByID(historical.SnapshotID).ManifestList)
+ serialized, err := json.Marshal(rebuilt)
+ require.NoError(t, err)
+ reparsed, err := ParseMetadataBytes(serialized)
+ require.NoError(t, err)
+ require.Len(t, reparsed.Snapshots(), 2)
+ assert.Equal(t, "replacement.avro",
reparsed.SnapshotByID(historical.SnapshotID).ManifestList)
+}
+
+func TestDeferredBuilderOnlyTombstonesBaseSnapshots(t *testing.T) {
+ meta, err :=
ParseMetadataBytesDeferredSnapshots(metadataWithUnreferencedSnapshot(t))
+ require.NoError(t, err)
+ builder, err := MetadataBuilderFromBase(meta, "")
+ require.NoError(t, err)
+
+ current := meta.CurrentSnapshot()
+ require.NotNil(t, current)
+ added := *current
+ added.SnapshotID = 987654321
+ added.ParentSnapshotID = ¤t.SnapshotID
+ added.SequenceNumber = meta.LastSequenceNumber() + 1
+ added.TimestampMs = max(meta.LastUpdatedMillis(), current.TimestampMs)
+ 1
+ require.NoError(t, builder.AddSnapshot(&added))
+ require.NoError(t, builder.RemoveSnapshots([]int64{added.SnapshotID,
-1}, false))
+ assert.Empty(t, builder.removedDeferredSnapshotIDs,
+ "addition-only and unknown IDs must not create deferred-base
tombstones")
+
+ require.NoError(t,
builder.RemoveSnapshots([]int64{3051729675574597004}, false))
+ assert.Contains(t, builder.removedDeferredSnapshotIDs,
int64(3051729675574597004))
+}
+
+func TestDeferredBuilderMutationsSurviveRebuild(t *testing.T) {
+ data := metadataWithUnreferencedSnapshot(t)
+ eager, err := ParseMetadataBytes(data)
+ require.NoError(t, err)
+ deferred, err := ParseMetadataBytesDeferredSnapshots(data)
+ require.NoError(t, err)
+
+ mutate := func(base Metadata) Metadata {
+ first, err := MetadataBuilderFromBase(base, "")
+ require.NoError(t, err)
+ current := base.CurrentSnapshot()
+ require.NotNil(t, current)
+ added := *current
+ added.SnapshotID = 987654321
+ added.ParentSnapshotID = ¤t.SnapshotID
+ added.SequenceNumber = base.LastSequenceNumber() + 1
+ added.TimestampMs = max(base.LastUpdatedMillis(),
current.TimestampMs) + 1
+ require.NoError(t, first.AddSnapshot(&added))
+ require.NoError(t, first.SetSnapshotRef(MainBranch,
added.SnapshotID, BranchRef))
+ require.NoError(t,
first.RemoveSnapshots([]int64{3051729675574597004}, false))
+ firstBuilt, err := first.Build()
+ require.NoError(t, err)
+
+ second, err := MetadataBuilderFromBase(firstBuilt, "")
+ require.NoError(t, err)
+ addedAgain := added
+ addedAgain.SnapshotID = 987654322
+ addedAgain.ParentSnapshotID = &added.SnapshotID
+ addedAgain.SequenceNumber++
+ addedAgain.TimestampMs++
+ require.NoError(t, second.AddSnapshot(&addedAgain))
+ require.NoError(t, second.SetSnapshotRef(MainBranch,
addedAgain.SnapshotID, BranchRef))
+ secondBuilt, err := second.Build()
+ require.NoError(t, err)
+
+ return secondBuilt
+ }
+
+ eagerBuilt := mutate(eager)
+ deferredBuilt := mutate(deferred)
+ assert.True(t, eagerBuilt.Equals(deferredBuilt))
+ eagerJSON, err := json.Marshal(eagerBuilt)
+ require.NoError(t, err)
+ deferredJSON, err := json.Marshal(deferredBuilt)
+ require.NoError(t, err)
+ assert.JSONEq(t, string(eagerJSON), string(deferredJSON))
+}
+
+func TestDeferredBuilderSetSnapshotRefUsesLogicalSnapshots(t *testing.T) {
+ const historicalID = int64(3051729675574597004)
+
+ t.Run("removed deferred snapshot", func(t *testing.T) {
+ meta, err :=
ParseMetadataBytesDeferredSnapshots(metadataWithUnreferencedSnapshot(t))
+ require.NoError(t, err)
+ builder, err := MetadataBuilderFromBase(meta, "")
+ require.NoError(t, err)
+ require.NoError(t,
builder.RemoveSnapshots([]int64{historicalID}, false))
+
+ err = builder.SetSnapshotRef("history", historicalID, TagRef)
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+ })
+
+ t.Run("unreferenced deferred snapshot", func(t *testing.T) {
+ meta, err :=
ParseMetadataBytesDeferredSnapshots(metadataWithUnreferencedSnapshot(t))
+ require.NoError(t, err)
+ state := commonMetadataOf(meta).deferredSnapshots
+ builder, err := MetadataBuilderFromBase(meta, "")
+ require.NoError(t, err)
+
+ require.NoError(t, builder.SetSnapshotRef("history",
historicalID, TagRef))
+ assert.False(t, state.isMaterialized(), "indexed ref validation
must not decode full history")
+ })
+}
+
+func TestDeferredBuilderCloneOwnsSnapshotOverlay(t *testing.T) {
+ data := metadataWithUnreferencedSnapshot(t)
+ eager, err := ParseMetadataBytes(data)
+ require.NoError(t, err)
+ historicalID := eager.Snapshots()[0].SnapshotID
+
+ meta, err := ParseMetadataBytesDeferredSnapshots(data)
+ require.NoError(t, err)
+ builder, err := MetadataBuilderFromBase(meta, "")
+ require.NoError(t, err)
+ cloned := builder.clone()
+ require.NoError(t, cloned.RemoveSnapshots([]int64{historicalID}, false))
+
+ originalBuilt, err := builder.Build()
+ require.NoError(t, err)
+ cloneBuilt, err := cloned.Build()
+ require.NoError(t, err)
+ assert.NotNil(t, originalBuilt.SnapshotByID(historicalID))
+ assert.Nil(t, cloneBuilt.SnapshotByID(historicalID))
+ assert.Len(t, originalBuilt.Snapshots(), 2)
+ assert.Len(t, cloneBuilt.Snapshots(), 1)
+}
+
+func TestDeferredBuiltMetadataOwnsSnapshotOverlay(t *testing.T) {
+ meta, err :=
ParseMetadataBytesDeferredSnapshots(metadataWithUnreferencedSnapshot(t))
+ require.NoError(t, err)
+ builder, err := MetadataBuilderFromBase(meta, "")
+ require.NoError(t, err)
+ current := meta.CurrentSnapshot()
+ require.NotNil(t, current)
+ added := *current
+ added.SnapshotID = 987654321
+ added.ParentSnapshotID = ¤t.SnapshotID
+ added.SequenceNumber = meta.LastSequenceNumber() + 1
+ added.TimestampMs = max(meta.LastUpdatedMillis(), current.TimestampMs)
+ 1
+ require.NoError(t, builder.AddSnapshot(&added))
+
+ built, err := builder.Build()
+ require.NoError(t, err)
+ require.NoError(t, builder.RemoveSnapshots([]int64{added.SnapshotID},
false))
+ assert.NotNil(t, built.SnapshotByID(added.SnapshotID))
+ assert.Len(t, built.Snapshots(), 3)
+ _, err = builder.SnapshotByID(added.SnapshotID)
+ assert.ErrorIs(t, err, ErrSnapshotNotFound)
+}
+
+func TestDeferredBuilderRoundTripAllMetadataVersions(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ data string
+ }{
+ {name: "v1", data: ExampleTableMetadataV1},
+ {name: "v2", data: ExampleTableMetadataV2},
+ {name: "v3", data: ExampleTableMetadataV3},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ var fields map[string]json.RawMessage
+ require.NoError(t, json.Unmarshal([]byte(tc.data),
&fields))
+ fields["refs"] = json.RawMessage(`{}`)
+ data, err := json.Marshal(fields)
+ require.NoError(t, err)
+
+ eager, err := ParseMetadataBytes(data)
+ require.NoError(t, err)
+ eagerBuilder, err := MetadataBuilderFromBase(eager, "")
+ require.NoError(t, err)
+ deferred, err :=
ParseMetadataBytesDeferredSnapshots(data)
+ require.NoError(t, err)
+ state := commonMetadataOf(deferred).deferredSnapshots
+ require.NotNil(t, state)
+ builder, err := MetadataBuilderFromBase(deferred, "")
+ require.NoError(t, err)
+ const builtTimestamp = int64(1800000000000)
+ eagerBuilder.lastUpdatedMS = builtTimestamp
+ builder.lastUpdatedMS = builtTimestamp
+ eagerRebuilt, err := eagerBuilder.Build()
+ require.NoError(t, err)
+ rebuilt, err := builder.Build()
+ require.NoError(t, err)
+ assert.False(t, state.isMaterialized())
+ assert.True(t, eagerRebuilt.Equals(rebuilt))
+ })
+ }
+}
+
func TestDeferredSnapshotsV1ToV2MaterializesHistory(t *testing.T) {
var fields map[string]json.RawMessage
require.NoError(t, json.Unmarshal([]byte(ExampleTableMetadataV1),
&fields))
@@ -395,7 +703,7 @@ func TestDeferredSnapshotsConcurrentLookupAndIndexRelease(t
*testing.T) {
}
state := commonMetadataOf(meta).deferredSnapshots
- assert.Nil(t, state.raw)
+ assert.True(t, state.isMaterialized())
assert.Nil(t, state.entries)
assert.Nil(t, state.byID)
}
@@ -415,7 +723,7 @@ func
TestDeferredSnapshotsConcurrentSingleLookupDoesNotMaterializeHistory(t *tes
wg.Wait()
common := commonMetadataOf(meta)
- assert.Nil(t, common.deferredSnapshots.snapshots)
+ assert.False(t, common.deferredSnapshots.isMaterialized())
entry :=
&common.deferredSnapshots.entries[common.deferredSnapshots.byID[historicalID]]
assert.Equal(t, historicalID, entry.snapshot.SnapshotID)
}
diff --git a/table/metadata.go b/table/metadata.go
index 4b26e898f..3caf45d71 100644
--- a/table/metadata.go
+++ b/table/metadata.go
@@ -431,15 +431,20 @@ type MetadataBuilder struct {
props iceberg.Properties
snapshotList []Snapshot
snapshotIndex *snapshotIndexData // Derived from snapshotList; not
serialized.
- currentSnapshotID *int64
- snapshotLog []SnapshotLogEntry
- metadataLog []MetadataLogEntry
- sortOrderList []SortOrder
- defaultSortOrderID int
- refs map[string]SnapshotRef
- statisticsList []StatisticsFile
- partitionStatsList []PartitionStatisticsFile
- encryptionKeyList []EncryptionKey
+ deferredSnapshots *deferredSnapshotState
+ // removedDeferredSnapshotIDs contains tombstones only for IDs in the
+ // immutable deferred base. Additions are derived from snapshotList at
Build
+ // time so snapshotList remains the builder's single mutable source of
truth.
+ removedDeferredSnapshotIDs map[int64]struct{}
+ currentSnapshotID *int64
+ snapshotLog []SnapshotLogEntry
+ metadataLog []MetadataLogEntry
+ sortOrderList []SortOrder
+ defaultSortOrderID int
+ refs map[string]SnapshotRef
+ statisticsList []StatisticsFile
+ partitionStatsList []PartitionStatisticsFile
+ encryptionKeyList []EncryptionKey
previousFileEntry *MetadataLogEntry
// >v1 specific
@@ -515,11 +520,14 @@ func MetadataBuilderFromBase(metadata Metadata,
currentFileLocation string) (*Me
if b.props == nil {
b.props = iceberg.Properties{}
}
- snapshots, err := common.snapshotsForMarshal()
- if err != nil {
- return nil, err
+ b.snapshotList = cloneSnapshots(common.SnapshotList)
+ if common.deferredSnapshots != nil {
+ b.deferredSnapshots = common.deferredSnapshots
+ b.removedDeferredSnapshotIDs = make(map[int64]struct{})
+ if common.snapshotOverlay != nil {
+ maps.Copy(b.removedDeferredSnapshotIDs,
common.snapshotOverlay.removed)
+ }
}
- b.snapshotList = cloneSnapshots(snapshots)
b.snapshotLog = cloneCollected(common.SnapshotLog)
b.metadataLog = cloneCollected(common.MetadataLog)
b.sortOrderList = cloneSortOrders(common.SortOrderList)
@@ -541,7 +549,7 @@ func MetadataBuilderFromBase(metadata Metadata,
currentFileLocation string) (*Me
b.nextRowID = &nextRowID
}
if common.CurrentSnapshotID != nil {
- if _, ok := snapshotIndexPosition(b.snapshotIndex,
b.snapshotList, *common.CurrentSnapshotID); ok {
+ if b.containsSnapshotID(*common.CurrentSnapshotID) {
b.currentSnapshotID =
clonePtr(common.CurrentSnapshotID)
}
}
@@ -612,35 +620,37 @@ func (b *MetadataBuilder) clone() *MetadataBuilder {
}
cloned := &MetadataBuilder{
- base: b.base,
- updates: append([]Update(nil), b.updates...),
- formatVersion: b.formatVersion,
- uuid: b.uuid,
- loc: b.loc,
- lastUpdatedMS: b.lastUpdatedMS,
- lastColumnId: b.lastColumnId,
- schemaList: slices.Clone(b.schemaList),
- currentSchemaID: b.currentSchemaID,
- specs: slices.Clone(b.specs),
- defaultSpecID: b.defaultSpecID,
- lastPartitionID: clonePtr(b.lastPartitionID),
- props: maps.Clone(b.props),
- snapshotList: slices.Clone(b.snapshotList),
- currentSnapshotID: clonePtr(b.currentSnapshotID),
- snapshotLog: slices.Clone(b.snapshotLog),
- metadataLog: slices.Clone(b.metadataLog),
- sortOrderList: slices.Clone(b.sortOrderList),
- defaultSortOrderID: b.defaultSortOrderID,
- refs: maps.Clone(b.refs),
- statisticsList: slices.Clone(b.statisticsList),
- partitionStatsList: slices.Clone(b.partitionStatsList),
- encryptionKeyList: slices.Clone(b.encryptionKeyList),
- previousFileEntry: clonePtr(b.previousFileEntry),
- lastSequenceNumber: clonePtr(b.lastSequenceNumber),
- nextRowID: clonePtr(b.nextRowID),
- lastAddedSchemaID: clonePtr(b.lastAddedSchemaID),
- lastAddedPartitionID: clonePtr(b.lastAddedPartitionID),
- lastAddedSortOrderID: clonePtr(b.lastAddedSortOrderID),
+ base: b.base,
+ updates: append([]Update(nil), b.updates...),
+ formatVersion: b.formatVersion,
+ uuid: b.uuid,
+ loc: b.loc,
+ lastUpdatedMS: b.lastUpdatedMS,
+ lastColumnId: b.lastColumnId,
+ schemaList: slices.Clone(b.schemaList),
+ currentSchemaID: b.currentSchemaID,
+ specs: slices.Clone(b.specs),
+ defaultSpecID: b.defaultSpecID,
+ lastPartitionID: clonePtr(b.lastPartitionID),
+ props: maps.Clone(b.props),
+ snapshotList: slices.Clone(b.snapshotList),
+ deferredSnapshots: b.deferredSnapshots,
+ removedDeferredSnapshotIDs:
maps.Clone(b.removedDeferredSnapshotIDs),
+ currentSnapshotID: clonePtr(b.currentSnapshotID),
+ snapshotLog: slices.Clone(b.snapshotLog),
+ metadataLog: slices.Clone(b.metadataLog),
+ sortOrderList: slices.Clone(b.sortOrderList),
+ defaultSortOrderID: b.defaultSortOrderID,
+ refs: maps.Clone(b.refs),
+ statisticsList: slices.Clone(b.statisticsList),
+ partitionStatsList: slices.Clone(b.partitionStatsList),
+ encryptionKeyList: slices.Clone(b.encryptionKeyList),
+ previousFileEntry: clonePtr(b.previousFileEntry),
+ lastSequenceNumber: clonePtr(b.lastSequenceNumber),
+ nextRowID: clonePtr(b.nextRowID),
+ lastAddedSchemaID: clonePtr(b.lastAddedSchemaID),
+ lastAddedPartitionID: clonePtr(b.lastAddedPartitionID),
+ lastAddedSortOrderID: clonePtr(b.lastAddedSortOrderID),
}
if b.schemaIndex != nil {
cloned.schemaIndex = &schemaIndexData{
@@ -702,16 +712,26 @@ func (b *MetadataBuilder) nextSequenceNumber() int64 {
}
func (b *MetadataBuilder) newSnapshotID() int64 {
- b.ensureSnapshotIndex()
-
for {
snapshotID := generateSnapshotID()
- if _, exists := b.snapshotIndex.positions[snapshotID]; !exists {
+ if !b.containsSnapshotID(snapshotID) {
return snapshotID
}
}
}
+func (b *MetadataBuilder) containsSnapshotID(id int64) bool {
+ b.ensureSnapshotIndex()
+ if _, ok := snapshotIndexPosition(b.snapshotIndex, b.snapshotList, id);
ok {
+ return true
+ }
+ if _, removed := b.removedDeferredSnapshotIDs[id]; removed {
+ return false
+ }
+
+ return b.deferredSnapshots != nil && b.deferredSnapshots.containsID(id)
+}
+
func (b *MetadataBuilder) ensureSchemaIndex() {
if schemaIndexNeedsRebuild(b.schemaIndex, b.schemaList) {
b.schemaIndex = buildSchemaIndex(b.schemaList)
@@ -1064,17 +1084,19 @@ func (b *MetadataBuilder) RemoveSnapshots(snapshotIds
[]int64, postCommit bool)
if len(b.snapshotList) != previousSnapshotCount {
b.snapshotIndex = buildSnapshotIndex(b.snapshotList)
}
+ if b.deferredSnapshots != nil {
+ for id := range removedIDs {
+ if b.deferredSnapshots.containsID(id) {
+ b.removedDeferredSnapshotIDs[id] = struct{}{}
+ }
+ }
+ }
// Snapshot-log pruning is deferred to updatedSnapshotLog during Build
so
// removed entries remain available when trimming history gaps.
- validSnapshotIDs := make(map[int64]struct{}, len(b.snapshotList))
- for _, snapshot := range b.snapshotList {
- validSnapshotIDs[snapshot.SnapshotID] = struct{}{}
- }
-
newRefs := make(map[string]SnapshotRef, len(b.refs))
for name, ref := range b.refs {
- if _, ok := validSnapshotIDs[ref.SnapshotID]; ok {
+ if b.containsSnapshotID(ref.SnapshotID) {
newRefs[name] = ref
}
}
@@ -1490,6 +1512,43 @@ func (b *MetadataBuilder) SetLastUpdatedMS()
*MetadataBuilder {
return b
}
+func (b *MetadataBuilder) deferredSnapshotOverlay() *deferredSnapshotOverlay {
+ if b.deferredSnapshots == nil {
+ return nil
+ }
+
+ additions := make([]Snapshot, 0)
+ for _, snapshot := range b.snapshotList {
+ _, removedBase :=
b.removedDeferredSnapshotIDs[snapshot.SnapshotID]
+ if removedBase ||
!b.deferredSnapshots.containsID(snapshot.SnapshotID) {
+ additions = append(additions, snapshot)
+ }
+ }
+ if len(additions) == 0 && len(b.removedDeferredSnapshotIDs) == 0 {
+ return nil
+ }
+
+ return &deferredSnapshotOverlay{
+ additions: cloneSnapshots(additions),
+ removed: maps.Clone(b.removedDeferredSnapshotIDs),
+ }
+}
+
+// snapshotsForFullScan intentionally materializes deferred history for builder
+// operations, such as expiration, whose result depends on every snapshot.
+func (b *MetadataBuilder) snapshotsForFullScan() ([]Snapshot, error) {
+ if b.deferredSnapshots == nil {
+ return b.snapshotList, nil
+ }
+
+ snapshots, _, err := b.deferredSnapshots.load()
+ if err != nil {
+ return nil, err
+ }
+
+ return b.deferredSnapshotOverlay().apply(snapshots), nil
+}
+
func (b *MetadataBuilder) buildCommonMetadata() (*commonMetadata, error) {
b.ensureSchemaIndex()
if b.schemaIndex != nil {
@@ -1532,6 +1591,14 @@ func (b *MetadataBuilder) buildCommonMetadata()
(*commonMetadata, error) {
MetadataPreviousVersionsMaxKey,
MetadataPreviousVersionsMaxDefault))
metadataLog = trimMetadataLog(append(metadataLog,
*b.previousFileEntry), maxMetadataLogEntries)
}
+ snapshotList := b.snapshotList
+ snapshotIndex := b.snapshotIndex
+ if b.deferredSnapshots != nil {
+ // Deferred builders remain mutable after Build. Keep the
eagerly known
+ // snapshots in built metadata independent from later overlay
mutations.
+ snapshotList = cloneSnapshots(b.snapshotList)
+ snapshotIndex = buildSnapshotIndex(snapshotList)
+ }
return &commonMetadata{
FormatVersion: b.formatVersion,
@@ -1547,8 +1614,10 @@ func (b *MetadataBuilder) buildCommonMetadata()
(*commonMetadata, error) {
DefaultSpecID: defaultSpecID,
LastPartitionID: b.lastPartitionID,
Props: b.props,
- SnapshotList: b.snapshotList,
- snapshotIndex: b.snapshotIndex,
+ SnapshotList: snapshotList,
+ snapshotIndex: snapshotIndex,
+ deferredSnapshots: b.deferredSnapshots,
+ snapshotOverlay: b.deferredSnapshotOverlay(),
CurrentSnapshotID: b.currentSnapshotID,
SnapshotLog: snapshotLog,
MetadataLog: metadataLog,
@@ -1587,14 +1656,9 @@ func (b *MetadataBuilder) updatedSnapshotLog()
([]SnapshotLogEntry, error) {
}
}
if len(intermediateIDs) != 0 || hasRemoved {
- validSnapshotIDs := make(map[int64]struct{},
len(b.snapshotList))
- for _, snapshot := range b.snapshotList {
- validSnapshotIDs[snapshot.SnapshotID] = struct{}{}
- }
-
newSnapsLog := make([]SnapshotLogEntry, 0, len(b.snapshotLog))
for _, s := range b.snapshotLog {
- if _, ok := validSnapshotIDs[s.SnapshotID]; ok {
+ if b.containsSnapshotID(s.SnapshotID) {
if _, ok := intermediateIDs[s.SnapshotID]; !ok {
newSnapsLog = append(newSnapsLog, s)
}
@@ -1662,6 +1726,18 @@ func (b *MetadataBuilder) SnapshotByID(id int64)
(*Snapshot, error) {
return &snapshot, nil
}
+ if _, removed := b.removedDeferredSnapshotIDs[id]; removed {
+ return nil, fmt.Errorf("%w: id %d", ErrSnapshotNotFound, id)
+ }
+ if b.deferredSnapshots != nil {
+ snapshot, err := b.deferredSnapshots.snapshotByID(id)
+ if err != nil {
+ return nil, err
+ }
+ if snapshot != nil {
+ return snapshot, nil
+ }
+ }
return nil, fmt.Errorf("%w: id %d", ErrSnapshotNotFound, id)
}
@@ -2307,6 +2383,7 @@ type commonMetadata struct {
partitionSpecIndex *partitionSpecIndexData
snapshotIndex *snapshotIndexData
deferredSnapshots *deferredSnapshotState
+ snapshotOverlay *deferredSnapshotOverlay
}
func (c *commonMetadata) metadataBuilderCommon() *commonMetadata { return c }
@@ -2546,6 +2623,9 @@ func (c *commonMetadata) SnapshotByID(id int64) *Snapshot
{
if ok {
return cloneSnapshotPtr(&c.SnapshotList[i])
}
+ if c.snapshotOverlay != nil && c.snapshotOverlay.isRemoved(id) {
+ return nil
+ }
if c.deferredSnapshots != nil {
snapshot, err := c.deferredSnapshots.snapshotByID(id)
if err != nil {
@@ -2580,8 +2660,11 @@ func (c *commonMetadata) snapshotsForMarshal()
([]Snapshot, error) {
}
snapshots, _, err := c.deferredSnapshots.load()
+ if err != nil {
+ return nil, err
+ }
- return snapshots, err
+ return c.snapshotOverlay.apply(snapshots), nil
}
func (c *commonMetadata) SnapshotByName(name string) *Snapshot {
@@ -3245,6 +3328,7 @@ func (m *metadataV1) ToV2() metadataV2 {
commonOut.SnapshotList = m.allSnapshots()
commonOut.snapshotIndex = buildSnapshotIndex(commonOut.SnapshotList)
commonOut.deferredSnapshots = nil
+ commonOut.snapshotOverlay = nil
commonOut.FormatVersion = 2
if commonOut.UUID == uuid.Nil {
commonOut.UUID = uuid.New()
diff --git a/table/metadata_builder_internal_test.go
b/table/metadata_builder_internal_test.go
index 5030647e2..639da6d7b 100644
--- a/table/metadata_builder_internal_test.go
+++ b/table/metadata_builder_internal_test.go
@@ -3785,6 +3785,7 @@ func TestSetFormatVersionV2ToV3FromDeserializedMetadata(t
*testing.T) {
// both the drift guard and its filler consult it.
var sharedCloneFields = map[string]struct{}{
"base": {}, // immutable snapshot, shared by design.
+ "deferredSnapshots": {}, // immutable snapshot source with
concurrency-safe caches.
"schemaIndex": {}, // immutable schema references, copied on
schema mutation.
"partitionSpecIndex": {}, // immutable index positions, copied on spec
mutation.
"snapshotIndex": {}, // immutable index positions, copied on
snapshot mutation.
diff --git a/table/transaction.go b/table/transaction.go
index 103ed5ba7..a90c9f83b 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -783,6 +783,10 @@ func (t *Transaction) ExpireSnapshots(opts
...ExpireSnapshotsOpt) error {
if cfg.maxSnapshotAgeMs != nil {
defaultMaxSnapshotAgeMs = *cfg.maxSnapshotAgeMs
}
+ snapshots, err := meta.snapshotsForFullScan()
+ if err != nil {
+ return err
+ }
lookup := func(id int64) *Snapshot {
s, _ := meta.SnapshotByID(id)
@@ -855,7 +859,7 @@ func (t *Transaction) ExpireSnapshots(opts
...ExpireSnapshotsOpt) error {
}
unreferencedSnapshotCutoff := nowMs - defaultMaxSnapshotAgeMs
- for _, snap := range meta.snapshotList {
+ for _, snap := range snapshots {
if _, referenced := referencedSnapshots[snap.SnapshotID];
referenced {
continue
}
@@ -867,7 +871,7 @@ func (t *Transaction) ExpireSnapshots(opts
...ExpireSnapshotsOpt) error {
var snapsToDelete []int64
- for _, snap := range meta.snapshotList {
+ for _, snap := range snapshots {
if _, found := snapsToKeep[snap.SnapshotID]; !found {
snapsToDelete = append(snapsToDelete, snap.SnapshotID)
}
diff --git a/table/transaction_internal_test.go
b/table/transaction_internal_test.go
index 4049d7b59..4a26c07b1 100644
--- a/table/transaction_internal_test.go
+++ b/table/transaction_internal_test.go
@@ -582,6 +582,38 @@ func
TestExpireSnapshotsWithOlderThanDoesNotExpireSnapshotRefs(t *testing.T) {
require.True(t, tagExists, "WithOlderThan must not expire a tag without
max-ref-age-ms")
}
+func TestExpireSnapshotsMaterializesDeferredHistory(t *testing.T) {
+ const historicalID = int64(3051729675574597004)
+
+ meta, err :=
ParseMetadataBytesDeferredSnapshots(metadataWithUnreferencedSnapshot(t))
+ require.NoError(t, err)
+ state := commonMetadataOf(meta).deferredSnapshots
+ require.NotNil(t, state)
+ builder, err := MetadataBuilderFromBase(meta, "")
+ require.NoError(t, err)
+ txn := &Transaction{
+ tbl: &Table{metadata: meta},
+ meta: builder,
+ reqs: []Requirement{},
+ }
+
+ require.NoError(t, txn.ExpireSnapshots(
+ WithOlderThan(0),
+ WithRetainLast(1),
+ WithPostCommit(false),
+ ))
+ assert.True(t, state.isMaterialized(), "expiration must inspect
complete deferred history")
+
+ var removal *removeSnapshotsUpdate
+ for _, update := range txn.meta.updates {
+ if candidate, ok := update.(*removeSnapshotsUpdate); ok {
+ removal = candidate
+ }
+ }
+ require.NotNil(t, removal)
+ assert.Contains(t, removal.SnapshotIDs, historicalID)
+}
+
func TestExpireSnapshotsUsesStandardRetentionPropertyNames(t *testing.T) {
tests := []struct {
name string
