You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/08/17 20:28:40 UTC
(iceberg-go) branch main updated: feat(table): add partitions metadata table (#1701)
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 f86d0ff05 feat(table): add partitions metadata table (#1701)
f86d0ff05 is described below
commit f86d0ff05fd0f721cc2e80dce96b5e266aba3872
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Mon Aug 17 22:28:34 2026 +0200
feat(table): add partitions metadata table (#1701)
* feat(table): add partitions metadata table
* chore: fix lint findings
* fix: share metadata file inspection helpers
* perf(table): index partition aggregates
* fix(table): match expired partition snapshot semantics
* fix(table): handle invalid partition metadata
* test(table): cover partition metadata aggregation
* fix(table): satisfy partition inspection lint
---
table/inspect_internal_test.go | 188 ++++++++++++++++++++++
table/inspect_partitions.go | 347 +++++++++++++++++++++++++++++++++++++++++
2 files changed, 535 insertions(+)
diff --git a/table/inspect_internal_test.go b/table/inspect_internal_test.go
index 91a8f65e3..71e75abe4 100644
--- a/table/inspect_internal_test.go
+++ b/table/inspect_internal_test.go
@@ -1992,6 +1992,194 @@ func TestInspectPartitionTypeRejectsIncompatibleFieldReuse(t *testing.T) {
require.ErrorIs(t, err, iceberg.ErrInvalidPartitionSpec)
}
+func TestPartitionsSchema(t *testing.T) {
+ partitionType := &iceberg.StructType{FieldList: []iceberg.NestedField{
+ {ID: 1000, Name: "bucket", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ }}
+ sc := PartitionsSchema(partitionType)
+ require.Equal(t, []string{
+ "partition", "spec_id", "record_count", "file_count",
+ "total_data_file_size_in_bytes", "position_delete_record_count", "position_delete_file_count",
+ "equality_delete_record_count", "equality_delete_file_count", "last_updated_at",
+ "last_updated_snapshot_id",
+ }, testFieldNames(sc))
+ require.Equal(t, 1, sc.Fields()[0].ID)
+ require.Equal(t, 4, sc.Fields()[1].ID)
+ require.Equal(t, 11, sc.Fields()[4].ID)
+ require.Equal(t, 10, sc.Fields()[10].ID)
+
+ unpartitioned := PartitionsSchema(&iceberg.StructType{})
+ require.Equal(t, []string{
+ "record_count", "file_count", "total_data_file_size_in_bytes",
+ "position_delete_record_count", "position_delete_file_count", "equality_delete_record_count",
+ "equality_delete_file_count", "last_updated_at", "last_updated_snapshot_id",
+ }, testFieldNames(unpartitioned))
+}
+
+func TestInspectPartitionAggregateTreeHandlesBinaryAndNaNValues(t *testing.T) {
+ tree := newInspectPartitionAggregateTree()
+ aggregate := &inspectPartitionAggregate{specID: 1}
+ record := partitionRecord{[]byte{1, 2, 3}, math.NaN()}
+ tree.insert(record, aggregate)
+
+ require.Same(t, aggregate, tree.lookup(partitionRecord{[]byte{1, 2, 3}, math.NaN()}))
+ require.Nil(t, tree.lookup(partitionRecord{[]byte{1, 2, 4}, math.NaN()}))
+}
+
+func TestInspectPartitionsAggregatesDataAndDeletes(t *testing.T) {
+ const snapshotID = int64(1)
+ spec := partitionedSpec()
+ txn, memIO := createTestTransactionWithMemIO(t, spec)
+ schema := simpleSchema()
+ partition := map[int]any{1000: int32(7)}
+ dataSequenceNumber := int64(1)
+ dataFiles := []iceberg.DataFile{
+ newTestDataFileWithCount(t, spec,
+ "mem://default/table-location/data/first.parquet", partition, 3),
+ newTestDataFileWithCount(t, spec,
+ "mem://default/table-location/data/second.parquet", partition, 2),
+ }
+ dataEntries := make([]iceberg.ManifestEntry, 0, len(dataFiles))
+ for _, file := range dataFiles {
+ dataEntries = append(dataEntries, iceberg.NewManifestEntry(
+ iceberg.EntryStatusADDED, int64Ptr(snapshotID), &dataSequenceNumber, &dataSequenceNumber, file))
+ }
+
+ dataManifestPath := "mem://default/table-location/metadata/data-manifest.avro"
+ var dataManifestBuf bytes.Buffer
+ dataManifest, err := iceberg.WriteManifest(dataManifestPath, &dataManifestBuf, 2, spec, schema, snapshotID, dataEntries)
+ require.NoError(t, err)
+ require.NoError(t, memIO.WriteFile(dataManifestPath, dataManifestBuf.Bytes()))
+
+ deleteFile := newTestPosDeleteFileForSpec(t, spec,
+ "mem://default/table-location/delete/positions.parquet", partition,
+ dataFiles[0].FilePath())
+ equalityDeleteBuilder, err := iceberg.NewDataFileBuilder(
+ spec,
+ iceberg.EntryContentEqDeletes,
+ "mem://default/table-location/delete/equality.parquet",
+ iceberg.ParquetFile,
+ partition,
+ nil,
+ nil,
+ 4,
+ 4,
+ )
+ require.NoError(t, err)
+ equalityDeleteFile := equalityDeleteBuilder.EqualityFieldIDs([]int{1}).Build()
+ deleteManifestPath := "mem://default/table-location/metadata/delete-manifest.avro"
+ var deleteManifestBuf bytes.Buffer
+ deleteWriter, err := iceberg.NewManifestWriter(2, &deleteManifestBuf, spec, schema, snapshotID,
+ iceberg.WithManifestWriterContent(iceberg.ManifestContentDeletes))
+ require.NoError(t, err)
+ require.NoError(t, deleteWriter.Add(iceberg.NewManifestEntry(
+ iceberg.EntryStatusADDED, int64Ptr(snapshotID), &dataSequenceNumber, &dataSequenceNumber, deleteFile)))
+ require.NoError(t, deleteWriter.Add(iceberg.NewManifestEntry(
+ iceberg.EntryStatusADDED, int64Ptr(snapshotID), &dataSequenceNumber, &dataSequenceNumber, equalityDeleteFile)))
+ require.NoError(t, deleteWriter.Close())
+ deleteManifest, err := deleteWriter.ToManifestFile(deleteManifestPath, int64(deleteManifestBuf.Len()),
+ iceberg.WithManifestFileContent(iceberg.ManifestContentDeletes))
+ require.NoError(t, err)
+ require.NoError(t, memIO.WriteFile(deleteManifestPath, deleteManifestBuf.Bytes()))
+
+ manifestListPath := "mem://default/table-location/metadata/snap-1-manifest-list.avro"
+ var listBuf bytes.Buffer
+ manifestListSequenceNumber := int64(1)
+ require.NoError(t, iceberg.WriteManifestList(2, &listBuf, snapshotID, nil,
+ &manifestListSequenceNumber, 0, []iceberg.ManifestFile{dataManifest, deleteManifest}))
+ require.NoError(t, memIO.WriteFile(manifestListPath, listBuf.Bytes()))
+
+ txn.meta.snapshotList = []Snapshot{{
+ SnapshotID: snapshotID,
+ ManifestList: manifestListPath,
+ SequenceNumber: manifestListSequenceNumber,
+ TimestampMs: 2000,
+ }}
+ txn.meta.currentSnapshotID = int64Ptr(snapshotID)
+ built, err := txn.meta.Build()
+ require.NoError(t, err)
+
+ tbl := New(Identifier{"db", "tbl"}, built, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return memIO, nil }, nil)
+ rr, err := tbl.Inspect().Partitions(context.Background())
+ require.NoError(t, err)
+ defer rr.Release()
+
+ record := collectRecord(t, rr)
+ defer record.Release()
+ require.EqualValues(t, 1, record.NumRows())
+
+ partitionValues := record.Column(0).(*array.Struct)
+ require.EqualValues(t, 7, partitionValues.Field(0).(*array.Int32).Value(0))
+ require.EqualValues(t, spec.ID(), record.Column(1).(*array.Int32).Value(0))
+ require.EqualValues(t, 5, record.Column(2).(*array.Int64).Value(0))
+ require.EqualValues(t, 2, record.Column(3).(*array.Int32).Value(0))
+ require.EqualValues(t, 5, record.Column(4).(*array.Int64).Value(0))
+ require.EqualValues(t, 1, record.Column(5).(*array.Int64).Value(0))
+ require.EqualValues(t, 1, record.Column(6).(*array.Int32).Value(0))
+ require.EqualValues(t, 4, record.Column(7).(*array.Int64).Value(0))
+ require.EqualValues(t, 1, record.Column(8).(*array.Int32).Value(0))
+ require.EqualValues(t, 2000*1000, record.Column(9).(*array.Timestamp).Value(0))
+ require.EqualValues(t, snapshotID, record.Column(10).(*array.Int64).Value(0))
+}
+
+func TestInspectPartitionsLeavesSpecIDUnsetForExpiredSnapshot(t *testing.T) {
+ const (
+ expiredSnapshotID = int64(1)
+ currentSnapshotID = int64(2)
+ )
+ spec := iceberg.NewPartitionSpecID(7, iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "id", Transform: iceberg.IdentityTransform{},
+ })
+ txn, memIO := createTestTransactionWithMemIO(t, spec)
+ schema := simpleSchema()
+ file := newTestDataFile(t, spec,
+ "mem://default/table-location/data/expired.parquet", map[int]any{1000: int32(7)})
+ sequenceNumber := int64(1)
+ manifestPath := "mem://default/table-location/metadata/expired-manifest.avro"
+ var manifestBuf bytes.Buffer
+ _, err := iceberg.WriteManifest(manifestPath, &manifestBuf, 2, spec, schema, expiredSnapshotID,
+ []iceberg.ManifestEntry{iceberg.NewManifestEntry(
+ iceberg.EntryStatusADDED, int64Ptr(expiredSnapshotID), &sequenceNumber, &sequenceNumber, file)})
+ require.NoError(t, err)
+ require.NoError(t, memIO.WriteFile(manifestPath, manifestBuf.Bytes()))
+ manifest := iceberg.NewManifestFile(2, manifestPath, int64(manifestBuf.Len()), int32(spec.ID()), expiredSnapshotID).
+ SequenceNum(sequenceNumber, sequenceNumber).
+ AddedFiles(1).
+ AddedRows(1).
+ Build()
+
+ manifestListPath := "mem://default/table-location/metadata/snap-2-manifest-list.avro"
+ var listBuf bytes.Buffer
+ currentSequenceNumber := int64(2)
+ require.NoError(t, iceberg.WriteManifestList(2, &listBuf, currentSnapshotID, nil,
+ ¤tSequenceNumber, 0, []iceberg.ManifestFile{manifest}))
+ require.NoError(t, memIO.WriteFile(manifestListPath, listBuf.Bytes()))
+
+ txn.meta.snapshotList = []Snapshot{{
+ SnapshotID: currentSnapshotID,
+ ManifestList: manifestListPath,
+ SequenceNumber: currentSequenceNumber,
+ TimestampMs: 2000,
+ }}
+ txn.meta.currentSnapshotID = int64Ptr(currentSnapshotID)
+ built, err := txn.meta.Build()
+ require.NoError(t, err)
+
+ tbl := New(Identifier{"db", "tbl"}, built, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return memIO, nil }, nil)
+ rr, err := tbl.Inspect().Partitions(context.Background())
+ require.NoError(t, err)
+ defer rr.Release()
+
+ record := collectRecord(t, rr)
+ defer record.Release()
+ require.EqualValues(t, 1, record.NumRows())
+ require.EqualValues(t, 0, record.Column(1).(*array.Int32).Value(0))
+ require.True(t, record.Column(9).IsNull(0))
+ require.True(t, record.Column(10).IsNull(0))
+}
+
func TestEntriesSchema(t *testing.T) {
sc := EntriesSchema(&iceberg.StructType{})
require.Equal(t, []string{"status", "snapshot_id", "sequence_number", "file_sequence_number", "data_file"}, testFieldNames(sc))
diff --git a/table/inspect_partitions.go b/table/inspect_partitions.go
new file mode 100644
index 000000000..730cca2f4
--- /dev/null
+++ b/table/inspect_partitions.go
@@ -0,0 +1,347 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "sort"
+ "strings"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/scalar"
+ "github.com/apache/iceberg-go"
+)
+
+type inspectPartitionAggregate struct {
+ specID int32
+ partition map[int]any
+ dataRecordCount int64
+ dataFileCount int32
+ dataFileSize int64
+ positionDeleteCount int64
+ positionDeleteFiles int32
+ equalityDeleteCount int64
+ equalityDeleteFiles int32
+ lastUpdatedAt *int64
+ lastUpdatedSnapshot *int64
+ partitionRecord partitionRecord
+ orderingKey string
+}
+
+type inspectPartitionAggregateTree struct {
+ children map[any]*inspectPartitionAggregateTree
+ aggregate *inspectPartitionAggregate
+}
+
+func newInspectPartitionAggregateTree() *inspectPartitionAggregateTree {
+ return &inspectPartitionAggregateTree{
+ children: make(map[any]*inspectPartitionAggregateTree),
+ }
+}
+
+func (t *inspectPartitionAggregateTree) lookup(record partitionRecord) *inspectPartitionAggregate {
+ node := t
+ for _, value := range record {
+ child, ok := node.children[comparablePartitionKey(value)]
+ if !ok {
+ return nil
+ }
+ node = child
+ }
+
+ return node.aggregate
+}
+
+func (t *inspectPartitionAggregateTree) insert(record partitionRecord, aggregate *inspectPartitionAggregate) {
+ node := t
+ for _, value := range record {
+ key := comparablePartitionKey(value)
+ child, ok := node.children[key]
+ if !ok {
+ child = newInspectPartitionAggregateTree()
+ node.children[key] = child
+ }
+ node = child
+ }
+ node.aggregate = aggregate
+}
+
+// Partitions returns one row per live partition, aggregating data and delete
+// files from the current snapshot. Files from evolved specs are coerced into
+// the table-wide partition type before grouping.
+func (i InspectTable) Partitions(ctx context.Context) (array.RecordReader, error) {
+ partitionType, err := inspectPartitionType(i.tbl.metadata)
+ if err != nil {
+ return nil, fmt.Errorf("inspect partitions: %w", err)
+ }
+ schema := PartitionsSchema(partitionType)
+ arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect partitions: build arrow schema: %w", err)
+ }
+
+ aggregates, err := i.partitionAggregates(ctx, partitionType)
+ if err != nil {
+ return nil, fmt.Errorf("inspect partitions: %w", err)
+ }
+
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
+ defer bldr.Release()
+ for _, aggregate := range aggregates {
+ if err := appendPartitionAggregate(bldr, partitionType, aggregate); err != nil {
+ return nil, fmt.Errorf("inspect partitions: append partition: %w", err)
+ }
+ }
+
+ rr, err := singleBatchReader(arrowSchema, bldr)
+ if err != nil {
+ return nil, fmt.Errorf("inspect partitions: %w", err)
+ }
+
+ return rr, nil
+}
+
+func (i InspectTable) partitionAggregates(ctx context.Context, partitionType *iceberg.StructType) ([]inspectPartitionAggregate, error) {
+ snapshot := i.tbl.metadata.CurrentSnapshot()
+ if snapshot == nil {
+ return nil, nil
+ }
+ if i.tbl.fsF == nil {
+ return nil, errors.New("table file IO is not configured")
+ }
+ fs, err := i.tbl.fsF(ctx)
+ if err != nil {
+ return nil, err
+ }
+ manifests, err := snapshot.Manifests(fs)
+ if err != nil {
+ return nil, err
+ }
+
+ snapshotTimes := make(map[int64]int64, len(i.tbl.metadata.Snapshots()))
+ for _, snapshot := range i.tbl.metadata.Snapshots() {
+ snapshotTimes[snapshot.SnapshotID] = snapshot.TimestampMs
+ }
+
+ aggregates := make([]*inspectPartitionAggregate, 0)
+ aggregateTree := newInspectPartitionAggregateTree()
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ for entry, err := range manifest.Entries(fs, true) {
+ if err != nil {
+ return nil, err
+ }
+ file := entry.DataFile()
+ partition := inspectCoercePartition(file.Partition(), partitionType)
+ record := newPartitionRecord(partition, partitionType)
+ aggregate := aggregateTree.lookup(record)
+ if aggregate == nil {
+ aggregate = &inspectPartitionAggregate{
+ partition: cloneInspectPartition(partition),
+ partitionRecord: record,
+ orderingKey: inspectPartitionKey(partition),
+ }
+ aggregates = append(aggregates, aggregate)
+ aggregateTree.insert(record, aggregate)
+ }
+
+ switch file.ContentType() {
+ case iceberg.EntryContentData:
+ aggregate.dataRecordCount += file.Count()
+ aggregate.dataFileCount++
+ aggregate.dataFileSize += file.FileSizeBytes()
+ case iceberg.EntryContentPosDeletes:
+ aggregate.positionDeleteCount += file.Count()
+ aggregate.positionDeleteFiles++
+ case iceberg.EntryContentEqDeletes:
+ aggregate.equalityDeleteCount += file.Count()
+ aggregate.equalityDeleteFiles++
+ }
+
+ if file.ContentType() == iceberg.EntryContentData ||
+ file.ContentType() == iceberg.EntryContentPosDeletes ||
+ file.ContentType() == iceberg.EntryContentEqDeletes {
+ i.updatePartitionTimestamp(aggregate, snapshotTimes, entry.SnapshotID(), file.SpecID())
+ }
+ }
+ }
+
+ result := make([]inspectPartitionAggregate, 0, len(aggregates))
+ for _, aggregate := range aggregates {
+ result = append(result, *aggregate)
+ }
+ sort.Slice(result, func(left, right int) bool { return result[left].orderingKey < result[right].orderingKey })
+
+ return result, nil
+}
+
+func (i InspectTable) updatePartitionTimestamp(
+ aggregate *inspectPartitionAggregate,
+ snapshotTimes map[int64]int64,
+ snapshotID int64,
+ specID int32,
+) {
+ if snapshotID < 0 {
+ return
+ }
+ timestampMs, ok := snapshotTimes[snapshotID]
+ if !ok {
+ return
+ }
+ timestamp := timestampMs * 1000
+ if aggregate.lastUpdatedAt != nil && timestamp <= *aggregate.lastUpdatedAt {
+ return
+ }
+ aggregate.lastUpdatedAt = ×tamp
+ id := snapshotID
+ aggregate.lastUpdatedSnapshot = &id
+ aggregate.specID = specID
+}
+
+func inspectCoercePartition(values map[int]any, partitionType *iceberg.StructType) map[int]any {
+ if partitionType == nil || len(partitionType.FieldList) == 0 {
+ return nil
+ }
+
+ coerced := make(map[int]any, len(partitionType.FieldList))
+ for _, field := range partitionType.FieldList {
+ if value, ok := values[field.ID]; ok {
+ coerced[field.ID] = value
+ }
+ }
+
+ return coerced
+}
+
+func inspectPartitionKey(partition map[int]any) string {
+ ids := make([]int, 0, len(partition))
+ for id := range partition {
+ ids = append(ids, id)
+ }
+ sort.Ints(ids)
+ var key strings.Builder
+ for _, id := range ids {
+ fmt.Fprintf(&key, "%d:%T:%v;", id, partition[id], partition[id])
+ }
+
+ return key.String()
+}
+
+func cloneInspectPartition(partition map[int]any) map[int]any {
+ if partition == nil {
+ return nil
+ }
+ returnMap := make(map[int]any, len(partition))
+ for id, value := range partition {
+ returnMap[id] = value
+ }
+
+ return returnMap
+}
+
+// PartitionsSchema returns the partition metadata-table schema. For an
+// unpartitioned table the partition and spec_id columns are omitted, matching
+// Java's PartitionsTable behavior.
+func PartitionsSchema(partitionType *iceberg.StructType) *iceberg.Schema {
+ if partitionType == nil || len(partitionType.FieldList) == 0 {
+ return iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 2, Name: "record_count", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 3, Name: "file_count", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 11, Name: "total_data_file_size_in_bytes", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 5, Name: "position_delete_record_count", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 6, Name: "position_delete_file_count", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 7, Name: "equality_delete_record_count", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 8, Name: "equality_delete_file_count", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 9, Name: "last_updated_at", Type: iceberg.PrimitiveTypes.TimestampTz, Required: false},
+ iceberg.NestedField{ID: 10, Name: "last_updated_snapshot_id", Type: iceberg.PrimitiveTypes.Int64, Required: false},
+ )
+ }
+
+ return iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "partition", Type: partitionType, Required: true},
+ iceberg.NestedField{ID: 4, Name: "spec_id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "record_count", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 3, Name: "file_count", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 11, Name: "total_data_file_size_in_bytes", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 5, Name: "position_delete_record_count", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 6, Name: "position_delete_file_count", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 7, Name: "equality_delete_record_count", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 8, Name: "equality_delete_file_count", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 9, Name: "last_updated_at", Type: iceberg.PrimitiveTypes.TimestampTz, Required: false},
+ iceberg.NestedField{ID: 10, Name: "last_updated_snapshot_id", Type: iceberg.PrimitiveTypes.Int64, Required: false},
+ )
+}
+
+func appendPartitionAggregate(bldr *array.RecordBuilder, partitionType *iceberg.StructType, aggregate inspectPartitionAggregate) error {
+ idx := 0
+ if partitionType != nil && len(partitionType.FieldList) > 0 {
+ partition := bldr.Field(idx).(*array.StructBuilder)
+ arrowType := partition.Type().(*arrow.StructType)
+ partition.Append(true)
+ for fieldIndex, field := range partitionType.FieldList {
+ value := aggregate.partition[field.ID]
+ if value == nil {
+ partition.FieldBuilder(fieldIndex).AppendNull()
+
+ continue
+ }
+ valueScalar, err := inspectValueScalar(value, field.Type, arrowType.Field(fieldIndex).Type)
+ if err != nil {
+ return fmt.Errorf("partition field %q: %w", field.Name, err)
+ }
+ if err := scalar.Append(partition.FieldBuilder(fieldIndex), valueScalar); err != nil {
+ return err
+ }
+ }
+ idx++
+ bldr.Field(idx).(*array.Int32Builder).Append(aggregate.specID)
+ idx++
+ }
+ bldr.Field(idx).(*array.Int64Builder).Append(aggregate.dataRecordCount)
+ idx++
+ bldr.Field(idx).(*array.Int32Builder).Append(aggregate.dataFileCount)
+ idx++
+ bldr.Field(idx).(*array.Int64Builder).Append(aggregate.dataFileSize)
+ idx++
+ bldr.Field(idx).(*array.Int64Builder).Append(aggregate.positionDeleteCount)
+ idx++
+ bldr.Field(idx).(*array.Int32Builder).Append(aggregate.positionDeleteFiles)
+ idx++
+ bldr.Field(idx).(*array.Int64Builder).Append(aggregate.equalityDeleteCount)
+ idx++
+ bldr.Field(idx).(*array.Int32Builder).Append(aggregate.equalityDeleteFiles)
+ idx++
+ if aggregate.lastUpdatedAt == nil {
+ bldr.Field(idx).(*array.TimestampBuilder).AppendNull()
+ } else {
+ bldr.Field(idx).(*array.TimestampBuilder).Append(arrow.Timestamp(*aggregate.lastUpdatedAt))
+ }
+ idx++
+ if aggregate.lastUpdatedSnapshot == nil {
+ bldr.Field(idx).(*array.Int64Builder).AppendNull()
+ } else {
+ bldr.Field(idx).(*array.Int64Builder).Append(*aggregate.lastUpdatedSnapshot)
+ }
+
+ return nil
+}