You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/07/17 15:30:58 UTC
(iceberg-go) branch main updated: feat(table): add snapshots metadata table (#1476)
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 c39baf45 feat(table): add snapshots metadata table (#1476)
c39baf45 is described below
commit c39baf45b387e83bef8301ef19617bd9346f8d68
Author: Andrei Tserakhau <la...@gmail.com>
AuthorDate: Fri Jul 17 17:30:53 2026 +0200
feat(table): add snapshots metadata table (#1476)
---
table/inspect.go | 181 ++++++++++++++++++++++++++++++++++----
table/inspect_internal_test.go | 195 ++++++++++++++++++++++++++++++++++++++++-
2 files changed, 359 insertions(+), 17 deletions(-)
diff --git a/table/inspect.go b/table/inspect.go
index 90aa1645..a8bbf6fd 100644
--- a/table/inspect.go
+++ b/table/inspect.go
@@ -34,12 +34,34 @@ import (
// Obtain one via Table.Inspect. Each method returns an array.RecordReader that
// the caller is responsible for releasing.
type InspectTable struct {
- tbl Table
+ tbl Table
+ alloc memory.Allocator
}
-// Inspect returns an InspectTable for reading this table's metadata tables.
-func (t Table) Inspect() InspectTable {
- return InspectTable{tbl: t}
+// Inspect returns an InspectTable for reading this table's metadata tables,
+// allocating Arrow buffers from opts, if provided, otherwise the default
+// allocator.
+func (t Table) Inspect(opts ...InspectOption) InspectTable {
+ i := InspectTable{tbl: t, alloc: memory.DefaultAllocator}
+ for _, opt := range opts {
+ opt(&i)
+ }
+
+ return i
+}
+
+// InspectOption configures an InspectTable.
+type InspectOption func(*InspectTable)
+
+// WithInspectAllocator sets the Arrow memory allocator used to build
+// metadata-table records. Tests can pass a memory.CheckedAllocator to detect
+// leaks; callers with memory accounting can inject their own pool.
+func WithInspectAllocator(alloc memory.Allocator) InspectOption {
+ return func(i *InspectTable) {
+ if alloc != nil {
+ i.alloc = alloc
+ }
+ }
}
// History returns the chronological log of every snapshot that was ever the
@@ -63,9 +85,7 @@ func (t Table) Inspect() InspectTable {
//
// The returned reader holds a single record batch. The caller must Release it.
func (i InspectTable) History(ctx context.Context) (array.RecordReader, error) {
- sc := historySchema()
-
- arrowSchema, err := SchemaToArrowSchema(sc, nil, true, false)
+ arrowSchema, err := SchemaToArrowSchema(HistorySchema(), nil, true, false)
if err != nil {
return nil, fmt.Errorf("inspect history: build arrow schema: %w", err)
}
@@ -81,10 +101,10 @@ func (i InspectTable) History(ctx context.Context) (array.RecordReader, error) {
}
}
- bldr := array.NewRecordBuilder(memory.DefaultAllocator, arrowSchema)
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
defer bldr.Release()
- // Field positions and concrete builder types follow historySchema; the
+ // Field positions and concrete builder types follow HistorySchema; the
// assertions are safe as long as the two stay in sync.
madeCurrentAt := bldr.Field(0).(*array.TimestampBuilder)
snapshotID := bldr.Field(1).(*array.Int64Builder)
@@ -113,22 +133,125 @@ func (i InspectTable) History(ctx context.Context) (array.RecordReader, error) {
isCurrentAncestor.Append(ok)
}
+ rr, err := singleBatchReader(arrowSchema, bldr)
+ if err != nil {
+ return nil, fmt.Errorf("inspect history: %w", err)
+ }
+
+ return rr, nil
+}
+
+// singleBatchReader finishes bldr into one record batch and wraps it in a
+// RecordReader. NewRecordReader retains the batch, so we release our own
+// reference unconditionally: on success the reader owns it, and on error
+// NewRecordReader has already released its own retain. This avoids a deferred
+// double-release.
+func singleBatchReader(arrowSchema *arrow.Schema, bldr *array.RecordBuilder) (array.RecordReader, error) {
rec := bldr.NewRecordBatch()
- // NewRecordReader retains rec, so release our reference unconditionally:
- // on success the reader owns it, and on error NewRecordReader has already
- // released its own retain. This avoids a deferred double-release.
rr, err := array.NewRecordReader(arrowSchema, []arrow.RecordBatch{rec})
rec.Release()
if err != nil {
- return nil, fmt.Errorf("inspect history: new record reader: %w", err)
+ return nil, fmt.Errorf("new record reader: %w", err)
}
return rr, nil
}
-// historySchema returns the Iceberg schema of the history metadata table. The
-// field IDs match the Java, PyIceberg, and Rust clients for cross-client parity.
-func historySchema() *iceberg.Schema {
+// Snapshots returns one row per snapshot known to the table, in the order they
+// are stored in metadata.
+//
+// Columns:
+// - committed_at (timestamptz, required): when the snapshot was committed
+// - snapshot_id (long, required): the snapshot id
+// - parent_id (long, optional): the parent snapshot id, null for a root
+// - operation (string, optional): the snapshot summary operation, null when
+// the snapshot carries no summary
+// - manifest_list (string, optional): path to the snapshot's manifest list
+// - summary (map<string,string>, optional): the stored snapshot summary,
+// including the "operation" key alongside its additional properties; null
+// when the snapshot carries no summary
+//
+// The returned reader holds a single record batch. The caller must Release it.
+func (i InspectTable) Snapshots(ctx context.Context) (array.RecordReader, error) {
+ arrowSchema, err := SchemaToArrowSchema(SnapshotsSchema(), nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect snapshots: build arrow schema: %w", err)
+ }
+
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
+ defer bldr.Release()
+
+ // Field positions and concrete builder types follow SnapshotsSchema; the
+ // assertions are safe as long as the two stay in sync.
+ committedAt := bldr.Field(0).(*array.TimestampBuilder)
+ snapshotID := bldr.Field(1).(*array.Int64Builder)
+ parentID := bldr.Field(2).(*array.Int64Builder)
+ operation := bldr.Field(3).(*array.StringBuilder)
+ manifestList := bldr.Field(4).(*array.StringBuilder)
+ summary := bldr.Field(5).(*array.MapBuilder)
+ summaryKeys := summary.KeyBuilder().(*array.StringBuilder)
+ summaryValues := summary.ItemBuilder().(*array.StringBuilder)
+
+ for _, snap := range i.tbl.metadata.Snapshots() {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ // Iceberg stores timestamps as epoch milliseconds; Arrow uses
+ // microseconds for the timestamptz type.
+ committedAt.Append(arrow.Timestamp(snap.TimestampMs * 1000))
+ snapshotID.Append(snap.SnapshotID)
+
+ if snap.ParentSnapshotID != nil {
+ parentID.Append(*snap.ParentSnapshotID)
+ } else {
+ parentID.AppendNull()
+ }
+
+ // operation and summary come from the snapshot summary, which is
+ // optional: a snapshot without one renders both as null.
+ if snap.Summary != nil {
+ operation.Append(string(snap.Summary.Operation))
+ summary.Append(true)
+ // The summary map mirrors the stored/serialized summary, which
+ // folds the operation back in under the "operation" key (see
+ // Summary.MarshalJSON). Emit it alongside the extra properties so
+ // the table faithfully reflects the on-disk summary.
+ if snap.Summary.Operation != "" {
+ summaryKeys.Append(operationKey)
+ summaryValues.Append(string(snap.Summary.Operation))
+ }
+ for k, v := range snap.Summary.Properties {
+ summaryKeys.Append(k)
+ summaryValues.Append(v)
+ }
+ } else {
+ operation.AppendNull()
+ summary.AppendNull()
+ }
+
+ // manifest_list is optional: a snapshot without a manifest-list path
+ // (e.g. some V1 snapshots) renders null rather than an empty string.
+ if snap.ManifestList != "" {
+ manifestList.Append(snap.ManifestList)
+ } else {
+ manifestList.AppendNull()
+ }
+ }
+
+ rr, err := singleBatchReader(arrowSchema, bldr)
+ if err != nil {
+ return nil, fmt.Errorf("inspect snapshots: %w", err)
+ }
+
+ return rr, nil
+}
+
+// HistorySchema returns the Iceberg schema of the history metadata table. The
+// field IDs are fixed by the Iceberg metadata-tables spec and match the Java,
+// PyIceberg, and Rust clients for cross-client parity. A fresh schema value is
+// returned on each call; callers must not mutate it or rely on pointer identity.
+func HistorySchema() *iceberg.Schema {
return iceberg.NewSchema(0,
iceberg.NestedField{ID: 1, Name: "made_current_at", Type: iceberg.PrimitiveTypes.TimestampTz, Required: true},
iceberg.NestedField{ID: 2, Name: "snapshot_id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
@@ -136,3 +259,29 @@ func historySchema() *iceberg.Schema {
iceberg.NestedField{ID: 4, Name: "is_current_ancestor", Type: iceberg.PrimitiveTypes.Bool, Required: true},
)
}
+
+// SnapshotsSchema returns the Iceberg schema of the snapshots metadata table.
+// The field IDs are fixed by the Iceberg metadata-tables spec and match the
+// Java, PyIceberg, and Rust clients for cross-client parity. A fresh schema
+// value is returned on each call; callers must not mutate it or rely on pointer
+// identity.
+//
+// The summary map's values are optional, matching PyIceberg and Rust. Java
+// declares them required; the choice is not observable in practice because
+// snapshot-summary values are always non-null strings.
+func SnapshotsSchema() *iceberg.Schema {
+ return iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "committed_at", Type: iceberg.PrimitiveTypes.TimestampTz, Required: true},
+ iceberg.NestedField{ID: 2, Name: "snapshot_id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 3, Name: "parent_id", Type: iceberg.PrimitiveTypes.Int64, Required: false},
+ iceberg.NestedField{ID: 4, Name: "operation", Type: iceberg.PrimitiveTypes.String, Required: false},
+ iceberg.NestedField{ID: 5, Name: "manifest_list", Type: iceberg.PrimitiveTypes.String, Required: false},
+ iceberg.NestedField{ID: 6, Name: "summary", Required: false, Type: &iceberg.MapType{
+ KeyID: 7,
+ KeyType: iceberg.PrimitiveTypes.String,
+ ValueID: 8,
+ ValueType: iceberg.PrimitiveTypes.String,
+ ValueRequired: false,
+ }},
+ )
+}
diff --git a/table/inspect_internal_test.go b/table/inspect_internal_test.go
index db408fe9..1450499c 100644
--- a/table/inspect_internal_test.go
+++ b/table/inspect_internal_test.go
@@ -23,6 +23,7 @@ import (
"github.com/apache/arrow-go/v18/arrow"
"github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
"github.com/apache/iceberg-go"
"github.com/google/uuid"
"github.com/stretchr/testify/require"
@@ -96,7 +97,7 @@ func collectRecord(t *testing.T, rr array.RecordReader) arrow.RecordBatch {
}
func TestInspectHistorySchema(t *testing.T) {
- sc := historySchema()
+ sc := HistorySchema()
require.Equal(t, []string{"made_current_at", "snapshot_id", "parent_id", "is_current_ancestor"},
testFieldNames(sc))
@@ -244,6 +245,198 @@ func TestInspectHistoryNoCurrentSnapshot(t *testing.T) {
require.False(t, isCurrentAncestor.Value(0), "no current snapshot means no ancestors")
}
+// snapshotsTestTable builds a table with two snapshots: a root carrying a
+// summary (operation + properties) and a child with no summary at all, to
+// exercise both the populated and null operation/summary paths.
+func snapshotsTestTable() *Table {
+ const (
+ s1 = int64(101)
+ s2 = int64(102)
+ )
+ current := s2
+ lastPartitionID := 999
+
+ meta := &metadataV2{commonMetadata: commonMetadata{
+ FormatVersion: 2,
+ UUID: uuid.New(),
+ Loc: "s3://test/snapshots",
+ LastUpdatedMS: 1200,
+ LastColumnId: 1,
+ SchemaList: []*iceberg.Schema{iceberg.NewSchema(0)},
+ CurrentSchemaID: 0,
+ Specs: []iceberg.PartitionSpec{*iceberg.UnpartitionedSpec},
+ DefaultSpecID: 0,
+ LastPartitionID: &lastPartitionID,
+ Props: iceberg.Properties{},
+ SnapshotList: []Snapshot{
+ {
+ SnapshotID: s1,
+ TimestampMs: 1100,
+ ManifestList: "/snap-101.avro",
+ Summary: &Summary{
+ Operation: OpAppend,
+ Properties: iceberg.Properties{"added-records": "10", "total-records": "10"},
+ },
+ },
+ // s2 intentionally carries no summary and no manifest-list path.
+ {SnapshotID: s2, ParentSnapshotID: int64Ptr(s1), TimestampMs: 1200},
+ },
+ CurrentSnapshotID: ¤t,
+ SortOrderList: []SortOrder{UnsortedSortOrder},
+ DefaultSortOrderID: 0,
+ SnapshotRefs: map[string]SnapshotRef{MainBranch: {SnapshotID: current, SnapshotRefType: BranchRef}},
+ }}
+
+ return New(Identifier{"snapshots"}, meta, "", nil, nil)
+}
+
+func TestInspectSnapshotsSchema(t *testing.T) {
+ sc := SnapshotsSchema()
+
+ require.Equal(t,
+ []string{"committed_at", "snapshot_id", "parent_id", "operation", "manifest_list", "summary"},
+ testFieldNames(sc))
+
+ fields := sc.Fields()
+ for i := range fields {
+ require.Equal(t, i+1, fields[i].ID)
+ }
+
+ require.True(t, fields[0].Required, "committed_at is required")
+ require.True(t, fields[1].Required, "snapshot_id is required")
+ require.False(t, fields[2].Required, "parent_id is optional")
+ require.False(t, fields[3].Required, "operation is optional")
+ require.False(t, fields[4].Required, "manifest_list is optional")
+ require.False(t, fields[5].Required, "summary is optional")
+
+ m, ok := fields[5].Type.(*iceberg.MapType)
+ require.True(t, ok, "summary must be a map")
+ require.Equal(t, 7, m.KeyID)
+ require.Equal(t, 8, m.ValueID)
+ require.Equal(t, iceberg.PrimitiveTypes.String, m.KeyType)
+ require.Equal(t, iceberg.PrimitiveTypes.String, m.ValueType)
+}
+
+func TestInspectSnapshots(t *testing.T) {
+ tbl := snapshotsTestTable()
+
+ rr, err := tbl.Inspect().Snapshots(context.Background())
+ require.NoError(t, err)
+ defer rr.Release()
+
+ rec := collectRecord(t, rr)
+ defer rec.Release()
+
+ require.EqualValues(t, 2, rec.NumRows())
+ require.EqualValues(t, 6, rec.NumCols())
+
+ // committed_at must be timestamptz: microsecond precision, UTC.
+ tsType, ok := rec.Schema().Field(0).Type.(*arrow.TimestampType)
+ require.True(t, ok, "committed_at must be an Arrow timestamp")
+ require.Equal(t, arrow.Microsecond, tsType.Unit)
+ require.Equal(t, "UTC", tsType.TimeZone)
+
+ committedAt := rec.Column(0).(*array.Timestamp)
+ snapshotID := rec.Column(1).(*array.Int64)
+ parentID := rec.Column(2).(*array.Int64)
+ operation := rec.Column(3).(*array.String)
+ manifestList := rec.Column(4).(*array.String)
+ summary := rec.Column(5).(*array.Map)
+
+ require.EqualValues(t, 1100*1000, committedAt.Value(0))
+ require.EqualValues(t, 1200*1000, committedAt.Value(1))
+
+ require.EqualValues(t, 101, snapshotID.Value(0))
+ require.EqualValues(t, 102, snapshotID.Value(1))
+
+ require.True(t, parentID.IsNull(0), "root snapshot has no parent")
+ require.False(t, parentID.IsNull(1))
+ require.EqualValues(t, 101, parentID.Value(1))
+
+ // s1 has a manifest-list path; s2 has none and must render null, not "".
+ require.False(t, manifestList.IsNull(0))
+ require.Equal(t, "/snap-101.avro", manifestList.Value(0))
+ require.True(t, manifestList.IsNull(1), "snapshot without a manifest list has null manifest_list")
+
+ // s1 has a summary; s2 does not.
+ require.False(t, operation.IsNull(0))
+ require.Equal(t, "append", operation.Value(0))
+ require.True(t, operation.IsNull(1), "snapshot without a summary has null operation")
+
+ // The summary map mirrors the stored summary: operation folded in with the
+ // extra properties.
+ require.False(t, summary.IsNull(0))
+ require.Equal(t,
+ map[string]string{"operation": "append", "added-records": "10", "total-records": "10"},
+ mapRow(t, summary, 0))
+ require.True(t, summary.IsNull(1), "snapshot without a summary has null summary")
+}
+
+// TestInspectSnapshotsEmpty covers a table with no snapshots.
+func TestInspectSnapshotsEmpty(t *testing.T) {
+ lastPartitionID := 999
+ meta := &metadataV2{commonMetadata: commonMetadata{
+ FormatVersion: 2,
+ UUID: uuid.New(),
+ Loc: "s3://test/empty",
+ LastUpdatedMS: 1000,
+ LastColumnId: 1,
+ SchemaList: []*iceberg.Schema{iceberg.NewSchema(0)},
+ CurrentSchemaID: 0,
+ Specs: []iceberg.PartitionSpec{*iceberg.UnpartitionedSpec},
+ DefaultSpecID: 0,
+ LastPartitionID: &lastPartitionID,
+ Props: iceberg.Properties{},
+ SortOrderList: []SortOrder{UnsortedSortOrder},
+ DefaultSortOrderID: 0,
+ SnapshotRefs: map[string]SnapshotRef{},
+ }}
+ tbl := New(Identifier{"empty"}, meta, "", nil, nil)
+
+ rr, err := tbl.Inspect().Snapshots(context.Background())
+ require.NoError(t, err)
+ defer rr.Release()
+
+ rec := collectRecord(t, rr)
+ defer rec.Release()
+
+ require.EqualValues(t, 0, rec.NumRows())
+ require.EqualValues(t, 6, rec.NumCols())
+}
+
+// TestInspectAllocatorOption verifies WithInspectAllocator routes allocations
+// through the supplied allocator, and that all buffers are released.
+func TestInspectAllocatorOption(t *testing.T) {
+ checked := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ // AssertSize runs after the releases below, on every exit path, so an early
+ // assertion failure cannot silently skip the leak check.
+ t.Cleanup(func() { checked.AssertSize(t, 0) })
+ tbl := snapshotsTestTable()
+
+ rr, err := tbl.Inspect(WithInspectAllocator(checked)).Snapshots(context.Background())
+ require.NoError(t, err)
+ defer rr.Release()
+
+ rec := collectRecord(t, rr)
+ defer rec.Release()
+
+ require.EqualValues(t, 2, rec.NumRows())
+}
+
+// mapRow reads one row of a string->string Arrow map column into a Go map.
+func mapRow(t *testing.T, m *array.Map, row int) map[string]string {
+ t.Helper()
+ keys := m.Keys().(*array.String)
+ values := m.Items().(*array.String)
+ start, end := m.ValueOffsets(row)
+ out := make(map[string]string, end-start)
+ for j := start; j < end; j++ {
+ out[keys.Value(int(j))] = values.Value(int(j))
+ }
+
+ return out
+}
+
// testFieldNames returns the top-level field names of an Iceberg schema in order.
func testFieldNames(sc *iceberg.Schema) []string {
fields := sc.Fields()