zeroshade commented on code in PR #2396: URL: https://github.com/apache/iceberg-go/pull/2396#discussion_r4470244897
########## table/partition_statistics_write.go: ########## @@ -0,0 +1,640 @@ +// 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" + "fmt" + "math" + "reflect" + "slices" + "strconv" + + "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) +} + +// UpdatePartitionStatistics 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) UpdatePartitionStatistics(ctx context.Context) (*PartitionStatisticsFile, error) { Review Comment: In Java, `table.updatePartitionStatistics()` is the set/remove builder, and the compute step is `PartitionStatsHandler.computeAndWriteStatsFile`. Since this is new exported API, consider `ComputePartitionStatistics`. ########## table/partition_statistics_write.go: ########## @@ -0,0 +1,640 @@ +// 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" + "fmt" + "math" + "reflect" + "slices" + "strconv" + + "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) +} + +// UpdatePartitionStatistics 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) UpdatePartitionStatistics(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 +} + +// inspectPartitionType omits fields whose source is no longer in the current Review Comment: The doc comment starts with `inspectPartitionType`; it should name `validatePartitionStatisticsSourceTypes`. ########## table/partition_statistics_write.go: ########## @@ -0,0 +1,640 @@ +// 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" + "fmt" + "math" + "reflect" + "slices" + "strconv" + + "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) +} + +// UpdatePartitionStatistics 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) UpdatePartitionStatistics(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 +} + +// inspectPartitionType omits fields whose source is no longer in the current +// schema. Statistics must instead include every historical partition field. +// Reject those tables until historical projection is supported. +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() + header, err := ocf.NewReader(f) + if err != nil { + return nil, fmt.Errorf("read embedded manifest %q: %w", manifest.FilePath(), err) + } + raw := header.Metadata()["partition-spec-id"] + var fields []iceberg.PartitionField + if err := json.Unmarshal(header.Metadata()["partition-spec"], &fields); err != nil { + return nil, fmt.Errorf("%w: embedded manifest %q has an invalid partition spec: %v", + iceberg.ErrInvalidArgument, manifest.FilePath(), err) + } + if err := header.Close(); err != nil { + return nil, err + } + // V1 manifests can omit both partition-spec-id and partition field IDs. + // The latter were assigned sequentially from 1000 by V1 writers. + for i := range fields { + if fields[i].FieldID == 0 { + fields[i].FieldID = iceberg.PartitionDataIDStart + i + } + } + + var id int64 + if len(raw) == 0 { + found := false + for candidateID, candidate := range specs { + actual := iceberg.NewPartitionSpecID(int(candidateID), fields...) + if !actual.Equals(candidate) { + continue + } + if found { + return nil, fmt.Errorf("%w: embedded manifest %q has an ambiguous partition spec without an ID", + iceberg.ErrNotImplemented, manifest.FilePath()) + } + id, found = int64(candidateID), true + } + if !found { + return nil, fmt.Errorf("%w: embedded manifest %q has no matching partition spec", + iceberg.ErrInvalidArgument, manifest.FilePath()) + } + } else { + id, err = strconv.ParseInt(string(raw), 10, 32) + if err != nil || id < 0 { + return nil, fmt.Errorf("%w: embedded manifest %q has invalid partition spec ID", iceberg.ErrInvalidArgument, manifest.FilePath()) + } + } + expected, ok := specs[int32(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), fields...) + if !actual.Equals(expected) { + return nil, fmt.Errorf("%w: embedded manifest %q partition spec differs from table metadata", iceberg.ErrInvalidArgument, manifest.FilePath()) + } + descriptor := iceberg.NewManifestBuilder(1, manifest.FilePath(), manifest.Length(), int32(id), manifest.SnapshotID()). + AddedFiles(-1).ExistingFiles(-1).DeletedFiles(-1).AddedRows(-1).ExistingRows(-1).DeletedRows(-1).Build() + if _, err := f.Seek(0, 0); err != nil { + return nil, err + } + reader, err := iceberg.NewManifestReader(descriptor, f) Review Comment: This builds a manifest reader and closes it straight away; only the descriptor is used, and the entries are read again through `manifest.Entries`. Could this be removed so each embedded manifest is opened once? ########## table/partition_statistics_write.go: ########## @@ -0,0 +1,640 @@ +// 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" + "fmt" + "math" + "reflect" + "slices" + "strconv" + + "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) +} + +// UpdatePartitionStatistics 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) UpdatePartitionStatistics(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 +} + +// inspectPartitionType omits fields whose source is no longer in the current +// schema. Statistics must instead include every historical partition field. +// Reject those tables until historical projection is supported. +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() + header, err := ocf.NewReader(f) + if err != nil { + return nil, fmt.Errorf("read embedded manifest %q: %w", manifest.FilePath(), err) + } + raw := header.Metadata()["partition-spec-id"] + var fields []iceberg.PartitionField + if err := json.Unmarshal(header.Metadata()["partition-spec"], &fields); err != nil { + return nil, fmt.Errorf("%w: embedded manifest %q has an invalid partition spec: %v", + iceberg.ErrInvalidArgument, manifest.FilePath(), err) + } + if err := header.Close(); err != nil { + return nil, err + } + // V1 manifests can omit both partition-spec-id and partition field IDs. + // The latter were assigned sequentially from 1000 by V1 writers. + for i := range fields { + if fields[i].FieldID == 0 { + fields[i].FieldID = iceberg.PartitionDataIDStart + i + } + } + + var id int64 + if len(raw) == 0 { + found := false + for candidateID, candidate := range specs { + actual := iceberg.NewPartitionSpecID(int(candidateID), fields...) + if !actual.Equals(candidate) { + continue + } + if found { + return nil, fmt.Errorf("%w: embedded manifest %q has an ambiguous partition spec without an ID", + iceberg.ErrNotImplemented, manifest.FilePath()) + } + id, found = int64(candidateID), true + } + if !found { + return nil, fmt.Errorf("%w: embedded manifest %q has no matching partition spec", + iceberg.ErrInvalidArgument, manifest.FilePath()) + } + } else { + id, err = strconv.ParseInt(string(raw), 10, 32) + if err != nil || id < 0 { + return nil, fmt.Errorf("%w: embedded manifest %q has invalid partition spec ID", iceberg.ErrInvalidArgument, manifest.FilePath()) + } + } + expected, ok := specs[int32(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), fields...) + if !actual.Equals(expected) { + return nil, fmt.Errorf("%w: embedded manifest %q partition spec differs from table metadata", iceberg.ErrInvalidArgument, manifest.FilePath()) + } + descriptor := iceberg.NewManifestBuilder(1, manifest.FilePath(), manifest.Length(), int32(id), manifest.SnapshotID()). + AddedFiles(-1).ExistingFiles(-1).DeletedFiles(-1).AddedRows(-1).ExistingRows(-1).DeletedRows(-1).Build() + if _, err := f.Seek(0, 0); err != nil { + return nil, err + } + reader, err := iceberg.NewManifestReader(descriptor, f) + if err != nil { + return nil, err + } + defer reader.Close() + + return descriptor, 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] + 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 { + for _, row := range rows { + for _, field := range partitionType.FieldList { + value := row.partition[field.ID] + if _, err := comparePartitionStatisticValues(value, value); err != nil { + return fmt.Errorf("partition field %q: %w", field.Name, err) + } + } + } + + 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 + } + 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: + if math.IsNaN(left.Float()) || math.IsNaN(right.Float()) { + return 0, fmt.Errorf("%w: NaN partition ordering is unsupported", + iceberg.ErrNotImplemented) + } + if (left.Float() == 0 && math.Signbit(left.Float())) || (right.Float() == 0 && math.Signbit(right.Float())) { + return 0, fmt.Errorf("%w: negative zero partition ordering is unsupported", iceberg.ErrNotImplemented) + } + + return compareOrderedStats(left.Float(), right.Float()), nil + case reflect.String: + return bytes.Compare([]byte(left.String()), []byte(right.String())), nil + case reflect.Slice, reflect.Array: Review Comment: UUIDs end up here and are compared as unsigned bytes, but Java's `UUID.compareTo` compares the two halves as signed 64-bit values, so UUIDs whose first byte is 0x80 or higher sort differently than in Java. ########## table/partition_statistics_write.go: ########## @@ -0,0 +1,640 @@ +// 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" + "fmt" + "math" + "reflect" + "slices" + "strconv" + + "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) +} + +// UpdatePartitionStatistics 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) UpdatePartitionStatistics(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 +} + +// inspectPartitionType omits fields whose source is no longer in the current +// schema. Statistics must instead include every historical partition field. +// Reject those tables until historical projection is supported. +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() + header, err := ocf.NewReader(f) + if err != nil { + return nil, fmt.Errorf("read embedded manifest %q: %w", manifest.FilePath(), err) + } + raw := header.Metadata()["partition-spec-id"] + var fields []iceberg.PartitionField + if err := json.Unmarshal(header.Metadata()["partition-spec"], &fields); err != nil { + return nil, fmt.Errorf("%w: embedded manifest %q has an invalid partition spec: %v", + iceberg.ErrInvalidArgument, manifest.FilePath(), err) + } + if err := header.Close(); err != nil { + return nil, err + } + // V1 manifests can omit both partition-spec-id and partition field IDs. + // The latter were assigned sequentially from 1000 by V1 writers. + for i := range fields { + if fields[i].FieldID == 0 { + fields[i].FieldID = iceberg.PartitionDataIDStart + i + } + } + + var id int64 + if len(raw) == 0 { + found := false + for candidateID, candidate := range specs { + actual := iceberg.NewPartitionSpecID(int(candidateID), fields...) + if !actual.Equals(candidate) { + continue + } + if found { + return nil, fmt.Errorf("%w: embedded manifest %q has an ambiguous partition spec without an ID", + iceberg.ErrNotImplemented, manifest.FilePath()) + } + id, found = int64(candidateID), true + } + if !found { + return nil, fmt.Errorf("%w: embedded manifest %q has no matching partition spec", + iceberg.ErrInvalidArgument, manifest.FilePath()) + } + } else { + id, err = strconv.ParseInt(string(raw), 10, 32) + if err != nil || id < 0 { + return nil, fmt.Errorf("%w: embedded manifest %q has invalid partition spec ID", iceberg.ErrInvalidArgument, manifest.FilePath()) + } + } + expected, ok := specs[int32(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), fields...) + if !actual.Equals(expected) { + return nil, fmt.Errorf("%w: embedded manifest %q partition spec differs from table metadata", iceberg.ErrInvalidArgument, manifest.FilePath()) + } + descriptor := iceberg.NewManifestBuilder(1, manifest.FilePath(), manifest.Length(), int32(id), manifest.SnapshotID()). + AddedFiles(-1).ExistingFiles(-1).DeletedFiles(-1).AddedRows(-1).ExistingRows(-1).DeletedRows(-1).Build() + if _, err := f.Seek(0, 0); err != nil { + return nil, err + } + reader, err := iceberg.NewManifestReader(descriptor, f) + if err != nil { + return nil, err + } + defer reader.Close() + + return descriptor, 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] + 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 { + for _, row := range rows { + for _, field := range partitionType.FieldList { + value := row.partition[field.ID] + if _, err := comparePartitionStatisticValues(value, value); err != nil { + return fmt.Errorf("partition field %q: %w", field.Name, err) + } + } + } + + 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 + } + 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: + if math.IsNaN(left.Float()) || math.IsNaN(right.Float()) { Review Comment: Java orders floats and doubles with `Comparator.naturalOrder()` (`Float.compareTo`/`Double.compareTo`), which handles −0.0 and NaN. Rejecting them here makes the whole computation fail for any float identity partition that has a NaN or −0.0 value. ########## table/partition_statistics_write.go: ########## @@ -0,0 +1,640 @@ +// 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" + "fmt" + "math" + "reflect" + "slices" + "strconv" + + "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) +} + +// UpdatePartitionStatistics 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) UpdatePartitionStatistics(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 +} + +// inspectPartitionType omits fields whose source is no longer in the current +// schema. Statistics must instead include every historical partition field. +// Reject those tables until historical projection is supported. +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() + header, err := ocf.NewReader(f) + if err != nil { + return nil, fmt.Errorf("read embedded manifest %q: %w", manifest.FilePath(), err) + } + raw := header.Metadata()["partition-spec-id"] + var fields []iceberg.PartitionField + if err := json.Unmarshal(header.Metadata()["partition-spec"], &fields); err != nil { + return nil, fmt.Errorf("%w: embedded manifest %q has an invalid partition spec: %v", + iceberg.ErrInvalidArgument, manifest.FilePath(), err) + } + if err := header.Close(); err != nil { + return nil, err + } + // V1 manifests can omit both partition-spec-id and partition field IDs. + // The latter were assigned sequentially from 1000 by V1 writers. + for i := range fields { + if fields[i].FieldID == 0 { + fields[i].FieldID = iceberg.PartitionDataIDStart + i + } + } + + var id int64 + if len(raw) == 0 { + found := false + for candidateID, candidate := range specs { + actual := iceberg.NewPartitionSpecID(int(candidateID), fields...) + if !actual.Equals(candidate) { + continue + } + if found { + return nil, fmt.Errorf("%w: embedded manifest %q has an ambiguous partition spec without an ID", + iceberg.ErrNotImplemented, manifest.FilePath()) + } + id, found = int64(candidateID), true + } + if !found { + return nil, fmt.Errorf("%w: embedded manifest %q has no matching partition spec", + iceberg.ErrInvalidArgument, manifest.FilePath()) + } + } else { + id, err = strconv.ParseInt(string(raw), 10, 32) + if err != nil || id < 0 { + return nil, fmt.Errorf("%w: embedded manifest %q has invalid partition spec ID", iceberg.ErrInvalidArgument, manifest.FilePath()) + } + } + expected, ok := specs[int32(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), fields...) + if !actual.Equals(expected) { + return nil, fmt.Errorf("%w: embedded manifest %q partition spec differs from table metadata", iceberg.ErrInvalidArgument, manifest.FilePath()) + } + descriptor := iceberg.NewManifestBuilder(1, manifest.FilePath(), manifest.Length(), int32(id), manifest.SnapshotID()). + AddedFiles(-1).ExistingFiles(-1).DeletedFiles(-1).AddedRows(-1).ExistingRows(-1).DeletedRows(-1).Build() + if _, err := f.Seek(0, 0); err != nil { + return nil, err + } + reader, err := iceberg.NewManifestReader(descriptor, f) + if err != nil { + return nil, err + } + defer reader.Close() + + return descriptor, 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] + if _, err := comparePartitionStatisticValues(value, value); err != nil { Review Comment: The same values are checked again per row in `sortPartitionStatistics` (line 418); one pass is enough. -- 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]
