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 45d5a65c1 feat(table): register Puffin statistics in transactions
(#2437)
45d5a65c1 is described below
commit 45d5a65c16b8c58bcccd3aa0d699f218a4821cfe
Author: Minh Vu <[email protected]>
AuthorDate: Mon Nov 9 19:01:56 2026 +0100
feat(table): register Puffin statistics in transactions (#2437)
* feat(table): register Puffin statistics files in transactions
* test(table): cover Puffin statistics registration and validation
* fix(table): align Puffin statistics registration with Java
* test(table): assert unchanged metadata for no-op statistics transaction
* test(table): preserve registered statistics on unmatched removal
* fix(table): address Puffin statistics review feedback
* test(puffin): cover writer size accessors across codecs
---
puffin/puffin_writer.go | 37 +++-
puffin/puffin_writer_state_test.go | 99 +++++++++
table/metadata.go | 10 +-
table/partition_statistics_write.go | 4 -
table/partition_statistics_write_test.go | 23 +++
table/statistics_write.go | 56 ++++++
table/statistics_write_test.go | 331 +++++++++++++++++++++++++++++++
website/src/feature-status.md | 2 +-
8 files changed, 548 insertions(+), 14 deletions(-)
diff --git a/puffin/puffin_writer.go b/puffin/puffin_writer.go
index 4da2849da..a639b8465 100644
--- a/puffin/puffin_writer.go
+++ b/puffin/puffin_writer.go
@@ -51,13 +51,14 @@ import (
// Writer cannot be reused after a physical write fails. Once poisoned, all
// subsequent operations return the original write error without changing
state.
type Writer struct {
- w io.Writer
- offset int64
- blobs []BlobMetadata
- props map[string]string
- done bool
- failed error
- createdBy string
+ w io.Writer
+ offset int64
+ footerSize int64
+ blobs []BlobMetadata
+ props map[string]string
+ done bool
+ failed error
+ createdBy string
}
// BlobMetadataInput contains fields the caller provides when adding a blob.
@@ -94,6 +95,26 @@ func NewWriter(w io.Writer) (*Writer, error) {
}, nil
}
+// FileSizeInBytes returns the total completed file size, including the header,
+// blobs and footer. It returns zero until Finish succeeds.
+func (w *Writer) FileSizeInBytes() int64 {
+ if !w.done {
+ return 0
+ }
+
+ return w.offset
+}
+
+// FileFooterSizeInBytes returns the complete footer size, including the
+// leading magic, payload and trailer. It returns zero until Finish succeeds.
+func (w *Writer) FileFooterSizeInBytes() int64 {
+ if !w.done {
+ return 0
+ }
+
+ return w.footerSize
+}
+
// checkState keeps the writer's finalized and failed-state checks consistent
// across operations. State checks must run before validating any input.
func (w *Writer) checkState(action string) error {
@@ -291,6 +312,8 @@ func (w *Writer) Finish() error {
return fmt.Errorf("puffin: write footer trailer: %w", err)
}
+ w.footerSize = int64(MagicSize + len(payload) + footerTrailerSize)
+ w.offset += w.footerSize
w.done = true
return nil
diff --git a/puffin/puffin_writer_state_test.go
b/puffin/puffin_writer_state_test.go
index 8e66695d5..a6264841e 100644
--- a/puffin/puffin_writer_state_test.go
+++ b/puffin/puffin_writer_state_test.go
@@ -18,6 +18,8 @@
package puffin
import (
+ "bytes"
+ "encoding/binary"
"errors"
"io"
"testing"
@@ -82,3 +84,100 @@ func TestWriterStateGuardErrors(t *testing.T) {
})
}
}
+
+func TestWriterReportsCompletedSizes(t *testing.T) {
+ var output bytes.Buffer
+ writer, err := NewWriter(&output)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if writer.FileSizeInBytes() != 0 || writer.FileFooterSizeInBytes() != 0
{
+ t.Fatal("file sizes must be unavailable before Finish")
+ }
+
+ _, err = writer.AddBlob(BlobMetadataInput{
+ Type: BlobTypeDataSketchesTheta,
+ Fields: []int32{1},
+ }, []byte("blob"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if writer.FileSizeInBytes() != 0 || writer.FileFooterSizeInBytes() != 0
{
+ t.Fatal("file sizes must be unavailable until Finish")
+ }
+ if err := writer.Finish(); err != nil {
+ t.Fatal(err)
+ }
+
+ data := output.Bytes()
+ trailer := data[len(data)-footerTrailerSize:]
+ payloadSize := int64(binary.LittleEndian.Uint32(trailer[:4]))
+ wantFooterSize := int64(MagicSize+footerTrailerSize) + payloadSize
+ if got := writer.FileFooterSizeInBytes(); got != wantFooterSize {
+ t.Fatalf("footer size = %d, want %d", got, wantFooterSize)
+ }
+ if got := writer.FileSizeInBytes(); got != int64(len(data)) {
+ t.Fatalf("file size = %d, want %d", got, len(data))
+ }
+}
+
+func TestWriterSizeAccessorsRoundTrip(t *testing.T) {
+ cases := []struct {
+ name string
+ codec string
+ blob bool
+ }{
+ {name: "no blobs"},
+ {name: "lz4", codec: CompressionCodecLZ4, blob: true},
+ {name: "zstd", codec: CompressionCodecZstd, blob: true},
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ var output bytes.Buffer
+ writer, err := NewWriter(&output)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ input := bytes.Repeat([]byte("iceberg-puffin"), 64)
+ if tc.blob {
+ _, err := writer.AddBlob(BlobMetadataInput{
+ Type:
BlobTypeDataSketchesTheta,
+ Fields: []int32{1},
+ CompressionCodec: tc.codec,
+ }, input)
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+
+ if err := writer.Finish(); err != nil {
+ t.Fatal(err)
+ }
+ reader, err :=
NewReader(bytes.NewReader(output.Bytes()))
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if got, want := writer.FileSizeInBytes(),
int64(output.Len()); got != want {
+ t.Fatalf("file size = %d, want %d", got, want)
+ }
+ if got, want := writer.FileFooterSizeInBytes(),
int64(output.Len())-reader.footerStart; got != want {
+ t.Fatalf("footer size = %d, want %d", got, want)
+ }
+
+ if tc.blob {
+ result, err := reader.ReadBlob(0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.Equal(result.Data, input) {
+ t.Fatal("Puffin round trip changed blob
contents")
+ }
+ } else if len(reader.Blobs()) != 0 {
+ t.Fatal("empty Puffin file unexpectedly
contains blobs")
+ }
+ })
+ }
+}
diff --git a/table/metadata.go b/table/metadata.go
index 01e203b18..d7ae9dd80 100644
--- a/table/metadata.go
+++ b/table/metadata.go
@@ -2021,10 +2021,13 @@ func (b *MetadataBuilder) SetStatistics(stats
StatisticsFile) error {
// RemoveStatistics removes the statistics file associated with the given
snapshot ID.
// It is not an error if no such file exists.
func (b *MetadataBuilder) RemoveStatistics(snapshotID int64) error {
+ previousLen := len(b.statisticsList)
b.statisticsList = slices.DeleteFunc(b.statisticsList, func(s
StatisticsFile) bool {
return s.SnapshotID == snapshotID
})
- b.updates = append(b.updates, NewRemoveStatisticsUpdate(snapshotID))
+ if len(b.statisticsList) != previousLen {
+ b.updates = append(b.updates,
NewRemoveStatisticsUpdate(snapshotID))
+ }
return nil
}
@@ -2055,10 +2058,13 @@ func (b *MetadataBuilder) SetPartitionStatistics(stats
PartitionStatisticsFile)
// RemovePartitionStatistics removes the partition statistics file associated
with the given
// snapshot ID. It is not an error if no such file exists.
func (b *MetadataBuilder) RemovePartitionStatistics(snapshotID int64) error {
+ previousLen := len(b.partitionStatsList)
b.partitionStatsList = slices.DeleteFunc(b.partitionStatsList, func(s
PartitionStatisticsFile) bool {
return s.SnapshotID == snapshotID
})
- b.updates = append(b.updates,
NewRemovePartitionStatisticsUpdate(snapshotID))
+ if len(b.partitionStatsList) != previousLen {
+ b.updates = append(b.updates,
NewRemovePartitionStatisticsUpdate(snapshotID))
+ }
return nil
}
diff --git a/table/partition_statistics_write.go
b/table/partition_statistics_write.go
index ccb2a9a2f..d740a91ad 100644
--- a/table/partition_statistics_write.go
+++ b/table/partition_statistics_write.go
@@ -75,10 +75,6 @@ func (t *Transaction) SetPartitionStatistics(stats
PartitionStatisticsFile) erro
// RemovePartitionStatistics removes the partition statistics file reference
// associated with snapshotID. It does not delete the physical statistics file.
func (t *Transaction) RemovePartitionStatistics(snapshotID int64) error {
- if _, err := t.txnMeta(); err != nil {
- return err
- }
-
return
t.apply([]Update{NewRemovePartitionStatisticsUpdate(snapshotID)}, nil)
}
diff --git a/table/partition_statistics_write_test.go
b/table/partition_statistics_write_test.go
index 35fe8bdba..52b2d111d 100644
--- a/table/partition_statistics_write_test.go
+++ b/table/partition_statistics_write_test.go
@@ -361,6 +361,29 @@ func TestTransactionPartitionStatisticsRegistration(t
*testing.T) {
require.Error(t, nilTx.SetPartitionStatistics(stats))
}
+func TestTransactionRemovePartitionStatisticsNoOp(t *testing.T) {
+ tbl := inspectTestTable(Identifier{"db", "stats"},
map[string]SnapshotRef{
+ MainBranch: {SnapshotID: 103, Type: BranchRef},
+ })
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.RemovePartitionStatistics(103))
+ require.NoError(t, tx.RemovePartitionStatistics(999))
+ require.Empty(t, tx.meta.updates)
+
+ stats := PartitionStatisticsFile{
+ SnapshotID: 103, StatisticsPath:
"s3://warehouse/db/stats/metadata/stats.parquet", FileSizeInBytes: 512,
+ }
+ require.NoError(t, tx.SetPartitionStatistics(stats))
+ updateCount := len(tx.meta.updates)
+ require.NoError(t, tx.RemovePartitionStatistics(999))
+ require.Len(t, tx.meta.updates, updateCount)
+
+ require.NoError(t, tx.RemovePartitionStatistics(stats.SnapshotID))
+ updateCount = len(tx.meta.updates)
+ require.NoError(t, tx.RemovePartitionStatistics(stats.SnapshotID))
+ require.Len(t, tx.meta.updates, updateCount)
+}
+
func TestComputePartitionStatisticsRejectsNonV1AndMissingSnapshot(t
*testing.T) {
tbl := inspectTestTable(Identifier{"db", "stats"},
map[string]SnapshotRef{
MainBranch: {SnapshotID: 103, Type: BranchRef},
diff --git a/table/statistics_write.go b/table/statistics_write.go
new file mode 100644
index 000000000..39ca50032
--- /dev/null
+++ b/table/statistics_write.go
@@ -0,0 +1,56 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "fmt"
+
+ "github.com/apache/iceberg-go"
+)
+
+// SetStatistics registers a previously written Puffin statistics file with
+// the file's snapshot. A file for the same snapshot replaces the previous
+// reference. This only stages a metadata update; Commit publishes it, and
+// the caller is responsible for writing a valid Puffin file beforehand.
+func (t *Transaction) SetStatistics(stats StatisticsFile) error {
+ if stats.SnapshotID <= 0 {
+ return fmt.Errorf("%w: statistics snapshot ID must be
positive", iceberg.ErrInvalidArgument)
+ }
+ if stats.StatisticsPath == "" || stats.FileSizeInBytes <= 0 ||
+ stats.FileFooterSizeInBytes <= 0 || stats.FileFooterSizeInBytes
> stats.FileSizeInBytes {
+ return fmt.Errorf("%w: statistics path must be nonempty, and
file and footer sizes must be positive with footer size at most file size",
+ iceberg.ErrInvalidArgument)
+ }
+ for idx, blob := range stats.BlobMetadata {
+ if blob.Type == "" {
+ return fmt.Errorf("%w: statistics blob %d requires a
type",
+ iceberg.ErrInvalidArgument, idx)
+ }
+ }
+ // A transaction may be committed after the caller has reused its
buffers.
+ // Keep the staged update independent of the caller's maps and slices.
+ registered := cloneStatisticsFiles([]StatisticsFile{stats})[0]
+
+ return t.apply([]Update{NewSetStatisticsUpdate(registered)}, nil)
+}
+
+// RemoveStatistics removes the statistics file reference for snapshotID. An
+// absent reference is not an error. The physical Puffin file is not deleted.
+func (t *Transaction) RemoveStatistics(snapshotID int64) error {
+ return t.apply([]Update{NewRemoveStatisticsUpdate(snapshotID)}, nil)
+}
diff --git a/table/statistics_write_test.go b/table/statistics_write_test.go
new file mode 100644
index 000000000..f6fb2adc1
--- /dev/null
+++ b/table/statistics_write_test.go
@@ -0,0 +1,331 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "context"
+ "encoding/json"
+ "slices"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/puffin"
+ "github.com/stretchr/testify/require"
+)
+
+func testStatisticsRegistrationFile() StatisticsFile {
+ return StatisticsFile{
+ SnapshotID: 103,
+ StatisticsPath:
"s3://warehouse/db/stats/metadata/sketch.puffin",
+ FileSizeInBytes: 1024,
+ FileFooterSizeInBytes: 128,
+ BlobMetadata: []BlobMetadata{{
+ Type: puffin.BlobTypeDataSketchesTheta,
+ SnapshotID: 103,
+ SequenceNumber: 3,
+ Fields: []int32{1},
+ Properties: map[string]string{"creator": "go"},
+ }},
+ }
+}
+
+func TestTransactionStatisticsRegistration(t *testing.T) {
+ tbl := inspectTestTable(Identifier{"db", "stats"},
map[string]SnapshotRef{
+ MainBranch: {SnapshotID: 103, Type: BranchRef},
+ })
+ tx := tbl.NewTransaction()
+ stats := testStatisticsRegistrationFile()
+
+ require.NoError(t, tx.SetStatistics(stats))
+ meta, err := tx.meta.Build()
+ require.NoError(t, err)
+ require.Equal(t, []StatisticsFile{stats},
slices.Collect(meta.Statistics()))
+
+ // Removing another snapshot must not stage a metadata update.
+ stagedUpdateCount := len(tx.meta.updates)
+ require.NoError(t, tx.RemoveStatistics(999))
+ require.Len(t, tx.meta.updates, stagedUpdateCount)
+ meta, err = tx.meta.Build()
+ require.NoError(t, err)
+ require.Equal(t, []StatisticsFile{stats},
slices.Collect(meta.Statistics()))
+
+ replacement := stats
+ replacement.StatisticsPath =
"s3://warehouse/db/stats/metadata/replaced.puffin"
+ replacement.FileSizeInBytes = 2048
+ require.NoError(t, tx.SetStatistics(replacement))
+ meta, err = tx.meta.Build()
+ require.NoError(t, err)
+ require.Equal(t, []StatisticsFile{replacement},
slices.Collect(meta.Statistics()))
+
+ require.NoError(t, tx.RemoveStatistics(stats.SnapshotID))
+ meta, err = tx.meta.Build()
+ require.NoError(t, err)
+ require.Empty(t, slices.Collect(meta.Statistics()))
+
+ // Java's removeStatistics is idempotent if no reference exists.
+ updateCount := len(tx.meta.updates)
+ require.NoError(t, tx.RemoveStatistics(stats.SnapshotID))
+ require.NoError(t, tx.RemoveStatistics(999))
+ require.Len(t, tx.meta.updates, updateCount)
+}
+
+func TestTransactionStatisticsMetadataRoundTrip(t *testing.T) {
+ base, err := getTestTableMetadata("TableMetadataV2Valid.json")
+ require.NoError(t, err)
+ require.NotEmpty(t, base.Snapshots())
+ tbl := New(Identifier{"db", "stats"}, base, "", nil, nil)
+ tx := tbl.NewTransaction()
+ stats := testStatisticsRegistrationFile()
+ stats.SnapshotID = base.Snapshots()[0].SnapshotID
+ stats.BlobMetadata[0].SnapshotID = stats.SnapshotID
+ stats.BlobMetadata[0].SequenceNumber =
base.Snapshots()[0].SequenceNumber
+ require.NoError(t, tx.SetStatistics(stats))
+
+ meta, err := tx.meta.Build()
+ require.NoError(t, err)
+ encoded, err := json.Marshal(meta)
+ require.NoError(t, err)
+ parsed, err := ParseMetadataBytes(encoded)
+ require.NoError(t, err)
+ require.Equal(t, []StatisticsFile{stats},
slices.Collect(parsed.Statistics()))
+}
+
+func TestTransactionStatisticsCatalogCommit(t *testing.T) {
+ base, err := getTestTableMetadata("TableMetadataV2Valid.json")
+ require.NoError(t, err)
+ require.NotEmpty(t, base.Snapshots())
+
+ cat := &sequentialCatalog{metadata: base}
+ ident := Identifier{"db", "stats"}
+ fs := iceio.NewMemFS()
+ tbl := New(ident, base, "mem://stats/metadata/v1.metadata.json",
+ func(context.Context) (iceio.IO, error) { return fs, nil }, cat)
+ tx := tbl.NewTransaction()
+
+ stats := testStatisticsRegistrationFile()
+ stats.SnapshotID = base.Snapshots()[0].SnapshotID
+ stats.BlobMetadata[0].SnapshotID = stats.SnapshotID
+ stats.BlobMetadata[0].SequenceNumber =
base.Snapshots()[0].SequenceNumber
+ require.NoError(t, tx.SetStatistics(stats))
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ require.NotNil(t, committed)
+
+ reloaded, err := cat.LoadTable(t.Context(), ident)
+ require.NoError(t, err)
+ require.Equal(t, []StatisticsFile{stats},
slices.Collect(reloaded.Metadata().Statistics()))
+ require.Equal(t, base.Snapshots(), reloaded.Metadata().Snapshots())
+ require.Equal(t, base.CurrentSnapshot(),
reloaded.Metadata().CurrentSnapshot())
+
+ removeTx := reloaded.NewTransaction()
+ require.NoError(t, removeTx.RemoveStatistics(stats.SnapshotID))
+ _, err = removeTx.Commit(t.Context())
+ require.NoError(t, err)
+
+ reloaded, err = cat.LoadTable(t.Context(), ident)
+ require.NoError(t, err)
+ require.Empty(t, slices.Collect(reloaded.Metadata().Statistics()))
+ require.Equal(t, base.Snapshots(), reloaded.Metadata().Snapshots())
+ require.Equal(t, base.CurrentSnapshot(),
reloaded.Metadata().CurrentSnapshot())
+}
+
+func TestTransactionSetStatisticsClonesCallerData(t *testing.T) {
+ tbl := inspectTestTable(Identifier{"db", "stats"},
map[string]SnapshotRef{
+ MainBranch: {SnapshotID: 103, Type: BranchRef},
+ })
+ tx := tbl.NewTransaction()
+ stats := testStatisticsRegistrationFile()
+ keyMetadata := "first"
+ stats.KeyMetadata = &keyMetadata
+ require.NoError(t, tx.SetStatistics(stats))
+
+ keyMetadata = "changed"
+ stats.BlobMetadata[0].Fields[0] = 99
+ stats.BlobMetadata[0].Properties["creator"] = "changed"
+ meta, err := tx.meta.Build()
+ require.NoError(t, err)
+ registered := slices.Collect(meta.Statistics())
+ require.Len(t, registered, 1)
+ require.Equal(t, "first", *registered[0].KeyMetadata)
+ require.Equal(t, []int32{1}, registered[0].BlobMetadata[0].Fields)
+ require.Equal(t, "go",
registered[0].BlobMetadata[0].Properties["creator"])
+}
+
+func TestTransactionSetStatisticsRejectsInvalidFiles(t *testing.T) {
+ tbl := inspectTestTable(Identifier{"db", "stats"},
map[string]SnapshotRef{
+ MainBranch: {SnapshotID: 103, Type: BranchRef},
+ })
+ cases := []struct {
+ name string
+ change func(*StatisticsFile)
+ }{
+ {"missing snapshot ID", func(s *StatisticsFile) { s.SnapshotID
= 0 }},
+ {"negative snapshot ID", func(s *StatisticsFile) { s.SnapshotID
= -1 }},
+ {"missing path", func(s *StatisticsFile) { s.StatisticsPath =
"" }},
+ {"nonpositive file size", func(s *StatisticsFile) {
s.FileSizeInBytes = 0 }},
+ {"negative file size", func(s *StatisticsFile) {
s.FileSizeInBytes = -1 }},
+ {"nonpositive footer size", func(s *StatisticsFile) {
s.FileFooterSizeInBytes = 0 }},
+ {"negative footer size", func(s *StatisticsFile) {
s.FileFooterSizeInBytes = -1 }},
+ {"footer exceeds file", func(s *StatisticsFile) {
s.FileFooterSizeInBytes = 1025 }},
+ {"empty blob type", func(s *StatisticsFile) {
s.BlobMetadata[0].Type = "" }},
+ {"empty second blob type", func(s *StatisticsFile) {
+ s.BlobMetadata = append(s.BlobMetadata, BlobMetadata{})
+ }},
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ tx := tbl.NewTransaction()
+ stats := testStatisticsRegistrationFile()
+ tc.change(&stats)
+ err := tx.SetStatistics(stats)
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+ if stats.SnapshotID <= 0 {
+ require.ErrorContains(t, err, "snapshot ID")
+ }
+ require.Empty(t, tx.meta.updates)
+ })
+ }
+
+ var nilTx *Transaction
+ require.Error(t, nilTx.SetStatistics(testStatisticsRegistrationFile()))
+ require.Error(t, nilTx.RemoveStatistics(103))
+}
+
+func TestTransactionRemoveStatisticsNoOp(t *testing.T) {
+ tbl := inspectTestTable(Identifier{"db", "stats"},
map[string]SnapshotRef{
+ MainBranch: {SnapshotID: 103, Type: BranchRef},
+ })
+ tx := tbl.NewTransaction()
+
+ require.NoError(t, tx.RemoveStatistics(103))
+ require.NoError(t, tx.RemoveStatistics(999))
+ require.Empty(t, tx.meta.updates)
+
+ commit, err := tx.TableCommit()
+ require.NoError(t, err)
+ require.Empty(t, commit.Updates)
+
+ // A no-op must not try to write a new metadata file through a catalog.
+ committed, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ require.Same(t, tbl.metadata, committed.metadata)
+ require.ErrorContains(t, tx.RemoveStatistics(103), "already been
committed")
+}
+
+func TestTransactionSetStatisticsAllowsUnretainedSnapshot(t *testing.T) {
+ tbl := inspectTestTable(Identifier{"db", "stats"},
map[string]SnapshotRef{
+ MainBranch: {SnapshotID: 103, Type: BranchRef},
+ })
+ tx := tbl.NewTransaction()
+ stats := testStatisticsRegistrationFile()
+ stats.SnapshotID = 999
+
+ // Java's TableMetadata.Builder.setStatistics does not require the
+ // statistics snapshot to be present in the current metadata.
+ require.NoError(t, tx.SetStatistics(stats))
+ meta, err := tx.meta.Build()
+ require.NoError(t, err)
+ require.Equal(t, []StatisticsFile{stats},
slices.Collect(meta.Statistics()))
+}
+
+func TestTransactionSetStatisticsAcceptsNilCollections(t *testing.T) {
+ tbl := inspectTestTable(Identifier{"db", "stats"},
map[string]SnapshotRef{
+ MainBranch: {SnapshotID: 103, Type: BranchRef},
+ })
+ cases := []struct {
+ name string
+ change func(*StatisticsFile)
+ wantBlobs int
+ }{
+ {"nil blob metadata", func(s *StatisticsFile) { s.BlobMetadata
= nil }, 0},
+ {"nil blob fields", func(s *StatisticsFile) {
s.BlobMetadata[0].Fields = nil }, 1},
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ tx := tbl.NewTransaction()
+ stats := testStatisticsRegistrationFile()
+ tc.change(&stats)
+ require.NoError(t, tx.SetStatistics(stats))
+
+ meta, err := tx.meta.Build()
+ require.NoError(t, err)
+ encoded, err := json.Marshal(meta)
+ require.NoError(t, err)
+
+ var payload struct {
+ Statistics []struct {
+ Blobs json.RawMessage
`json:"blob-metadata"`
+ } `json:"statistics"`
+ }
+ require.NoError(t, json.Unmarshal(encoded, &payload))
+ require.Len(t, payload.Statistics, 1)
+ if tc.wantBlobs == 0 {
+ require.JSONEq(t, "[]",
string(payload.Statistics[0].Blobs))
+ } else {
+ var blobs []struct {
+ Fields json.RawMessage `json:"fields"`
+ }
+ require.NoError(t,
json.Unmarshal(payload.Statistics[0].Blobs, &blobs))
+ require.Len(t, blobs, 1)
+ require.JSONEq(t, "[]", string(blobs[0].Fields))
+ }
+
+ parsed, err := ParseMetadataBytes(encoded)
+ require.NoError(t, err)
+ registered := slices.Collect(parsed.Statistics())
+ require.Len(t, registered, 1)
+ require.Len(t, registered[0].BlobMetadata, tc.wantBlobs)
+ if tc.wantBlobs == 1 {
+ require.Empty(t,
registered[0].BlobMetadata[0].Fields)
+ }
+ })
+ }
+}
+
+func TestTransactionSetStatisticsOmitsUnsetBlobProperties(t *testing.T) {
+ tbl := inspectTestTable(Identifier{"db", "stats"},
map[string]SnapshotRef{
+ MainBranch: {SnapshotID: 103, Type: BranchRef},
+ })
+ tx := tbl.NewTransaction()
+ stats := testStatisticsRegistrationFile()
+ stats.BlobMetadata[0].Type = "vendor-stats-v1"
+ stats.BlobMetadata[0].Fields = []int32{}
+ stats.BlobMetadata[0].Properties = nil
+
+ require.NoError(t, tx.SetStatistics(stats))
+ meta, err := tx.meta.Build()
+ require.NoError(t, err)
+ encoded, err := json.Marshal(meta)
+ require.NoError(t, err)
+
+ var payload struct {
+ Statistics []struct {
+ Blobs []map[string]json.RawMessage
`json:"blob-metadata"`
+ } `json:"statistics"`
+ }
+ require.NoError(t, json.Unmarshal(encoded, &payload))
+ require.Len(t, payload.Statistics, 1)
+ require.Len(t, payload.Statistics[0].Blobs, 1)
+ require.NotContains(t, payload.Statistics[0].Blobs[0], "properties",
+ "optional properties must be omitted, not encoded as null")
+
+ parsed, err := ParseMetadataBytes(encoded)
+ require.NoError(t, err)
+ require.Equal(t, []StatisticsFile{stats},
slices.Collect(parsed.Statistics()))
+}
diff --git a/website/src/feature-status.md b/website/src/feature-status.md
index c4c732521..4d538fcd1 100644
--- a/website/src/feature-status.md
+++ b/website/src/feature-status.md
@@ -60,7 +60,7 @@ All V1 features are supported. V1 is the format-version
baseline.
| Geography: read, write, and schema | Supported (bounds/pruning intentionally
omitted for geodesic safety) |
| Geometry / Geography: nested-geo null counts | Supported (like other nested
fields, no bounds are written) |
| Multi-argument transforms | Infrastructure present; no concrete
implementations exercised yet |
-| Statistics files (read / round-trip) | Supported |
+| Statistics files (read / round-trip / register / remove) | Supported |
| NDV (Theta-sketch) statistics computation | Roadmap
([#1929](https://github.com/apache/iceberg-go/issues/1929)) |
## FileSystem support
