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 c5661c0ca feat(table): write and register v1 partition statistics
(#2396)
c5661c0ca is described below
commit c5661c0caab434d72861424b5239c62708baaf13
Author: Minh Vu <[email protected]>
AuthorDate: Wed Nov 4 19:03:48 2026 +0100
feat(table): write and register v1 partition statistics (#2396)
* feat(table): compute and write v1 partition statistics
* test(table): cover v1 partition statistics schema, sorting and metadata
updates
* fix(table): order binary and UUID partition tuples without reflection
copy panic
* test(table): use schema field slice in partition statistics assertions
* test(table): verify partition statistics from live manifests and deleted
entries
* fix(table): guard partition statistics for unsupported layouts
Signed-off-by: Minh Vu <[email protected]>
* fix(table): refuse duplicate files in partition statistics
Signed-off-by: Minh Vu <[email protected]>
* fix(table): sort decimal partition statistics by numeric value
* test(table): cover decimal partition statistics ordering
* feat(table): support embedded partition statistics manifests
Signed-off-by: Minh Vu <[email protected]>
* fix(table): infer omitted v1 embedded manifest spec IDs
* fix(table): accept legacy v1 fields without explicit IDs
* test(table): cover optional v1 partition spec IDs
* fix(table): match Java partition statistics ordering and API
* test(table): cover float and UUID partition statistics sorting
* style(table): separate early returns in partition statistics comparator
* refactor(table): share embedded v1 manifest spec parsing with audit
* fix(table): normalize decoded decimal partition statistics
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/embedded_manifest_spec.go | 77 ++++
table/embedded_manifest_spec_test.go | 89 +++++
table/partition_statistics_write.go | 652 +++++++++++++++++++++++++++++++
table/partition_statistics_write_test.go | 614 +++++++++++++++++++++++++++++
4 files changed, 1432 insertions(+)
diff --git a/table/embedded_manifest_spec.go b/table/embedded_manifest_spec.go
new file mode 100644
index 000000000..503b6ed6f
--- /dev/null
+++ b/table/embedded_manifest_spec.go
@@ -0,0 +1,77 @@
+// 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 (
+ "encoding/json"
+ "fmt"
+ "strconv"
+
+ "github.com/apache/iceberg-go"
+)
+
+// embeddedManifestSpecHeader is the partition spec from a v1 Avro manifest
+// header. Legacy writers may omit the spec ID or individual field IDs.
+type embeddedManifestSpecHeader struct {
+ id int32
+ hasID bool
+ fields []iceberg.PartitionField
+}
+
+func parseEmbeddedManifestSpecHeader(metadata map[string][]byte)
(embeddedManifestSpecHeader, error) {
+ var header embeddedManifestSpecHeader
+ if raw, present := metadata["partition-spec-id"]; present {
+ id, err := strconv.ParseInt(string(raw), 10, 32)
+ if err != nil || id < 0 {
+ return header, fmt.Errorf("invalid partition-spec-id
%q", raw)
+ }
+ header.id, header.hasID = int32(id), true
+ }
+ if err := json.Unmarshal(metadata["partition-spec"], &header.fields);
err != nil {
+ return header, fmt.Errorf("invalid v1 partition-spec: %w", err)
+ }
+ for i := range header.fields {
+ if header.fields[i].FieldID == 0 {
+ header.fields[i].FieldID = iceberg.PartitionDataIDStart
+ i
+ }
+ }
+
+ return header, nil
+}
+
+// resolve uses a header ID if present, including unknown IDs that the caller
+// may want to report. Without one, it only infers an ID from exactly one
+// matching historical spec. The caller decides whether an absent or ambiguous
+// match is inconclusive (audit) or an error (statistics writer).
+func (h embeddedManifestSpecHeader) resolve(specs
map[int32]iceberg.PartitionSpec) (id int32, known, ambiguous bool) {
+ if h.hasID {
+ return h.id, true, false
+ }
+ for candidateID, candidate := range specs {
+ actual := iceberg.NewPartitionSpecID(int(candidateID),
h.fields...)
+ if !actual.Equals(candidate) {
+ continue
+ }
+ if known {
+ return 0, false, true
+ }
+ id, known = candidateID, true
+ }
+
+ return id, known, false
+}
diff --git a/table/embedded_manifest_spec_test.go
b/table/embedded_manifest_spec_test.go
new file mode 100644
index 000000000..794bdb130
--- /dev/null
+++ b/table/embedded_manifest_spec_test.go
@@ -0,0 +1,89 @@
+// 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 (
+ "encoding/json"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/require"
+)
+
+func TestEmbeddedManifestSpecHeaderResolution(t *testing.T) {
+ field := iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "id", Transform:
iceberg.IdentityTransform{},
+ }
+ spec := iceberg.NewPartitionSpecID(7, field)
+ sameFields := iceberg.NewPartitionSpecID(8, field)
+ fields, err := json.Marshal([]iceberg.PartitionField{field})
+ require.NoError(t, err)
+
+ header, err :=
parseEmbeddedManifestSpecHeader(map[string][]byte{"partition-spec": fields})
+ require.NoError(t, err)
+ id, known, ambiguous :=
header.resolve(map[int32]iceberg.PartitionSpec{7: spec})
+ require.Equal(t, int32(7), id)
+ require.True(t, known)
+ require.False(t, ambiguous)
+
+ _, known, ambiguous = header.resolve(map[int32]iceberg.PartitionSpec{7:
spec, 8: sameFields})
+ require.False(t, known)
+ require.True(t, ambiguous)
+
+ _, known, ambiguous = header.resolve(map[int32]iceberg.PartitionSpec{})
+ require.False(t, known)
+ require.False(t, ambiguous)
+
+ // Old v1 headers can omit each field-id, which starts at 1000.
+ var rawFields []map[string]json.RawMessage
+ require.NoError(t, json.Unmarshal(fields, &rawFields))
+ delete(rawFields[0], "field-id")
+ fields, err = json.Marshal(rawFields)
+ require.NoError(t, err)
+ header, err =
parseEmbeddedManifestSpecHeader(map[string][]byte{"partition-spec": fields})
+ require.NoError(t, err)
+ require.Equal(t, 1000, header.fields[0].FieldID)
+ id, known, ambiguous =
header.resolve(map[int32]iceberg.PartitionSpec{7: spec})
+ require.Equal(t, int32(7), id)
+ require.True(t, known)
+ require.False(t, ambiguous)
+
+ // Explicit IDs are returned even when the table has no such spec, so
+ // validation can report the unknown ID rather than silently skipping
it.
+ header, err = parseEmbeddedManifestSpecHeader(map[string][]byte{
+ "partition-spec": fields, "partition-spec-id": []byte("42"),
+ })
+ require.NoError(t, err)
+ id, known, ambiguous =
header.resolve(map[int32]iceberg.PartitionSpec{7: spec})
+ require.Equal(t, int32(42), id)
+ require.True(t, known)
+ require.False(t, ambiguous)
+}
+
+func TestEmbeddedManifestSpecHeaderRejectsInvalidMetadata(t *testing.T) {
+ for _, value := range []string{"", "-1", "abc", "2147483648"} {
+ t.Run(value, func(t *testing.T) {
+ _, err :=
parseEmbeddedManifestSpecHeader(map[string][]byte{
+ "partition-spec": []byte("[]"),
"partition-spec-id": []byte(value),
+ })
+ require.ErrorContains(t, err, "invalid
partition-spec-id")
+ })
+ }
+ _, err :=
parseEmbeddedManifestSpecHeader(map[string][]byte{"partition-spec":
[]byte("{bad")})
+ require.ErrorContains(t, err, "invalid v1 partition-spec")
+}
diff --git a/table/partition_statistics_write.go
b/table/partition_statistics_write.go
new file mode 100644
index 000000000..0d6bfe854
--- /dev/null
+++ b/table/partition_statistics_write.go
@@ -0,0 +1,652 @@
+// 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 (
+ "bytes"
+ "context"
+ "encoding/binary"
+ "fmt"
+ "math"
+ "reflect"
+ "slices"
+
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/pqarrow"
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/google/uuid"
+ "github.com/twmb/avro/ocf"
+)
+
+// PartitionStatisticsSchemaV1 is the v1 partition statistics file schema.
+// It uses the unified partition type across all partition specs, not the
+// current spec's partition type. Optional delete metrics are omitted in v1.
+//
+// https://iceberg.apache.org/spec/#partition-statistics-file
+func PartitionStatisticsSchemaV1(partitionType *iceberg.StructType)
*iceberg.Schema {
+ return iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "partition", Type:
partitionType, Required: true},
+ iceberg.NestedField{ID: 2, Name: "spec_id", Type:
iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 3, Name: "data_record_count", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 4, Name: "data_file_count", Type:
iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 5, Name:
"total_data_file_size_in_bytes", Type: iceberg.PrimitiveTypes.Int64, Required:
true},
+ iceberg.NestedField{ID: 10, Name: "total_record_count", Type:
iceberg.PrimitiveTypes.Int64, Required: false},
+ )
+}
+
+// SetPartitionStatistics registers a previously written partition statistics
+// file with a snapshot. Files must follow the Iceberg partition statistics
+// file spec. This stages a metadata update; Commit publishes the reference.
+// Re-registering a snapshot replaces its existing statistics file reference.
+func (t *Transaction) SetPartitionStatistics(stats PartitionStatisticsFile)
error {
+ meta, err := t.txnMeta()
+ if err != nil {
+ return err
+ }
+ if stats.StatisticsPath == "" || stats.FileSizeInBytes <= 0 {
+ return fmt.Errorf("%w: partition statistics path must be
nonempty and file size must be positive",
+ iceberg.ErrInvalidArgument)
+ }
+ if _, err := meta.SnapshotByID(stats.SnapshotID); err != nil {
+ return fmt.Errorf("set partition statistics: %w", err)
+ }
+
+ return t.apply([]Update{NewSetPartitionStatisticsUpdate(stats)}, nil)
+}
+
+// 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)
+}
+
+// ComputePartitionStatistics computes and writes a Parquet partition
statistics
+// file for the transaction's current v1 branch head, and stages its metadata
+// reference. The caller must Commit to publish the reference. It does not
+// rewrite any data files or create a new table snapshot.
+//
+// The statistics are derived from the live entries in the head's manifests,
+// grouped by both partition spec ID and partition tuple. The file is sorted by
+// the unified partition tuple (ascending, null first), then by spec ID for
+// deterministic ties. Only tables with at least one partition field are
+// currently supported.
+// Partition fields whose source was dropped or whose result type changed are
+// not yet supported. Embedded manifests may omit the v1 partition spec ID
when the header
+// fields identify exactly one historical spec.
+//
+// Use a fresh transaction: a staged schema, spec, or snapshot change would
+// make the original head's partition type ambiguous. The branch head is pinned
+// to prevent an OCC retry from publishing statistics for a different snapshot.
+// The returned file may become orphaned if Commit fails; orphan-file cleanup
+// can remove it once it is no longer needed.
+func (t *Transaction) ComputePartitionStatistics(ctx context.Context)
(*PartitionStatisticsFile, error) {
+ meta, err := t.txnMeta()
+ if err != nil {
+ return nil, err
+ }
+ if t.tbl == nil || t.tbl.Metadata() == nil {
+ return nil, fmt.Errorf("%w: partition statistics require a
table", iceberg.ErrInvalidArgument)
+ }
+ if meta.formatVersion != 1 {
+ return nil, fmt.Errorf("%w: partition statistics computation
currently supports format v1 only",
+ iceberg.ErrNotImplemented)
+ }
+ if meta.HasChanges() {
+ return nil, fmt.Errorf("%w: compute partition statistics using
a fresh transaction",
+ iceberg.ErrInvalidArgument)
+ }
+
+ snapshot := t.planningSnapshot(meta)
+ if snapshot == nil {
+ return nil, fmt.Errorf("%w: cannot compute partition statistics
without a snapshot",
+ iceberg.ErrInvalidArgument)
+ }
+ if err := validatePartitionStatisticsSourceTypes(t.tbl.Metadata()); err
!= nil {
+ return nil, err
+ }
+ partitionType, err := inspectPartitionType(t.tbl.Metadata())
+ if err != nil {
+ return nil, fmt.Errorf("partition statistics: %w", err)
+ }
+ if partitionType == nil || len(partitionType.FieldList) == 0 {
+ return nil, fmt.Errorf("%w: partition statistics for
unpartitioned tables",
+ iceberg.ErrNotImplemented)
+ }
+ // This fences the snapshot whose manifests are about to be read. A
+ // concurrent append/overwrite on this branch must not silently rebase
the
+ // stats registration to that newer head.
+ if err := t.AssertRefSnapshotID(""); err != nil {
+ return nil, err
+ }
+ if err := t.apply(nil, []Requirement{
+ AssertCurrentSchemaID(t.tbl.Metadata().CurrentSchema().ID),
+ AssertDefaultSpecID(t.tbl.Metadata().DefaultPartitionSpec()),
+
AssertLastAssignedPartitionID(lastAssignedPartitionID(t.tbl.Metadata())),
+ }); err != nil {
+ return nil, err
+ }
+
+ fs, err := t.tbl.FS(ctx)
+ if err != nil {
+ return nil, err
+ }
+ wfs, err := requireWriteFileIO(fs)
+ if err != nil {
+ return nil, err
+ }
+ rows, err := collectPartitionStatistics(ctx, fs, *snapshot,
partitionType, t.tbl.Metadata().PartitionSpecs())
+ if err != nil {
+ return nil, fmt.Errorf("partition statistics: %w", err)
+ }
+ if err := sortPartitionStatistics(rows, partitionType); err != nil {
+ return nil, fmt.Errorf("partition statistics: %w", err)
+ }
+
+ locationProvider, err := t.tbl.LocationProvider()
+ if err != nil {
+ return nil, err
+ }
+ path := locationProvider.NewMetadataLocation(uuid.NewString() +
"-partition-stats.parquet")
+ size, err := writePartitionStatisticsParquet(ctx, wfs, path,
partitionType, rows)
+ if err != nil {
+ return nil, err
+ }
+
+ stats := PartitionStatisticsFile{
+ SnapshotID: snapshot.SnapshotID,
+ StatisticsPath: path,
+ FileSizeInBytes: size,
+ }
+ if err := t.SetPartitionStatistics(stats); err != nil {
+ _ = wfs.Remove(context.WithoutCancel(ctx), path)
+
+ return nil, err
+ }
+
+ return &stats, nil
+}
+
+// validatePartitionStatisticsSourceTypes rejects historical partition fields
+// whose source is missing or whose result type changed. The unified partition
+// type would otherwise omit historical values or misrepresent their types.
+func validatePartitionStatisticsSourceTypes(meta Metadata) error {
+ current := meta.CurrentSchema()
+ for _, spec := range meta.PartitionSpecs() {
+ for _, field := range spec.Fields() {
+ for _, id := range field.SourceIDs {
+ typ, ok := current.FindTypeByID(id)
+ if !ok {
+ return fmt.Errorf("%w: partition source
field %d was dropped", iceberg.ErrNotImplemented, id)
+ }
+ for _, schema := range meta.Schemas() {
+ if old, exists :=
schema.FindTypeByID(id); exists &&
+
!field.Transform.ResultType(typ).Equals(field.Transform.ResultType(old)) {
+ return fmt.Errorf("%w:
partition source field %d changed result type", iceberg.ErrNotImplemented, id)
+ }
+ }
+ }
+ }
+ }
+
+ return nil
+}
+
+// Embedded v1 paths lack manifest-list descriptors. Read and validate the
+// header spec before constructing a descriptor with the real spec ID.
+func partitionStatisticsEmbeddedManifest(ctx context.Context, fs iceio.IO,
manifest iceberg.ManifestFile, specs map[int32]iceberg.PartitionSpec)
(iceberg.ManifestFile, error) {
+ f, err := fs.Open(ctx, manifest.FilePath())
+ if err != nil {
+ return nil, err
+ }
+ defer f.Close()
+ ocfReader, err := ocf.NewReader(f)
+ if err != nil {
+ return nil, fmt.Errorf("read embedded manifest %q: %w",
manifest.FilePath(), err)
+ }
+ header, parseErr :=
parseEmbeddedManifestSpecHeader(ocfReader.Metadata())
+ if err := ocfReader.Close(); err != nil {
+ return nil, err
+ }
+ if parseErr != nil {
+ return nil, fmt.Errorf("%w: embedded manifest %q: %v",
+ iceberg.ErrInvalidArgument, manifest.FilePath(),
parseErr)
+ }
+
+ id, found, ambiguous := header.resolve(specs)
+ if !found {
+ if ambiguous {
+ return nil, fmt.Errorf("%w: embedded manifest %q has an
ambiguous partition spec without an ID",
+ iceberg.ErrNotImplemented, manifest.FilePath())
+ }
+
+ return nil, fmt.Errorf("%w: embedded manifest %q has no
matching partition spec",
+ iceberg.ErrInvalidArgument, manifest.FilePath())
+ }
+ expected, ok := specs[id]
+ if !ok {
+ return nil, fmt.Errorf("%w: embedded manifest %q references
unknown partition spec %d",
+ iceberg.ErrInvalidArgument, manifest.FilePath(), id)
+ }
+ actual := iceberg.NewPartitionSpecID(int(id), header.fields...)
+ if !actual.Equals(expected) {
+ return nil, fmt.Errorf("%w: embedded manifest %q partition spec
differs from table metadata",
+ iceberg.ErrInvalidArgument, manifest.FilePath())
+ }
+ // Entries will open and validate the descriptor through
ManifestFile.Entries.
+ // Constructing and closing another reader here only repeats that work.
+ return iceberg.NewManifestBuilder(1, manifest.FilePath(),
manifest.Length(), id, manifest.SnapshotID()).
+
AddedFiles(-1).ExistingFiles(-1).DeletedFiles(-1).AddedRows(-1).ExistingRows(-1).DeletedRows(-1).Build(),
nil
+}
+
+// collectPartitionStatistics reads only manifest metadata, not Parquet data.
+// Each partition spec gets a separate lookup tree: partitions that have the
+// same tuple under two spec IDs are distinct statistics rows.
+func collectPartitionStatistics(
+ ctx context.Context,
+ fs iceio.IO,
+ snapshot Snapshot,
+ partitionType *iceberg.StructType,
+ specs []iceberg.PartitionSpec,
+) ([]*inspectPartitionAggregate, error) {
+ specByID := make(map[int32]iceberg.PartitionSpec, len(specs))
+ for _, spec := range specs {
+ specByID[int32(spec.ID())] = spec
+ }
+ manifests, err := snapshot.Manifests(fs)
+ if err != nil {
+ return nil, err
+ }
+
+ trees := make(map[int32]*inspectPartitionAggregateTree)
+ seenPaths := make(map[string]string)
+ rows := make([]*inspectPartitionAggregate, 0)
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ if snapshot.ManifestList == "" && snapshot.ManifestLocations !=
nil {
+ manifest, err =
partitionStatisticsEmbeddedManifest(ctx, fs, manifest, specByID)
+ if err != nil {
+ return nil, err
+ }
+ }
+ specID := manifest.PartitionSpecID()
+ spec, ok := specByID[specID]
+ if !ok {
+ return nil, fmt.Errorf("%w: manifest %q references
unknown partition spec %d", iceberg.ErrInvalidArgument, manifest.FilePath(),
specID)
+ }
+ tree := trees[specID]
+ if tree == nil {
+ tree = newInspectPartitionAggregateTree()
+ trees[specID] = tree
+ }
+
+ for entry, err := range manifest.Entries(fs, true) {
+ if err != nil {
+ return nil, fmt.Errorf("read manifest %s: %w",
manifest.FilePath(), err)
+ }
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ if entry == nil || entry.DataFile() == nil {
+ return nil, fmt.Errorf("%w: manifest %q
contains a nil data file",
+ iceberg.ErrInvalidArgument,
manifest.FilePath())
+ }
+ file := entry.DataFile()
+ path := file.FilePath()
+ if path == "" {
+ return nil, fmt.Errorf("%w: manifest %q
contains an empty data file path",
+ iceberg.ErrInvalidArgument,
manifest.FilePath())
+ }
+ if previous, duplicate := seenPaths[path]; duplicate {
+ return nil, fmt.Errorf("%w: duplicate live data
file %q in manifests %q and %q",
+ iceberg.ErrInvalidArgument, path,
previous, manifest.FilePath())
+ }
+ seenPaths[path] = manifest.FilePath()
+ if file.ContentType() != iceberg.EntryContentData {
+ return nil, fmt.Errorf("%w: v1 manifest %s
contains a delete file",
+ iceberg.ErrInvalidArgument,
manifest.FilePath())
+ }
+ if file.Count() < 0 || file.FileSizeBytes() < 0 {
+ return nil, fmt.Errorf("%w: file %s has a
negative count or size",
+ iceberg.ErrInvalidArgument,
file.FilePath())
+ }
+ values := dataFilePartition(file)
+ if err := validatePartitionStatisticsTuple(spec,
file.FilePath(), values); err != nil {
+ return nil, err
+ }
+ partition := inspectCoercePartition(values,
partitionType)
+ for _, field := range partitionType.FieldList {
+ value := partition[field.ID]
+ // Manifest decoding uses DecimalLiteral, while
callers can
+ // supply Decimal. Normalize before building
aggregate keys
+ // so both representations count toward the
same partition.
+ if literal, ok :=
value.(iceberg.DecimalLiteral); ok {
+ value = iceberg.Decimal(literal)
+ partition[field.ID] = value
+ }
+ if decimal, ok := value.(iceberg.Decimal); ok {
+ if typ, ok :=
field.Type.(iceberg.DecimalType); ok && decimal.Scale != typ.Scale() {
+ return nil, fmt.Errorf("%w:
partition field %q has decimal scale %d, expected %d",
+
iceberg.ErrInvalidArgument, field.Name, decimal.Scale, typ.Scale())
+ }
+ }
+ if _, err :=
comparePartitionStatisticValues(value, value); err != nil {
+ return nil, fmt.Errorf("partition field
%q: %w", field.Name, err)
+ }
+ }
+ row := tree.lookupPartition(partition, partitionType)
+ if row == nil {
+ row = &inspectPartitionAggregate{
+ specID: specID,
+ partition:
cloneInspectPartition(partition),
+ }
+ rows = append(rows, row)
+ tree.insert(newPartitionRecord(partition,
partitionType), row)
+ }
+ if row.dataFileCount == math.MaxInt32 ||
+ row.dataRecordCount >
math.MaxInt64-file.Count() ||
+ row.dataFileSize >
math.MaxInt64-file.FileSizeBytes() {
+ return nil, fmt.Errorf("%w: partition
statistics count overflow",
+ iceberg.ErrInvalidArgument)
+ }
+ row.dataFileCount++
+ row.dataRecordCount += file.Count()
+ row.dataFileSize += file.FileSizeBytes()
+ }
+ }
+
+ return rows, nil
+}
+
+func validatePartitionStatisticsTuple(spec iceberg.PartitionSpec, path string,
values map[int]any) error {
+ for _, field := range spec.Fields() {
+ if _, exists := values[field.FieldID]; !exists {
+ return fmt.Errorf("%w: file %q is missing partition
field %d", iceberg.ErrInvalidArgument, path, field.FieldID)
+ }
+ }
+
+ return nil
+}
+
+// sortPartitionStatistics obeys the spec's NULL FIRST tuple ordering, not
+// inspectPartitionKey's text ordering (where "10" sorts before "2").
+// A spec ID only breaks equal tuple ties; it is not the primary sort key.
+func sortPartitionStatistics(rows []*inspectPartitionAggregate, partitionType
*iceberg.StructType) error {
+ // Collection already validates each partition value before using it as
+ // an aggregate key. Do not repeat that validation for every output row.
+ var sortErr error
+ slices.SortStableFunc(rows, func(a, b *inspectPartitionAggregate) int {
+ for _, field := range partitionType.FieldList {
+ comp, err :=
comparePartitionStatisticValues(a.partition[field.ID], b.partition[field.ID])
+ if err != nil {
+ sortErr = fmt.Errorf("partition field %q: %w",
field.Name, err)
+
+ return 0
+ }
+ if comp != 0 {
+ return comp
+ }
+ }
+ switch {
+ case a.specID < b.specID:
+ return -1
+ case a.specID > b.specID:
+ return 1
+ default:
+ return 0
+ }
+ })
+
+ return sortErr
+}
+
+// comparePartitionStatisticValues compares primitive partition values by
+// their underlying types, including named date/time integer types and binary
+// UUIDs. Unsupported types are rejected instead of emitting a file whose
+// partition ordering violates the Iceberg spec.
+func comparePartitionStatisticValues(a, b any) (int, error) {
+ if a == nil || b == nil {
+ if a == nil && b == nil {
+ return 0, nil
+ }
+ if a == nil {
+ return -1, nil
+ }
+
+ return 1, nil
+ }
+ // Decimal partitions sort by their numeric value, not by the underlying
+ // Decimal struct or the lexicographic encoding of the unscaled integer.
+ if left, ok := a.(iceberg.Decimal); ok {
+ right, valid := b.(iceberg.Decimal)
+ if !valid || left.Scale != right.Scale {
+ return 0, fmt.Errorf("%w: incompatible decimal
partition values %v and %v",
+ iceberg.ErrInvalidArgument, a, b)
+ }
+
+ return left.Val.Cmp(right.Val), nil
+ }
+ // Java's UUID.compareTo orders each 64-bit half as a signed integer.
+ // Binary and fixed partitions retain unsigned lexicographic ordering.
+ if left, ok := a.(uuid.UUID); ok {
+ right, valid := b.(uuid.UUID)
+ if !valid {
+ return 0, fmt.Errorf("%w: incompatible partition values
%T and %T",
+ iceberg.ErrInvalidArgument, a, b)
+ }
+ if cmp :=
compareOrderedStats(int64(binary.BigEndian.Uint64(left[:8])),
+ int64(binary.BigEndian.Uint64(right[:8]))); cmp != 0 {
+ return cmp, nil
+ }
+
+ return
compareOrderedStats(int64(binary.BigEndian.Uint64(left[8:])),
+ int64(binary.BigEndian.Uint64(right[8:]))), nil
+ }
+ if _, ok := b.(uuid.UUID); ok {
+ return 0, fmt.Errorf("%w: incompatible partition values %T and
%T",
+ iceberg.ErrInvalidArgument, a, b)
+ }
+ left, right := reflect.ValueOf(a), reflect.ValueOf(b)
+ if left.Kind() != right.Kind() {
+ return 0, fmt.Errorf("%w: incompatible partition values %T and
%T",
+ iceberg.ErrInvalidArgument, a, b)
+ }
+
+ switch left.Kind() {
+ case reflect.Bool:
+ if left.Bool() == right.Bool() {
+ return 0, nil
+ }
+ if !left.Bool() {
+ return -1, nil
+ }
+
+ return 1, nil
+ case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32,
reflect.Int64:
+ return compareOrderedStats(left.Int(), right.Int()), nil
+ case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32,
reflect.Uint64:
+ return compareOrderedStats(left.Uint(), right.Uint()), nil
+ case reflect.Float32, reflect.Float64:
+ // Match Java Float.compare/Double.compare: all NaNs compare
equal
+ // and sort after +Inf; -0 sorts before +0.
+ aFloat, bFloat := left.Float(), right.Float()
+ switch {
+ case math.IsNaN(aFloat):
+ if math.IsNaN(bFloat) {
+ return 0, nil
+ }
+
+ return 1, nil
+ case math.IsNaN(bFloat):
+ return -1, nil
+ case aFloat == 0 && bFloat == 0:
+ switch {
+ case math.Signbit(aFloat) == math.Signbit(bFloat):
+ return 0, nil
+ case math.Signbit(aFloat):
+ return -1, nil
+ default:
+ return 1, nil
+ }
+ default:
+ return compareOrderedStats(aFloat, bFloat), nil
+ }
+ case reflect.String:
+ return bytes.Compare([]byte(left.String()),
[]byte(right.String())), nil
+ case reflect.Slice, reflect.Array:
+ if left.Type().Elem().Kind() != reflect.Uint8 ||
right.Type().Elem().Kind() != reflect.Uint8 {
+ break
+ }
+ limit := min(left.Len(), right.Len())
+ for index := range limit {
+ aByte, bByte := left.Index(index).Uint(),
right.Index(index).Uint()
+ if aByte != bByte {
+ return compareOrderedStats(aByte, bByte), nil
+ }
+ }
+ switch {
+ case left.Len() < right.Len():
+ return -1, nil
+ case left.Len() > right.Len():
+ return 1, nil
+ default:
+ return 0, nil
+ }
+ }
+
+ return 0, fmt.Errorf("%w: cannot order partition values of type %T",
+ iceberg.ErrNotImplemented, a)
+}
+
+func compareOrderedStats[T interface{ ~int64 | ~uint64 | ~float64 }](a, b T)
int {
+ switch {
+ case a < b:
+ return -1
+ case a > b:
+ return 1
+ default:
+ return 0
+ }
+}
+
+type partitionStatisticsByteCounter struct {
+ writer iceio.FileWriter
+ size int64
+}
+
+func (w *partitionStatisticsByteCounter) Write(data []byte) (int, error) {
+ n, err := w.writer.Write(data)
+ w.size += int64(n)
+
+ return n, err
+}
+
+// writePartitionStatisticsParquet streams bounded Arrow batches to the table
+// metadata directory and removes incomplete files on failure.
+func writePartitionStatisticsParquet(
+ ctx context.Context,
+ fs iceio.WriteFileIO,
+ path string,
+ partitionType *iceberg.StructType,
+ rows []*inspectPartitionAggregate,
+) (_ int64, err error) {
+ sc, err :=
SchemaToArrowSchema(PartitionStatisticsSchemaV1(partitionType), nil, true,
false)
+ if err != nil {
+ return 0, err
+ }
+ fw, err := fs.Create(ctx, path)
+ if err != nil {
+ return 0, err
+ }
+ defer func() {
+ if err != nil {
+ _ = fs.Remove(context.WithoutCancel(ctx), path)
+ }
+ }()
+
+ counter := &partitionStatisticsByteCounter{writer: fw}
+ writer, err := pqarrow.NewFileWriter(sc, counter,
parquet.NewWriterProperties(), pqarrow.DefaultWriterProps())
+ if err != nil {
+ _ = fw.Close()
+
+ return 0, err
+ }
+ bldr := array.NewRecordBuilder(memory.DefaultAllocator, sc)
+ defer bldr.Release()
+
+ part, err :=
newInspectPartitionBuilder(bldr.Field(0).(*array.StructBuilder), partitionType)
+ if err != nil {
+ _ = writer.Close()
+ _ = fw.Close()
+
+ return 0, err
+ }
+ for start := 0; start < len(rows); start += inspectRecordBatchSize {
+ if err := ctx.Err(); err != nil {
+ _ = writer.Close()
+ _ = fw.Close()
+
+ return 0, err
+ }
+ end := min(start+inspectRecordBatchSize, len(rows))
+ for _, row := range rows[start:end] {
+ if err := part.append(row.partition); err != nil {
+ _ = writer.Close()
+ _ = fw.Close()
+
+ return 0, err
+ }
+ bldr.Field(1).(*array.Int32Builder).Append(row.specID)
+
bldr.Field(2).(*array.Int64Builder).Append(row.dataRecordCount)
+
bldr.Field(3).(*array.Int32Builder).Append(row.dataFileCount)
+
bldr.Field(4).(*array.Int64Builder).Append(row.dataFileSize)
+ // v1 has no delete files, so the exact live record
count equals
+ // the sum of data-file record counts.
+
bldr.Field(5).(*array.Int64Builder).Append(row.dataRecordCount)
+ }
+
+ rec := bldr.NewRecordBatch()
+ writeErr := writer.Write(rec)
+ rec.Release()
+ if writeErr != nil {
+ _ = writer.Close()
+ _ = fw.Close()
+
+ return 0, writeErr
+ }
+ }
+ if err := writer.Close(); err != nil {
+ _ = fw.Close()
+
+ return 0, err
+ }
+ if err := fw.Close(); err != nil {
+ return 0, err
+ }
+
+ return counter.size, nil
+}
diff --git a/table/partition_statistics_write_test.go
b/table/partition_statistics_write_test.go
new file mode 100644
index 000000000..35fe8bdba
--- /dev/null
+++ b/table/partition_statistics_write_test.go
@@ -0,0 +1,614 @@
+// 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 (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "io"
+ "math"
+ "slices"
+ "strconv"
+ "testing"
+ "time"
+
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/decimal128"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet/file"
+ "github.com/apache/arrow-go/v18/parquet/pqarrow"
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/google/uuid"
+ "github.com/stretchr/testify/require"
+ "github.com/twmb/avro/ocf"
+)
+
+func statsTestPartitionType() *iceberg.StructType {
+ return &iceberg.StructType{FieldList: []iceberg.NestedField{
+ {ID: 1000, Name: "day", Type: iceberg.PrimitiveTypes.Int32},
+ {ID: 1001, Name: "region", Type: iceberg.PrimitiveTypes.String},
+ }}
+}
+
+func TestPartitionStatisticsSchemaV1(t *testing.T) {
+ sc := PartitionStatisticsSchemaV1(statsTestPartitionType())
+ require.Equal(t, []string{
+ "partition", "spec_id", "data_record_count", "data_file_count",
+ "total_data_file_size_in_bytes", "total_record_count",
+ }, testFieldNames(sc))
+ fields := sc.Fields()
+ require.Equal(t, []int{1, 2, 3, 4, 5, 10}, []int{
+ fields[0].ID, fields[1].ID, fields[2].ID,
+ fields[3].ID, fields[4].ID, fields[5].ID,
+ })
+ for index := range 5 {
+ require.True(t, fields[index].Required)
+ }
+ require.False(t, fields[5].Required)
+ require.Equal(t, 1000,
fields[0].Type.(*iceberg.StructType).FieldList[0].ID)
+}
+
+func TestPartitionStatisticValuesNullFirstNumericAndBinary(t *testing.T) {
+ for _, tc := range []struct {
+ a, b any
+ want int
+ }{
+ {nil, int32(2), -1},
+ {int32(2), nil, 1},
+ {nil, nil, 0},
+ {int32(-2), int32(10), -1},
+ {int64(10), int64(-2), 1},
+ {uint64(1), uint64(2), -1},
+ {float64(-1.5), float64(0), -1},
+ {float64(math.Copysign(0, -1)), float64(0), -1},
+ {float64(0), float64(math.Copysign(0, -1)), 1},
+ {float32(math.Copysign(0, -1)), float32(0), -1},
+ {math.Inf(-1), -math.MaxFloat64, -1},
+ {math.NaN(), math.Inf(1), 1},
+ {math.Inf(1), math.NaN(), -1},
+ {math.NaN(), math.NaN(), 0},
+ {float32(math.NaN()), float32(math.NaN()), 0},
+ {float32(math.NaN()), float32(math.Inf(1)), 1},
+ {false, true, -1},
+ {"10", "2", -1},
+ {[]byte{0, 255}, []byte{1}, -1},
+ {[2]byte{0, 255}, [2]byte{1, 0}, -1},
+ {[]byte{1, 2}, []byte{1, 2}, 0},
+ {uuid.MustParse("80000000-0000-0000-0000-000000000000"),
uuid.MustParse("7fffffff-ffff-ffff-ffff-ffffffffffff"), -1},
+ {uuid.MustParse("00000000-0000-0000-8000-000000000000"),
uuid.MustParse("00000000-0000-0000-7fff-ffffffffffff"), -1},
+ {iceberg.Decimal{Val: decimal128.FromI64(-120), Scale: 2},
iceberg.Decimal{Val: decimal128.FromI64(30), Scale: 2}, -1},
+ {iceberg.Decimal{Val: decimal128.FromI64(30), Scale: 2},
iceberg.Decimal{Val: decimal128.FromI64(30), Scale: 2}, 0},
+ {iceberg.Decimal{Val: decimal128.FromI64(120), Scale: 2},
iceberg.Decimal{Val: decimal128.FromI64(30), Scale: 2}, 1},
+ } {
+ got, err := comparePartitionStatisticValues(tc.a, tc.b)
+ require.NoError(t, err, "compare %v to %v", tc.a, tc.b)
+ require.Equal(t, tc.want, got)
+ }
+ _, err := comparePartitionStatisticValues(struct{ X int }{}, struct{ X
int }{})
+ require.ErrorIs(t, err, iceberg.ErrNotImplemented)
+ _, err = comparePartitionStatisticValues(
+ iceberg.Decimal{Val: decimal128.FromI64(10), Scale: 1},
+ iceberg.Decimal{Val: decimal128.FromI64(10), Scale: 2})
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+}
+
+func TestSortPartitionStatisticsByTupleThenSpecID(t *testing.T) {
+ pt := statsTestPartitionType()
+ rows := []*inspectPartitionAggregate{
+ {partition: map[int]any{1000: int32(10), 1001: "z"}, specID: 3},
+ {partition: map[int]any{1000: int32(2), 1001: "z"}, specID: 4},
+ {partition: map[int]any{1000: int32(-1), 1001: "a"}, specID: 1},
+ {partition: map[int]any{1000: nil, 1001: "z"}, specID: 2},
+ {partition: map[int]any{1000: int32(2), 1001: "z"}, specID: 1},
+ {partition: map[int]any{1000: int32(2), 1001: "a"}, specID: 5},
+ }
+ require.NoError(t, sortPartitionStatistics(rows, pt))
+ require.Equal(t, []int32{2, 1, 5, 1, 4, 3}, []int32{
+ rows[0].specID, rows[1].specID, rows[2].specID,
+ rows[3].specID, rows[4].specID, rows[5].specID,
+ })
+ require.Nil(t, rows[0].partition[1000])
+ require.EqualValues(t, -1, rows[1].partition[1000])
+ require.EqualValues(t, 10, rows[5].partition[1000])
+}
+
+func TestSortPartitionStatisticsFloatAndUUID(t *testing.T) {
+ t.Run("float", func(t *testing.T) {
+ pt := &iceberg.StructType{FieldList: []iceberg.NestedField{
+ {ID: 1000, Name: "value", Type:
iceberg.PrimitiveTypes.Float64},
+ }}
+ rows := []*inspectPartitionAggregate{
+ {partition: map[int]any{1000: math.NaN()}, specID: 6},
+ {partition: map[int]any{1000: float64(0)}, specID: 3},
+ {partition: map[int]any{1000: math.Inf(1)}, specID: 5},
+ {partition: map[int]any{1000: math.Copysign(0, -1)},
specID: 2},
+ {partition: map[int]any{1000: math.Inf(-1)}, specID: 1},
+ {partition: map[int]any{1000:
math.Float64frombits(0xfff8000000000001)}, specID: 7},
+ }
+ require.NoError(t, sortPartitionStatistics(rows, pt))
+ require.Equal(t, []int32{1, 2, 3, 5, 6, 7}, []int32{
+ rows[0].specID, rows[1].specID, rows[2].specID,
+ rows[3].specID, rows[4].specID, rows[5].specID,
+ })
+ require.True(t, math.Signbit(rows[1].partition[1000].(float64)))
+ require.False(t,
math.Signbit(rows[2].partition[1000].(float64)))
+ require.True(t, math.IsNaN(rows[4].partition[1000].(float64)))
+ })
+ t.Run("uuid", func(t *testing.T) {
+ pt := &iceberg.StructType{FieldList: []iceberg.NestedField{
+ {ID: 1000, Name: "value", Type:
iceberg.PrimitiveTypes.UUID},
+ }}
+ rows := []*inspectPartitionAggregate{
+ {partition: map[int]any{1000:
uuid.MustParse("7fffffff-ffff-ffff-0000-000000000000")}, specID: 4},
+ {partition: map[int]any{1000:
uuid.MustParse("00000000-0000-0000-7fff-ffffffffffff")}, specID: 3},
+ {partition: map[int]any{1000:
uuid.MustParse("80000000-0000-0000-0000-000000000000")}, specID: 1},
+ {partition: map[int]any{1000:
uuid.MustParse("00000000-0000-0000-8000-000000000000")}, specID: 2},
+ }
+ require.NoError(t, sortPartitionStatistics(rows, pt))
+ require.Equal(t, []int32{1, 2, 3, 4}, []int32{
+ rows[0].specID, rows[1].specID, rows[2].specID,
rows[3].specID,
+ })
+ })
+}
+
+func TestPartitionStatisticsParquetWrite(t *testing.T) {
+ ctx := t.Context()
+ fs := iceio.NewMemFS()
+ path := "mem://bucket/table/metadata/v1-stats.parquet"
+ rows := []*inspectPartitionAggregate{
+ {
+ specID: 0, partition: map[int]any{1000: int32(-3),
1001: "a"},
+ dataRecordCount: 17, dataFileCount: 2, dataFileSize:
2048,
+ },
+ {
+ specID: 1, partition: map[int]any{1000: int32(10),
1001: nil},
+ dataRecordCount: 3, dataFileCount: 1, dataFileSize: 512,
+ },
+ }
+ size, err := writePartitionStatisticsParquet(ctx, fs, path,
statsTestPartitionType(), rows)
+ require.NoError(t, err)
+ require.Positive(t, size)
+
+ f, err := fs.Open(ctx, path)
+ require.NoError(t, err)
+ data, err := io.ReadAll(f)
+ require.NoError(t, err)
+ require.NoError(t, f.Close())
+ require.EqualValues(t, len(data), size)
+ require.True(t, bytes.HasPrefix(data, []byte("PAR1")))
+ require.True(t, bytes.HasSuffix(data, []byte("PAR1")))
+ reader, err := file.NewParquetReader(bytes.NewReader(data))
+ require.NoError(t, err)
+ defer reader.Close()
+ arrowReader, err := pqarrow.NewFileReader(reader,
pqarrow.ArrowReadProperties{}, memory.DefaultAllocator)
+ require.NoError(t, err)
+ table, err := arrowReader.ReadTable(ctx)
+ require.NoError(t, err)
+ defer table.Release()
+ require.EqualValues(t, 2, table.NumRows())
+ partitions := table.Column(0).Data().Chunk(0).(*array.Struct)
+ require.Equal(t, int32(-3), partitions.Field(0).(*array.Int32).Value(0))
+ require.Equal(t, int32(10), partitions.Field(0).(*array.Int32).Value(1))
+ require.True(t, partitions.Field(1).IsNull(1))
+ require.Equal(t, int64(17),
table.Column(2).Data().Chunk(0).(*array.Int64).Value(0))
+ require.Equal(t, int64(3),
table.Column(5).Data().Chunk(0).(*array.Int64).Value(1))
+
+ // Cancellation must not leave an incomplete metadata file.
+ cancelled, cancel := context.WithCancel(ctx)
+ cancel()
+ failedPath := "mem://bucket/table/metadata/cancelled.parquet"
+ _, err = writePartitionStatisticsParquet(cancelled, fs, failedPath,
statsTestPartitionType(), rows)
+ require.ErrorIs(t, err, context.Canceled)
+ _, err = fs.Open(ctx, failedPath)
+ require.Error(t, err)
+}
+
+func TestCollectPartitionStatisticsFromLiveManifests(t *testing.T) {
+ spec := partitionedSpec()
+ const snapshotID = int64(1)
+ sequenceNumber := int64(1)
+ entry := func(name string, partition int32, count int64, status
iceberg.ManifestEntryStatus) iceberg.ManifestEntry {
+ file := newTestDataFileWithCount(
+ t, spec,
"mem://default/table-location/data/"+name+".parquet",
+ map[int]any{1000: partition}, count)
+
+ return iceberg.NewManifestEntry(status, new(snapshotID),
+ &sequenceNumber, &sequenceNumber, file)
+ }
+ tbl := inspectFilesTableWithManifests(t, spec, inspectManifestSpec{
+ content: iceberg.ManifestContentData,
+ entries: []iceberg.ManifestEntry{
+ entry("first", 2, 5, iceberg.EntryStatusADDED),
+ entry("second", 2, 7, iceberg.EntryStatusADDED),
+ entry("third", 10, 3, iceberg.EntryStatusADDED),
+ entry("removed", 10, 100, iceberg.EntryStatusDELETED),
+ },
+ })
+ ctx := t.Context()
+ fs, err := tbl.FS(ctx)
+ require.NoError(t, err)
+ pt, err := inspectPartitionType(tbl.Metadata())
+ require.NoError(t, err)
+
+ rows, err := collectPartitionStatistics(ctx, fs,
*tbl.CurrentSnapshot(), pt, tbl.Metadata().PartitionSpecs())
+ require.NoError(t, err)
+ require.Len(t, rows, 2)
+ require.NoError(t, sortPartitionStatistics(rows, pt))
+ require.Equal(t, int32(2), rows[0].partition[1000])
+ require.Equal(t, int64(12), rows[0].dataRecordCount)
+ require.Equal(t, int32(2), rows[0].dataFileCount)
+ require.Equal(t, int32(10), rows[1].partition[1000])
+ require.Equal(t, int64(3), rows[1].dataRecordCount)
+ require.Equal(t, int32(1), rows[1].dataFileCount)
+}
+
+func TestCollectPartitionStatisticsDecimalFromManifest(t *testing.T) {
+ decimalType := iceberg.MustDecimalTypeOf(10, 2)
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "amount", Transform:
iceberg.IdentityTransform{},
+ })
+ schema := iceberg.NewSchema(0, iceberg.NestedField{
+ ID: 1, Name: "amount", Type: decimalType, Required: true,
+ })
+ partitionType := &iceberg.StructType{FieldList: []iceberg.NestedField{
+ {ID: 1000, Name: "amount", Type: decimalType},
+ }}
+ amount := iceberg.Decimal{Val: decimal128.FromI64(1234), Scale: 2}
+ const snapshotID = int64(7)
+ files := []iceberg.DataFile{
+ newTestDataFileWithCount(t, spec,
"mem://stats/decimal-a.parquet", map[int]any{1000: amount}, 4),
+ newTestDataFileWithCount(t, spec,
"mem://stats/decimal-b.parquet", map[int]any{1000:
iceberg.DecimalLiteral(amount)}, 6),
+ }
+ entries := make([]iceberg.ManifestEntry, 0, len(files))
+ for _, df := range files {
+ entries = append(entries,
iceberg.NewManifestEntry(iceberg.EntryStatusADDED, new(snapshotID), nil, nil,
df))
+ }
+
+ const manifestPath = "mem://stats/decimal-manifest.avro"
+ var encoded bytes.Buffer
+ _, err := iceberg.WriteManifest(manifestPath, &encoded, 1, spec,
schema, snapshotID, entries)
+ require.NoError(t, err)
+ fs := iceio.NewMemFS()
+ require.NoError(t, fs.WriteFile(t.Context(), manifestPath,
encoded.Bytes()))
+
+ snapshot := Snapshot{SnapshotID: snapshotID, ManifestLocations:
[]string{manifestPath}}
+ rows, err := collectPartitionStatistics(t.Context(), fs, snapshot,
partitionType, []iceberg.PartitionSpec{spec})
+ require.NoError(t, err)
+ require.Len(t, rows, 1)
+ require.Equal(t, amount, rows[0].partition[1000])
+ require.Equal(t, int64(10), rows[0].dataRecordCount)
+ require.Equal(t, int32(2), rows[0].dataFileCount)
+ require.NoError(t, sortPartitionStatistics(rows, partitionType))
+
+ const statsPath = "mem://stats/decimal-stats.parquet"
+ _, err = writePartitionStatisticsParquet(t.Context(), fs, statsPath,
partitionType, rows)
+ require.NoError(t, err)
+ stored, err := fs.Open(t.Context(), statsPath)
+ require.NoError(t, err)
+ content, err := io.ReadAll(stored)
+ require.NoError(t, err)
+ require.NoError(t, stored.Close())
+ reader, err := file.NewParquetReader(bytes.NewReader(content))
+ require.NoError(t, err)
+ defer reader.Close()
+ arrowReader, err := pqarrow.NewFileReader(reader,
pqarrow.ArrowReadProperties{}, memory.DefaultAllocator)
+ require.NoError(t, err)
+ result, err := arrowReader.ReadTable(t.Context())
+ require.NoError(t, err)
+ defer result.Release()
+ require.EqualValues(t, 1, result.NumRows())
+ partition := result.Column(0).Data().Chunk(0).(*array.Struct)
+ require.Equal(t, amount.Val,
partition.Field(0).(*array.Decimal128).Value(0))
+}
+
+func TestTransactionPartitionStatisticsRegistration(t *testing.T) {
+ tbl := inspectTestTable(Identifier{"db", "stats"},
map[string]SnapshotRef{
+ MainBranch: {SnapshotID: 103, Type: BranchRef},
+ })
+ tx := tbl.NewTransaction()
+ stats := PartitionStatisticsFile{
+ SnapshotID: 103,
+ StatisticsPath:
"s3://warehouse/db/stats/metadata/stats.parquet",
+ FileSizeInBytes: 512,
+ }
+
+ require.NoError(t, tx.SetPartitionStatistics(stats))
+ meta, err := tx.meta.Build()
+ require.NoError(t, err)
+ require.Equal(t, []PartitionStatisticsFile{stats},
slices.Collect(meta.PartitionStatistics()))
+
+ replacement := stats
+ replacement.FileSizeInBytes = 1024
+ require.NoError(t, tx.SetPartitionStatistics(replacement))
+ meta, err = tx.meta.Build()
+ require.NoError(t, err)
+ require.Equal(t, []PartitionStatisticsFile{replacement},
slices.Collect(meta.PartitionStatistics()))
+
+ require.NoError(t, tx.RemovePartitionStatistics(stats.SnapshotID))
+ meta, err = tx.meta.Build()
+ require.NoError(t, err)
+ require.Empty(t, slices.Collect(meta.PartitionStatistics()))
+
+ require.ErrorIs(t, tx.SetPartitionStatistics(PartitionStatisticsFile{
+ SnapshotID: 999, StatisticsPath: stats.StatisticsPath,
FileSizeInBytes: 512,
+ }), ErrSnapshotNotFound)
+ require.ErrorIs(t, tx.SetPartitionStatistics(PartitionStatisticsFile{
+ SnapshotID: 103, FileSizeInBytes: 512,
+ }), iceberg.ErrInvalidArgument)
+ require.ErrorIs(t, tx.SetPartitionStatistics(PartitionStatisticsFile{
+ SnapshotID: 103, StatisticsPath: stats.StatisticsPath,
FileSizeInBytes: -1,
+ }), iceberg.ErrInvalidArgument)
+ var nilTx *Transaction
+ require.Error(t, nilTx.RemovePartitionStatistics(103))
+ require.Error(t, nilTx.SetPartitionStatistics(stats))
+}
+
+func TestComputePartitionStatisticsRejectsNonV1AndMissingSnapshot(t
*testing.T) {
+ tbl := inspectTestTable(Identifier{"db", "stats"},
map[string]SnapshotRef{
+ MainBranch: {SnapshotID: 103, Type: BranchRef},
+ })
+ tx := tbl.NewTransaction()
+ _, err := tx.ComputePartitionStatistics(t.Context())
+ require.True(t, errors.Is(err, iceberg.ErrNotImplemented))
+
+ tx.meta.formatVersion = 1
+ _, err = tx.ComputePartitionStatistics(t.Context())
+ // The fixture's schema is unpartitioned. Fail before any filesystem IO.
+ require.True(t, errors.Is(err, iceberg.ErrNotImplemented))
+}
+
+type statsSchemaHistory struct {
+ Metadata
+ current *iceberg.Schema
+ schemas []*iceberg.Schema
+ specs []iceberg.PartitionSpec
+}
+
+func (m statsSchemaHistory) CurrentSchema() *iceberg.Schema { return
m.current }
+func (m statsSchemaHistory) Schemas() []*iceberg.Schema { return
m.schemas }
+func (m statsSchemaHistory) PartitionSpecs() []iceberg.PartitionSpec { return
m.specs }
+
+func TestPartitionStatisticsRejectsDroppedAndPromotedSources(t *testing.T) {
+ old := iceberg.NewSchema(1, iceberg.NestedField{ID: 1, Name: "id",
Type: iceberg.PrimitiveTypes.Int32})
+ promoted := iceberg.NewSchema(2, iceberg.NestedField{ID: 1, Name: "id",
Type: iceberg.PrimitiveTypes.Int64})
+ dropped := iceberg.NewSchema(3, iceberg.NestedField{ID: 2, Name:
"other", Type: iceberg.PrimitiveTypes.String})
+ meta := statsSchemaHistory{current: old, schemas:
[]*iceberg.Schema{old}, specs: []iceberg.PartitionSpec{partitionedSpec()}}
+ require.NoError(t, validatePartitionStatisticsSourceTypes(meta))
+ meta.current, meta.schemas = promoted, []*iceberg.Schema{old, promoted}
+ require.ErrorIs(t, validatePartitionStatisticsSourceTypes(meta),
iceberg.ErrNotImplemented)
+ meta.specs =
[]iceberg.PartitionSpec{iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "bucket", Transform:
iceberg.BucketTransform{NumBuckets: 4},
+ })}
+ require.NoError(t, validatePartitionStatisticsSourceTypes(meta))
+ meta.current, meta.schemas = dropped, []*iceberg.Schema{old, dropped}
+ require.ErrorIs(t, validatePartitionStatisticsSourceTypes(meta),
iceberg.ErrNotImplemented)
+}
+
+func TestComputePartitionStatisticsWritesAndStages(t *testing.T) {
+ spec := partitionedSpec()
+ const snapshotID = int64(1)
+ sequenceNumber := int64(1)
+ data := newTestDataFileWithCount(t, spec,
"mem://default/table-location/data/a.parquet", map[int]any{1000: int32(7)}, 12)
+ entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED,
new(snapshotID), &sequenceNumber, &sequenceNumber, data)
+ tbl := inspectFilesTableWithManifests(t, spec,
inspectManifestSpec{content: iceberg.ManifestContentData, entries:
[]iceberg.ManifestEntry{entry}})
+ base := tbl.Metadata().(*metadataV2).commonMetadata
+ base.FormatVersion = 1
+ fs, err := tbl.FS(t.Context())
+ require.NoError(t, err)
+ tbl = New(tbl.Identifier(), &metadataV1{commonMetadata: base},
"metadata.json", func(context.Context) (iceio.IO, error) { return fs, nil },
nil)
+ tx := tbl.NewTransaction()
+ stats, err := tx.ComputePartitionStatistics(t.Context())
+ require.NoError(t, err)
+ require.Equal(t, snapshotID, stats.SnapshotID)
+ metadata, err := tx.meta.Build()
+ require.NoError(t, err)
+ require.Equal(t, []PartitionStatisticsFile{*stats},
slices.Collect(metadata.PartitionStatistics()))
+ f, err := fs.Open(t.Context(), stats.StatisticsPath)
+ require.NoError(t, err)
+ require.NoError(t, f.Close())
+}
+
+func TestCollectPartitionStatisticsRejectsDuplicateLiveFiles(t *testing.T) {
+ spec := partitionedSpec()
+ const snapshotID = int64(1)
+ seq := int64(1)
+ file := newTestDataFileWithCount(t, spec,
+ "mem://default/table-location/data/duplicate.parquet",
+ map[int]any{1000: int32(2)}, 5)
+ entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED,
new(snapshotID),
+ &seq, &seq, file)
+ tbl := inspectFilesTableWithManifests(t, spec, inspectManifestSpec{
+ content: iceberg.ManifestContentData,
+ entries: []iceberg.ManifestEntry{entry, entry},
+ })
+ fs, err := tbl.FS(t.Context())
+ require.NoError(t, err)
+ pt, err := inspectPartitionType(tbl.Metadata())
+ require.NoError(t, err)
+
+ _, err = collectPartitionStatistics(t.Context(), fs,
*tbl.CurrentSnapshot(), pt, tbl.Metadata().PartitionSpecs())
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+ require.ErrorContains(t, err, "duplicate live data file")
+}
+
+func TestPartitionStatisticsEmbeddedManifests(t *testing.T) {
+ fs := iceio.NewMemFS()
+ spec0 := partitionedSpec()
+ spec7 := iceberg.NewPartitionSpecID(7,
iceberg.PartitionField{SourceIDs: []int{1}, FieldID: 1000, Name: "id",
Transform: iceberg.IdentityTransform{}})
+ write := func(path string, spec iceberg.PartitionSpec, name string,
value any, count int64) {
+ file := newTestDataFileWithCount(t, spec,
"mem://stats/"+name+".parquet", map[int]any{1000: value}, count)
+ entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED,
new(int64(1)), nil, nil, file)
+ var out bytes.Buffer
+ _, err := iceberg.WriteManifest(path, &out, 1, spec,
simpleSchema(), 1, []iceberg.ManifestEntry{entry})
+ require.NoError(t, err)
+ require.NoError(t, fs.WriteFile(t.Context(), path, out.Bytes()))
+ }
+ write("mem://stats/spec0.avro", spec0, "a", int32(3), 10)
+ write("mem://stats/spec7.avro", spec7, "b", int32(3), 20)
+ write("mem://stats/null.avro", spec7, "null", nil, 1)
+ snapshot := Snapshot{SnapshotID: 1, ManifestLocations:
[]string{"mem://stats/spec0.avro", "mem://stats/spec7.avro",
"mem://stats/null.avro"}}
+ pt := &iceberg.StructType{FieldList: []iceberg.NestedField{{ID: 1000,
Name: "id", Type: iceberg.Int32Type{}}}}
+ rows, err := collectPartitionStatistics(t.Context(), fs, snapshot, pt,
[]iceberg.PartitionSpec{spec0, spec7})
+ require.NoError(t, err)
+ require.Len(t, rows, 3)
+ require.NoError(t, sortPartitionStatistics(rows, pt))
+ require.Nil(t, rows[0].partition[1000])
+ require.Equal(t, int32(7), rows[0].specID)
+ require.Equal(t, int32(0), rows[1].specID)
+ require.Equal(t, int64(10), rows[1].dataRecordCount)
+ require.Equal(t, int32(7), rows[2].specID)
+ require.Equal(t, int64(20), rows[2].dataRecordCount)
+ statsPath := "mem://stats/partition-stats.parquet"
+ _, err = writePartitionStatisticsParquet(t.Context(), fs, statsPath,
pt, rows)
+ require.NoError(t, err)
+ statsFile, err := fs.Open(t.Context(), statsPath)
+ require.NoError(t, err)
+ encoded, err := io.ReadAll(statsFile)
+ require.NoError(t, err)
+ require.NoError(t, statsFile.Close())
+ parquetReader, err := file.NewParquetReader(bytes.NewReader(encoded))
+ require.NoError(t, err)
+ defer parquetReader.Close()
+ arrowReader, err := pqarrow.NewFileReader(parquetReader,
pqarrow.ArrowReadProperties{}, memory.DefaultAllocator)
+ require.NoError(t, err)
+ result, err := arrowReader.ReadTable(t.Context())
+ require.NoError(t, err)
+ defer result.Release()
+ require.EqualValues(t, 3, result.NumRows())
+ require.Equal(t, []int32{7, 0, 7},
result.Column(1).Data().Chunk(0).(*array.Int32).Int32Values())
+ _, err = collectPartitionStatistics(t.Context(), fs, snapshot, pt,
[]iceberg.PartitionSpec{spec0})
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+ wrong := iceberg.NewPartitionSpecID(7,
iceberg.PartitionField{SourceIDs: []int{1}, FieldID: 1000, Name: "id",
Transform: iceberg.BucketTransform{NumBuckets: 4}})
+ _, err = collectPartitionStatistics(t.Context(), fs, snapshot, pt,
[]iceberg.PartitionSpec{spec0, wrong})
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+}
+
+func TestPartitionStatisticsEmbeddedManifestWithoutSpecID(t *testing.T) {
+ fs := iceio.NewMemFS()
+ spec := partitionedSpec()
+ file := newTestDataFileWithCount(t, spec, "mem://stats/a.parquet",
map[int]any{1000: int32(3)}, 10)
+ entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED,
new(int64(1)), nil, nil, file)
+ var original bytes.Buffer
+ _, err := iceberg.WriteManifest("mem://stats/a.avro", &original, 1,
spec, simpleSchema(), 1, []iceberg.ManifestEntry{entry})
+ require.NoError(t, err)
+ reader, err := ocf.NewReader(bytes.NewReader(original.Bytes()))
+ require.NoError(t, err)
+ metadata := make(map[string][]byte)
+ for k, v := range reader.Metadata() {
+ if k != "partition-spec-id" && k != "avro.schema" && k !=
"avro.codec" {
+ metadata[k] = v
+ }
+ }
+ schema := reader.Schema()
+ require.NoError(t, reader.Close())
+ writeEmpty := func(path string) {
+ var rewritten bytes.Buffer
+ writer, err := ocf.NewWriter(&rewritten, schema,
ocf.WithMetadata(metadata))
+ require.NoError(t, err)
+ require.NoError(t, writer.Close())
+ require.NoError(t, fs.WriteFile(t.Context(), path,
rewritten.Bytes()))
+ }
+ const missingID = "mem://stats/missing.avro"
+ writeEmpty(missingID)
+ pt := &iceberg.StructType{FieldList: []iceberg.NestedField{{ID: 1000,
Name: "id", Type: iceberg.Int32Type{}}}}
+ snapshot := Snapshot{SnapshotID: 1, ManifestLocations:
[]string{missingID}}
+ rows, err := collectPartitionStatistics(t.Context(), fs, snapshot, pt,
[]iceberg.PartitionSpec{spec})
+ require.NoError(t, err)
+ require.Empty(t, rows)
+
+ fields := make([]iceberg.PartitionField, 0, spec.NumFields())
+ for _, field := range spec.Fields() {
+ fields = append(fields, field)
+ }
+ duplicateSpec := iceberg.NewPartitionSpecID(spec.ID()+7, fields...)
+ _, err = collectPartitionStatistics(t.Context(), fs, snapshot, pt,
[]iceberg.PartitionSpec{spec, duplicateSpec})
+ require.ErrorIs(t, err, iceberg.ErrNotImplemented)
+
+ wrongFields := slices.Clone(fields)
+ wrongFields[0].Transform = iceberg.BucketTransform{NumBuckets: 4}
+ wrongSpec := iceberg.NewPartitionSpecID(spec.ID(), wrongFields...)
+ _, err = collectPartitionStatistics(t.Context(), fs, snapshot, pt,
[]iceberg.PartitionSpec{wrongSpec})
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+
+ // Historical v1 partition fields may also omit their field IDs.
+ var rawFields []map[string]json.RawMessage
+ require.NoError(t, json.Unmarshal(metadata["partition-spec"],
&rawFields))
+ for _, field := range rawFields {
+ delete(field, "field-id")
+ }
+ metadata["partition-spec"], err = json.Marshal(rawFields)
+ require.NoError(t, err)
+ const missingBoth = "mem://stats/missing-both.avro"
+ writeEmpty(missingBoth)
+ snapshot.ManifestLocations = []string{missingBoth}
+ rows, err = collectPartitionStatistics(t.Context(), fs, snapshot, pt,
[]iceberg.PartitionSpec{spec})
+ require.NoError(t, err)
+ require.Empty(t, rows)
+
+ // The generic manifest reader requires an ID, but v1 field IDs are
+ // still optional even if partition-spec-id is present.
+ metadata["partition-spec-id"] = []byte(strconv.Itoa(spec.ID()))
+ const missingFieldIDs = "mem://stats/missing-field-ids.avro"
+ writeEmpty(missingFieldIDs)
+ snapshot.ManifestLocations = []string{missingFieldIDs}
+ rows, err = collectPartitionStatistics(t.Context(), fs, snapshot, pt,
[]iceberg.PartitionSpec{spec})
+ require.NoError(t, err)
+ require.Empty(t, rows)
+}
+
+func TestPartitionStatisticsRequiresPartitionFieldIDs(t *testing.T) {
+ spec := partitionedSpec()
+ require.NoError(t, validatePartitionStatisticsTuple(spec, "file",
map[int]any{1000: nil}))
+ require.NoError(t, validatePartitionStatisticsTuple(spec, "file",
map[int]any{1000: int32(3)}))
+ require.ErrorIs(t, validatePartitionStatisticsTuple(spec, "file", nil),
iceberg.ErrInvalidArgument)
+ require.ErrorIs(t, validatePartitionStatisticsTuple(spec, "file",
map[int]any{1001: int32(3)}), iceberg.ErrInvalidArgument)
+}
+
+func TestComputePartitionStatisticsEmbeddedManifest(t *testing.T) {
+ fs := iceio.NewMemFS()
+ spec := partitionedSpec()
+ f := newTestDataFileWithCount(t, spec, "mem://table/data/a.parquet",
map[int]any{1000: int32(7)}, 12)
+ entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED,
new(int64(1)), nil, nil, f)
+ var out bytes.Buffer
+ const path = "mem://table/metadata/manifest.avro"
+ _, err := iceberg.WriteManifest(path, &out, 1, spec, simpleSchema(), 1,
[]iceberg.ManifestEntry{entry})
+ require.NoError(t, err)
+ require.NoError(t, fs.WriteFile(t.Context(), path, out.Bytes()))
+ metadata, err := NewMetadata(simpleSchema(), &spec, UnsortedSortOrder,
"mem://table", iceberg.Properties{PropertyFormatVersion: "1"})
+ require.NoError(t, err)
+ builder, err := MetadataBuilderFromBase(metadata, "")
+ require.NoError(t, err)
+ require.NoError(t, builder.AddSnapshot(&Snapshot{SnapshotID: 1,
TimestampMs: time.Now().UnixMilli() + 1000, ManifestLocations: []string{path}}))
+ require.NoError(t, builder.SetSnapshotRef(MainBranch, 1, BranchRef))
+ metadata, err = builder.Build()
+ require.NoError(t, err)
+ tbl := New(Identifier{"db", "table"}, metadata,
"mem://table/metadata.json", func(context.Context) (iceio.IO, error) { return
fs, nil }, nil)
+ tx := tbl.NewTransaction()
+ stats, err := tx.ComputePartitionStatistics(t.Context())
+ require.NoError(t, err)
+ require.Equal(t, int64(1), stats.SnapshotID)
+ result, err := tx.meta.Build()
+ require.NoError(t, err)
+ require.Equal(t, []PartitionStatisticsFile{*stats},
slices.Collect(result.PartitionStatistics()))
+ stored, err := fs.Open(t.Context(), stats.StatisticsPath)
+ require.NoError(t, err)
+ require.NoError(t, stored.Close())
+}
