laskoviymishka commented on code in PR #1696:
URL: https://github.com/apache/iceberg-go/pull/1696#discussion_r3739096243
##########
table/inspect.go:
##########
@@ -247,6 +247,104 @@ func (i InspectTable) Snapshots(ctx context.Context)
(array.RecordReader, error)
return rr, nil
}
+// MetadataLogEntries returns one row for every metadata file in the table's
+// metadata log, including the current metadata file. Snapshot information is
+// resolved from the snapshot log at each metadata file's timestamp.
+//
+// Columns:
+// - timestamp (timestamptz, required): when the metadata file was written
+// - file (string, required): metadata file location
+// - latest_snapshot_id (long, optional): latest snapshot visible then
+// - latest_schema_id (int, optional): schema used by that snapshot
+// - latest_sequence_number (long, optional): sequence number of that
snapshot
+//
+// The returned reader holds a single record batch. The caller must Release it.
+func (i InspectTable) MetadataLogEntries(ctx context.Context)
(array.RecordReader, error) {
+ arrowSchema, err := SchemaToArrowSchema(MetadataLogEntriesSchema(),
nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect metadata log entries: build
arrow schema: %w", err)
+ }
+
+ entries := make([]MetadataLogEntry, 0)
+ for entry := range i.tbl.metadata.PreviousFiles() {
+ entries = append(entries, entry)
+ }
+ entries = append(entries, MetadataLogEntry{
+ MetadataFile: i.tbl.metadataLocation,
+ TimestampMs: i.tbl.metadata.LastUpdatedMillis(),
+ })
+
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
+ defer bldr.Release()
+
+ timestamp := bldr.Field(0).(*array.TimestampBuilder)
+ file := bldr.Field(1).(*array.StringBuilder)
+ latestSnapshotID := bldr.Field(2).(*array.Int64Builder)
+ latestSchemaID := bldr.Field(3).(*array.Int32Builder)
+ latestSequenceNumber := bldr.Field(4).(*array.Int64Builder)
+
+ for _, entry := range entries {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ timestamp.Append(arrow.Timestamp(entry.TimestampMs * 1000))
+ file.Append(entry.MetadataFile)
+
+ snapshotID, snapshot, found := latestSnapshotAt(i.tbl.metadata,
entry.TimestampMs)
+ if !found {
+ latestSnapshotID.AppendNull()
+ latestSchemaID.AppendNull()
+ latestSequenceNumber.AppendNull()
+
+ continue
+ }
+
+ latestSnapshotID.Append(snapshotID)
+ if snapshot == nil {
+ latestSchemaID.AppendNull()
+ latestSequenceNumber.AppendNull()
+
+ continue
+ }
+ if snapshot.SchemaID != nil {
+ latestSchemaID.Append(int32(*snapshot.SchemaID))
+ } else {
+ latestSchemaID.AppendNull()
+ }
+ latestSequenceNumber.Append(snapshot.SequenceNumber)
+ }
+
+ rr, err := singleBatchReader(arrowSchema, bldr)
+ if err != nil {
+ return nil, fmt.Errorf("inspect metadata log entries: %w", err)
+ }
+
+ return rr, nil
+}
+
+// latestSnapshotAt follows the metadata-table behavior used by Java's
+// MetadataLogEntriesTable: find the snapshot-log entry at or before the
+// metadata timestamp, then resolve its details independently. The snapshot
+// ID remains available when the snapshot itself has expired from metadata.
+func latestSnapshotAt(metadata Metadata, timestampMs int64) (int64, *Snapshot,
bool) {
+ var snapshotID int64
+ var latestTimestamp int64
+ found := false
+ for entry := range metadata.SnapshotLogs() {
+ if entry.TimestampMs <= timestampMs && (!found ||
entry.TimestampMs > latestTimestamp) {
Review Comment:
The behavior here is right and matches Java's
`SnapshotUtil.snapshotIdAsOfTime`, but I'd tighten the doc comment above so
nobody unwinds it later.
Two things it should call out. First, this resolves against the *current*
snapshot log, so if an older snapshot was trimmed by ExpireSnapshots the result
can shift to a later entry (or none). The doc's "the snapshot-log entry at or
before the metadata timestamp" reads like point-in-time semantics, which it
isn't under trimming — it's the correct, Java-compatible choice, it just needs
to say so.
Second, the `> latestTimestamp` tie-break keeps the *first* entry on
equal-ms timestamps. That matches Java, but PyIceberg's
`snapshot_as_of_timestamp` iterates in reverse and returns the *last*. One line
on the `>` noting the divergence is intentional would keep someone from
"correcting" it. wdyt?
##########
table/inspect.go:
##########
@@ -247,6 +247,104 @@ func (i InspectTable) Snapshots(ctx context.Context)
(array.RecordReader, error)
return rr, nil
}
+// MetadataLogEntries returns one row for every metadata file in the table's
+// metadata log, including the current metadata file. Snapshot information is
+// resolved from the snapshot log at each metadata file's timestamp.
+//
+// Columns:
+// - timestamp (timestamptz, required): when the metadata file was written
+// - file (string, required): metadata file location
+// - latest_snapshot_id (long, optional): latest snapshot visible then
+// - latest_schema_id (int, optional): schema used by that snapshot
+// - latest_sequence_number (long, optional): sequence number of that
snapshot
+//
+// The returned reader holds a single record batch. The caller must Release it.
+func (i InspectTable) MetadataLogEntries(ctx context.Context)
(array.RecordReader, error) {
+ arrowSchema, err := SchemaToArrowSchema(MetadataLogEntriesSchema(),
nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect metadata log entries: build
arrow schema: %w", err)
+ }
+
+ entries := make([]MetadataLogEntry, 0)
+ for entry := range i.tbl.metadata.PreviousFiles() {
+ entries = append(entries, entry)
+ }
+ entries = append(entries, MetadataLogEntry{
+ MetadataFile: i.tbl.metadataLocation,
Review Comment:
If a table is ever built with an empty `metadataLocation`, this synthetic
current entry writes `""` into `file`, which is a required (non-null) column —
non-null but semantically empty. TestInspectMetadataLogEntriesEmpty passes a
real path so it doesn't cover this. Java skips the synthetic entry when the
location isn't set; I'd either match that or document the assumption that
`metadataLocation` is always populated here.
##########
table/inspect.go:
##########
@@ -247,6 +247,104 @@ func (i InspectTable) Snapshots(ctx context.Context)
(array.RecordReader, error)
return rr, nil
}
+// MetadataLogEntries returns one row for every metadata file in the table's
+// metadata log, including the current metadata file. Snapshot information is
+// resolved from the snapshot log at each metadata file's timestamp.
+//
+// Columns:
+// - timestamp (timestamptz, required): when the metadata file was written
+// - file (string, required): metadata file location
+// - latest_snapshot_id (long, optional): latest snapshot visible then
+// - latest_schema_id (int, optional): schema used by that snapshot
+// - latest_sequence_number (long, optional): sequence number of that
snapshot
+//
+// The returned reader holds a single record batch. The caller must Release it.
+func (i InspectTable) MetadataLogEntries(ctx context.Context)
(array.RecordReader, error) {
+ arrowSchema, err := SchemaToArrowSchema(MetadataLogEntriesSchema(),
nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect metadata log entries: build
arrow schema: %w", err)
+ }
+
+ entries := make([]MetadataLogEntry, 0)
+ for entry := range i.tbl.metadata.PreviousFiles() {
+ entries = append(entries, entry)
+ }
+ entries = append(entries, MetadataLogEntry{
+ MetadataFile: i.tbl.metadataLocation,
+ TimestampMs: i.tbl.metadata.LastUpdatedMillis(),
+ })
+
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
+ defer bldr.Release()
+
+ timestamp := bldr.Field(0).(*array.TimestampBuilder)
+ file := bldr.Field(1).(*array.StringBuilder)
+ latestSnapshotID := bldr.Field(2).(*array.Int64Builder)
+ latestSchemaID := bldr.Field(3).(*array.Int32Builder)
+ latestSequenceNumber := bldr.Field(4).(*array.Int64Builder)
+
+ for _, entry := range entries {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ timestamp.Append(arrow.Timestamp(entry.TimestampMs * 1000))
+ file.Append(entry.MetadataFile)
+
+ snapshotID, snapshot, found := latestSnapshotAt(i.tbl.metadata,
entry.TimestampMs)
+ if !found {
+ latestSnapshotID.AppendNull()
+ latestSchemaID.AppendNull()
+ latestSequenceNumber.AppendNull()
+
+ continue
+ }
+
+ latestSnapshotID.Append(snapshotID)
+ if snapshot == nil {
+ latestSchemaID.AppendNull()
+ latestSequenceNumber.AppendNull()
+
+ continue
+ }
+ if snapshot.SchemaID != nil {
+ latestSchemaID.Append(int32(*snapshot.SchemaID))
Review Comment:
`SchemaID` is spec-bounded to int32 so this never actually overflows, but
the bare `int32(*snapshot.SchemaID)` is a narrowing cast that'll trip gosec
G115 if we ever enable it, and the sibling tables don't cast (snapshots.go uses
the value as a plain int). I'd add a short `//nolint:gosec` noting it's bounded
by spec, or a guard — either's fine, just so the intent is explicit. wdyt?
##########
table/inspect.go:
##########
@@ -247,6 +247,104 @@ func (i InspectTable) Snapshots(ctx context.Context)
(array.RecordReader, error)
return rr, nil
}
+// MetadataLogEntries returns one row for every metadata file in the table's
+// metadata log, including the current metadata file. Snapshot information is
+// resolved from the snapshot log at each metadata file's timestamp.
+//
+// Columns:
+// - timestamp (timestamptz, required): when the metadata file was written
+// - file (string, required): metadata file location
+// - latest_snapshot_id (long, optional): latest snapshot visible then
+// - latest_schema_id (int, optional): schema used by that snapshot
+// - latest_sequence_number (long, optional): sequence number of that
snapshot
+//
+// The returned reader holds a single record batch. The caller must Release it.
+func (i InspectTable) MetadataLogEntries(ctx context.Context)
(array.RecordReader, error) {
+ arrowSchema, err := SchemaToArrowSchema(MetadataLogEntriesSchema(),
nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect metadata log entries: build
arrow schema: %w", err)
+ }
+
+ entries := make([]MetadataLogEntry, 0)
+ for entry := range i.tbl.metadata.PreviousFiles() {
+ entries = append(entries, entry)
+ }
+ entries = append(entries, MetadataLogEntry{
+ MetadataFile: i.tbl.metadataLocation,
+ TimestampMs: i.tbl.metadata.LastUpdatedMillis(),
+ })
+
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
Review Comment:
This builds five column builders and can early-return on `ctx.Err()`
mid-build, so I'd like a checked-allocator test asserting everything's
released. TestInspectAllocatorOption only exercises Snapshots today. Could we
add a TestInspectMetadataLogEntriesAllocator mirroring it with
`memory.NewCheckedAllocator` and asserting zero bytes after release? wdyt?
##########
table/inspect.go:
##########
@@ -247,6 +247,104 @@ func (i InspectTable) Snapshots(ctx context.Context)
(array.RecordReader, error)
return rr, nil
}
+// MetadataLogEntries returns one row for every metadata file in the table's
+// metadata log, including the current metadata file. Snapshot information is
+// resolved from the snapshot log at each metadata file's timestamp.
+//
+// Columns:
+// - timestamp (timestamptz, required): when the metadata file was written
+// - file (string, required): metadata file location
+// - latest_snapshot_id (long, optional): latest snapshot visible then
+// - latest_schema_id (int, optional): schema used by that snapshot
+// - latest_sequence_number (long, optional): sequence number of that
snapshot
+//
+// The returned reader holds a single record batch. The caller must Release it.
+func (i InspectTable) MetadataLogEntries(ctx context.Context)
(array.RecordReader, error) {
+ arrowSchema, err := SchemaToArrowSchema(MetadataLogEntriesSchema(),
nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect metadata log entries: build
arrow schema: %w", err)
+ }
+
+ entries := make([]MetadataLogEntry, 0)
+ for entry := range i.tbl.metadata.PreviousFiles() {
+ entries = append(entries, entry)
+ }
+ entries = append(entries, MetadataLogEntry{
+ MetadataFile: i.tbl.metadataLocation,
+ TimestampMs: i.tbl.metadata.LastUpdatedMillis(),
+ })
+
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
+ defer bldr.Release()
+
+ timestamp := bldr.Field(0).(*array.TimestampBuilder)
+ file := bldr.Field(1).(*array.StringBuilder)
+ latestSnapshotID := bldr.Field(2).(*array.Int64Builder)
+ latestSchemaID := bldr.Field(3).(*array.Int32Builder)
+ latestSequenceNumber := bldr.Field(4).(*array.Int64Builder)
+
+ for _, entry := range entries {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ timestamp.Append(arrow.Timestamp(entry.TimestampMs * 1000))
+ file.Append(entry.MetadataFile)
+
+ snapshotID, snapshot, found := latestSnapshotAt(i.tbl.metadata,
entry.TimestampMs)
+ if !found {
+ latestSnapshotID.AppendNull()
+ latestSchemaID.AppendNull()
+ latestSequenceNumber.AppendNull()
+
+ continue
+ }
+
+ latestSnapshotID.Append(snapshotID)
+ if snapshot == nil {
+ latestSchemaID.AppendNull()
+ latestSequenceNumber.AppendNull()
+
+ continue
+ }
+ if snapshot.SchemaID != nil {
+ latestSchemaID.Append(int32(*snapshot.SchemaID))
+ } else {
+ latestSchemaID.AppendNull()
+ }
+ latestSequenceNumber.Append(snapshot.SequenceNumber)
Review Comment:
For a V1 snapshot `SequenceNumber` is the Go zero value 0, and it's appended
unconditionally to this optional column, so there's no way to tell "V1, no
sequence numbers" from "V2, sequence 0". Java and PyIceberg both emit 0 here
too, so this matches every client and I wouldn't block on it. If we wanted to
be stricter than Java we could emit null for V1 (`Version() < 2`), but that's
genuinely optional.
##########
table/inspect_internal_test.go:
##########
@@ -404,6 +404,216 @@ func TestInspectSnapshotsEmpty(t *testing.T) {
require.EqualValues(t, 6, rec.NumCols())
}
+func TestInspectMetadataLogEntriesSchema(t *testing.T) {
+ sc := MetadataLogEntriesSchema()
+
+ require.Equal(t,
+ []string{"timestamp", "file", "latest_snapshot_id",
"latest_schema_id", "latest_sequence_number"},
+ testFieldNames(sc))
+
+ fields := sc.Fields()
+ for i := range fields {
+ require.Equal(t, i+1, fields[i].ID)
+ }
+
+ require.True(t, fields[0].Required, "timestamp is required")
+ require.True(t, fields[1].Required, "file is required")
+ require.False(t, fields[2].Required, "latest_snapshot_id is optional")
+ require.False(t, fields[3].Required, "latest_schema_id is optional")
+ require.False(t, fields[4].Required, "latest_sequence_number is
optional")
+
+ require.Equal(t, iceberg.PrimitiveTypes.TimestampTz, fields[0].Type)
+ require.Equal(t, iceberg.PrimitiveTypes.String, fields[1].Type)
+ require.Equal(t, iceberg.PrimitiveTypes.Int64, fields[2].Type)
+ require.Equal(t, iceberg.PrimitiveTypes.Int32, fields[3].Type)
+ require.Equal(t, iceberg.PrimitiveTypes.Int64, fields[4].Type)
+}
+
+func TestInspectMetadataLogEntries(t *testing.T) {
+ const (
+ s1 = int64(101)
+ s2 = int64(102)
+ expired = int64(999)
+ )
+ schema1 := 1
+ schema2 := 2
+ current := s2
+ lastPartitionID := 999
+ meta := &metadataV2{commonMetadata: commonMetadata{
+ FormatVersion: 2,
+ UUID: uuid.New(),
+ Loc: "s3://test/metadata-log-entries",
+ LastUpdatedMS: 4000,
+ LastColumnId: 1,
+ SchemaList: []*iceberg.Schema{
+ iceberg.NewSchema(0),
+ iceberg.NewSchema(schema1),
+ iceberg.NewSchema(schema2),
+ },
+ CurrentSchemaID: schema2,
+ Specs:
[]iceberg.PartitionSpec{*iceberg.UnpartitionedSpec},
+ DefaultSpecID: 0,
+ LastPartitionID: &lastPartitionID,
+ Props: iceberg.Properties{},
+ MetadataLog: []MetadataLogEntry{
+ {MetadataFile: "/metadata/v1.json", TimestampMs: 1000},
+ {MetadataFile: "/metadata/v2.json", TimestampMs: 2000},
+ },
+ SnapshotList: []Snapshot{
+ {SnapshotID: s1, SequenceNumber: 7, TimestampMs: 1500,
SchemaID: &schema1},
+ {SnapshotID: s2, SequenceNumber: 8, TimestampMs: 3000,
SchemaID: &schema2},
+ },
+ SnapshotLog: []SnapshotLogEntry{
+ {SnapshotID: s1, TimestampMs: 1500},
+ {SnapshotID: s2, TimestampMs: 3000},
+ {SnapshotID: expired, TimestampMs: 3500},
+ },
+ CurrentSnapshotID: ¤t,
+ SortOrderList: []SortOrder{UnsortedSortOrder},
+ DefaultSortOrderID: 0,
+ SnapshotRefs: map[string]SnapshotRef{MainBranch:
{SnapshotID: s2, SnapshotRefType: BranchRef}},
+ }}
+ tbl := New(Identifier{"metadata-log-entries"}, meta,
"/metadata/v3.json", nil, nil)
+
+ rr, err := tbl.Inspect().MetadataLogEntries(context.Background())
+ require.NoError(t, err)
+ defer rr.Release()
+
+ rec := collectRecord(t, rr)
+ defer rec.Release()
+
+ require.EqualValues(t, 3, rec.NumRows())
+ require.EqualValues(t, 5, rec.NumCols())
+
+ tsType, ok := rec.Schema().Field(0).Type.(*arrow.TimestampType)
+ require.True(t, ok, "timestamp must be an Arrow timestamp")
+ require.Equal(t, arrow.Microsecond, tsType.Unit)
+ require.Equal(t, "UTC", tsType.TimeZone)
+
+ timestamp := rec.Column(0).(*array.Timestamp)
+ file := rec.Column(1).(*array.String)
+ latestSnapshotID := rec.Column(2).(*array.Int64)
+ latestSchemaID := rec.Column(3).(*array.Int32)
+ latestSequenceNumber := rec.Column(4).(*array.Int64)
+
+ require.EqualValues(t, 1000*1000, timestamp.Value(0))
+ require.EqualValues(t, 2000*1000, timestamp.Value(1))
+ require.EqualValues(t, 4000*1000, timestamp.Value(2))
+ require.Equal(t, "/metadata/v1.json", file.Value(0))
+ require.Equal(t, "/metadata/v2.json", file.Value(1))
+ require.Equal(t, "/metadata/v3.json", file.Value(2))
+
+ // No snapshot had been committed at the first metadata-file timestamp.
+ require.True(t, latestSnapshotID.IsNull(0))
+ require.True(t, latestSchemaID.IsNull(0))
+ require.True(t, latestSequenceNumber.IsNull(0))
+
+ require.EqualValues(t, s1, latestSnapshotID.Value(1))
+ require.EqualValues(t, schema1, latestSchemaID.Value(1))
+ require.EqualValues(t, 7, latestSequenceNumber.Value(1))
+ // The snapshot log can retain an entry after the snapshot itself
expires.
+ // Keep its ID while leaving details that require the missing snapshot
null.
+ require.EqualValues(t, expired, latestSnapshotID.Value(2))
+ require.True(t, latestSchemaID.IsNull(2))
+ require.True(t, latestSequenceNumber.IsNull(2))
+}
+
+func TestLatestSnapshotAtScansAllSnapshotLogEntries(t *testing.T) {
+ const (
+ firstSnapshot = int64(101)
+ secondSnapshot = int64(102)
+ lateSnapshot = int64(103)
+ )
+ meta := &metadataV2{commonMetadata: commonMetadata{
+ SnapshotList: []Snapshot{
+ {SnapshotID: firstSnapshot},
+ {SnapshotID: secondSnapshot},
+ {SnapshotID: lateSnapshot},
+ },
+ SnapshotLog: []SnapshotLogEntry{
+ {SnapshotID: firstSnapshot, TimestampMs: 2000},
+ {SnapshotID: secondSnapshot, TimestampMs: 3000},
+ {SnapshotID: lateSnapshot, TimestampMs: 2500},
Review Comment:
This fixture is out of chronological order by 500ms (2000, 3000, 2500),
which passes today only because `validateChronologicalSnapshotLogs` allows a
1-minute skew tolerance. If that tolerance is ever tightened this becomes
invalid metadata and the test fails for an unrelated reason. A one-line comment
on whether this is meant to be genuinely out-of-order or just within-tolerance
clock skew would make the intent clear.
##########
table/inspect.go:
##########
@@ -247,6 +247,104 @@ func (i InspectTable) Snapshots(ctx context.Context)
(array.RecordReader, error)
return rr, nil
}
+// MetadataLogEntries returns one row for every metadata file in the table's
+// metadata log, including the current metadata file. Snapshot information is
+// resolved from the snapshot log at each metadata file's timestamp.
+//
+// Columns:
+// - timestamp (timestamptz, required): when the metadata file was written
+// - file (string, required): metadata file location
+// - latest_snapshot_id (long, optional): latest snapshot visible then
+// - latest_schema_id (int, optional): schema used by that snapshot
+// - latest_sequence_number (long, optional): sequence number of that
snapshot
+//
+// The returned reader holds a single record batch. The caller must Release it.
+func (i InspectTable) MetadataLogEntries(ctx context.Context)
(array.RecordReader, error) {
+ arrowSchema, err := SchemaToArrowSchema(MetadataLogEntriesSchema(),
nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect metadata log entries: build
arrow schema: %w", err)
+ }
+
+ entries := make([]MetadataLogEntry, 0)
Review Comment:
Small one: `PreviousFiles()` has a known length so this could be pre-sized,
or the loop collapsed with `slices.Collect` (already used in metadata.go). Not
load-bearing.
--
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]
