laskoviymishka commented on code in PR #1698: URL: https://github.com/apache/iceberg-go/pull/1698#discussion_r3739120777
########## table/inspect_files.go: ########## @@ -0,0 +1,533 @@ +// 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" + "iter" + "slices" + "sort" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/arrow-go/v18/arrow/scalar" + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +// DataFiles returns the live data files in the current snapshot. Deleted +// manifest entries are omitted, matching the data_files metadata table. +func (i InspectTable) DataFiles(ctx context.Context) (array.RecordReader, error) { + partitionType, err := inspectPartitionType(i.tbl.metadata) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + schema := DataFilesSchema(partitionType) + arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false) + if err != nil { + return nil, fmt.Errorf("inspect data files: build arrow schema: %w", err) + } + + rr, err := i.manifestEntryReader(ctx, arrowSchema, true, + func(manifest iceberg.ManifestFile) bool { + return manifest.ManifestContent() == iceberg.ManifestContentData + }, + func(bldr *array.RecordBuilder, entry iceberg.ManifestEntry) error { + return appendContentFileRecord(bldr, partitionType, entry.DataFile()) + }) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + + return rr, nil +} + +const inspectRecordBatchSize = 4096 + +// manifestEntryReader streams manifest entries into bounded Arrow record +// batches. It keeps only the current batch and manifest decoder state in +// memory instead of materializing every entry before returning a reader. +func (i InspectTable) manifestEntryReader( + ctx context.Context, + arrowSchema *arrow.Schema, + discardDeleted bool, + includeManifest func(iceberg.ManifestFile) bool, + appendEntry func(*array.RecordBuilder, iceberg.ManifestEntry) error, +) (array.RecordReader, error) { + snapshot := i.tbl.metadata.CurrentSnapshot() + if snapshot == nil { + return array.ReaderFromIter(arrowSchema, emptyInspectRecordBatch(i.alloc, arrowSchema)), 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 + } + + return array.ReaderFromIter(arrowSchema, func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(i.alloc, arrowSchema) + defer bldr.Release() + + rows := 0 + emitted := false + emit := func() bool { + if rows == 0 { + return true + } + + batch := bldr.NewRecordBatch() + rows = 0 + emitted = true + if yield(batch, nil) { + return true + } + + batch.Release() + + return false + } + emitEmpty := func() { + batch := bldr.NewRecordBatch() + emitted = true + if !yield(batch, nil) { + batch.Release() + } + } + yieldError := func(err error) { + _ = yield(nil, err) + } + + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if includeManifest != nil && !includeManifest(manifest) { + continue + } + + for entry, err := range manifest.Entries(fs, discardDeleted) { + if err != nil { + yieldError(fmt.Errorf("read manifest %s: %w", manifest.FilePath(), err)) + + return + } + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if err := appendEntry(bldr, entry); err != nil { + yieldError(fmt.Errorf("append manifest entry from %s: %w", manifest.FilePath(), err)) + + return + } + rows++ + if rows == inspectRecordBatchSize && !emit() { + return + } + } + } + + if rows > 0 { + _ = emit() + } else if !emitted { + emitEmpty() + } + }), nil +} + +func emptyInspectRecordBatch(alloc memory.Allocator, schema *arrow.Schema) iter.Seq2[arrow.RecordBatch, error] { + return func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(alloc, schema) + defer bldr.Release() + + batch := bldr.NewRecordBatch() + if !yield(batch, nil) { + batch.Release() + } + } +} + +// inspectPartitionType returns the table-wide partition type. It contains the +// union of partition fields from every spec, which lets metadata tables +// represent live files written before partition evolution. +func inspectPartitionType(metadata Metadata) (*iceberg.StructType, error) { + currentSchema := metadata.CurrentSchema() + specs := metadata.PartitionSpecs() + sort.Slice(specs, func(left, right int) bool { + return specs[left].ID() > specs[right].ID() + }) + + selected := make(map[int]iceberg.PartitionField) + fieldsByID := make(map[int]iceberg.NestedField) + for _, spec := range specs { + partitionType := spec.PartitionType(currentSchema) + for idx, field := range spec.Fields() { + active := true + for _, sourceID := range field.SourceIDs { + if _, ok := currentSchema.FindTypeByID(sourceID); !ok { + active = false + + break + } + } + if !active || idx >= len(partitionType.FieldList) { + continue + } + + if previous, exists := selected[field.FieldID]; exists { + if !slices.Equal(previous.SourceIDs, field.SourceIDs) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible source IDs %v and %v", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.SourceIDs, field.SourceIDs) + } + + previousVoid := isInspectVoidTransform(previous.Transform) + fieldVoid := isInspectVoidTransform(field.Transform) + if previousVoid || fieldVoid { + if previousVoid && !fieldVoid { + selected[field.FieldID] = field + old := fieldsByID[field.FieldID] + old.Type = partitionType.FieldList[idx].Type + fieldsByID[field.FieldID] = old + } + + continue + } + + if !previous.Transform.Equals(field.Transform) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible transforms %q and %q", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.Transform, field.Transform) + } + + continue + } + + selected[field.FieldID] = field + fieldsByID[field.FieldID] = iceberg.NestedField{ + ID: field.FieldID, + Name: field.Name, + Type: partitionType.FieldList[idx].Type, + Required: false, + } + } + } + + fields := make([]iceberg.NestedField, 0, len(fieldsByID)) + for _, field := range fieldsByID { + fields = append(fields, field) + } + sort.Slice(fields, func(left, right int) bool { return fields[left].ID < fields[right].ID }) + + return &iceberg.StructType{FieldList: fields}, nil +} + +func isInspectVoidTransform(transform iceberg.Transform) bool { + switch transform.(type) { + case iceberg.VoidTransform, *iceberg.VoidTransform: + return true + default: + return false + } +} + +// DataFilesSchema returns the common content-file schema used by the data_files +// and delete_files metadata tables. The partition field is omitted for an +// unpartitioned table, as required by the Iceberg metadata-table spec. +func DataFilesSchema(partitionType *iceberg.StructType) *iceberg.Schema { + return iceberg.NewSchema(0, inspectContentFileFields(partitionType)...) +} + +func inspectContentFileType(partitionType *iceberg.StructType) *iceberg.StructType { + return &iceberg.StructType{FieldList: inspectContentFileFields(partitionType)} +} + +func inspectContentFileFields(partitionType *iceberg.StructType) []iceberg.NestedField { + fields := []iceberg.NestedField{ + {ID: 134, Name: "content", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + {ID: 100, Name: "file_path", Type: iceberg.PrimitiveTypes.String, Required: true}, + {ID: 101, Name: "file_format", Type: iceberg.PrimitiveTypes.String, Required: true}, + {ID: 141, Name: "spec_id", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + } + if partitionType != nil && len(partitionType.FieldList) > 0 { + fields = append(fields, iceberg.NestedField{ID: 102, Name: "partition", Type: partitionType, Required: true}) + } + fields = append(fields, + iceberg.NestedField{ID: 103, Name: "record_count", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 104, Name: "file_size_in_bytes", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 108, Name: "column_sizes", Type: inspectInt64MapType(117, 118), Required: false}, + iceberg.NestedField{ID: 109, Name: "value_counts", Type: inspectInt64MapType(119, 120), Required: false}, + iceberg.NestedField{ID: 110, Name: "null_value_counts", Type: inspectInt64MapType(121, 122), Required: false}, Review Comment: `distinct_value_counts` (field 111, `map<int, long>` with key 112 / value 113) is missing from the schema and the append loop. `DataFile` already exposes `DistinctValueCounts()` and it's populated, so the data's right there, a query for it against an iceberg-go table just won't find the column, and that gap propagates when this helper backs delete_files/entries. Spec order puts it between `null_value_counts` (110) and `nan_value_counts` (137), so I'd add the field here and the matching `appendInspectInt64Map(...file.DistinctValueCounts())` at the same position. ########## table/inspect_files.go: ########## @@ -0,0 +1,533 @@ +// 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" + "iter" + "slices" + "sort" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/arrow-go/v18/arrow/scalar" + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +// DataFiles returns the live data files in the current snapshot. Deleted +// manifest entries are omitted, matching the data_files metadata table. +func (i InspectTable) DataFiles(ctx context.Context) (array.RecordReader, error) { + partitionType, err := inspectPartitionType(i.tbl.metadata) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + schema := DataFilesSchema(partitionType) + arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false) + if err != nil { + return nil, fmt.Errorf("inspect data files: build arrow schema: %w", err) + } + + rr, err := i.manifestEntryReader(ctx, arrowSchema, true, + func(manifest iceberg.ManifestFile) bool { + return manifest.ManifestContent() == iceberg.ManifestContentData + }, + func(bldr *array.RecordBuilder, entry iceberg.ManifestEntry) error { + return appendContentFileRecord(bldr, partitionType, entry.DataFile()) + }) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + + return rr, nil +} + +const inspectRecordBatchSize = 4096 + +// manifestEntryReader streams manifest entries into bounded Arrow record +// batches. It keeps only the current batch and manifest decoder state in +// memory instead of materializing every entry before returning a reader. +func (i InspectTable) manifestEntryReader( + ctx context.Context, + arrowSchema *arrow.Schema, + discardDeleted bool, + includeManifest func(iceberg.ManifestFile) bool, + appendEntry func(*array.RecordBuilder, iceberg.ManifestEntry) error, +) (array.RecordReader, error) { + snapshot := i.tbl.metadata.CurrentSnapshot() + if snapshot == nil { + return array.ReaderFromIter(arrowSchema, emptyInspectRecordBatch(i.alloc, arrowSchema)), 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 + } + + return array.ReaderFromIter(arrowSchema, func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(i.alloc, arrowSchema) + defer bldr.Release() + + rows := 0 + emitted := false + emit := func() bool { + if rows == 0 { + return true + } + + batch := bldr.NewRecordBatch() + rows = 0 + emitted = true + if yield(batch, nil) { + return true + } + + batch.Release() + + return false + } + emitEmpty := func() { + batch := bldr.NewRecordBatch() + emitted = true + if !yield(batch, nil) { + batch.Release() + } + } + yieldError := func(err error) { + _ = yield(nil, err) + } + + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if includeManifest != nil && !includeManifest(manifest) { + continue + } + + for entry, err := range manifest.Entries(fs, discardDeleted) { + if err != nil { + yieldError(fmt.Errorf("read manifest %s: %w", manifest.FilePath(), err)) + + return + } + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if err := appendEntry(bldr, entry); err != nil { + yieldError(fmt.Errorf("append manifest entry from %s: %w", manifest.FilePath(), err)) + + return + } + rows++ + if rows == inspectRecordBatchSize && !emit() { + return + } + } + } + + if rows > 0 { + _ = emit() + } else if !emitted { + emitEmpty() + } + }), nil +} + +func emptyInspectRecordBatch(alloc memory.Allocator, schema *arrow.Schema) iter.Seq2[arrow.RecordBatch, error] { + return func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(alloc, schema) + defer bldr.Release() + + batch := bldr.NewRecordBatch() + if !yield(batch, nil) { + batch.Release() + } + } +} + +// inspectPartitionType returns the table-wide partition type. It contains the +// union of partition fields from every spec, which lets metadata tables +// represent live files written before partition evolution. +func inspectPartitionType(metadata Metadata) (*iceberg.StructType, error) { + currentSchema := metadata.CurrentSchema() + specs := metadata.PartitionSpecs() + sort.Slice(specs, func(left, right int) bool { + return specs[left].ID() > specs[right].ID() + }) + + selected := make(map[int]iceberg.PartitionField) + fieldsByID := make(map[int]iceberg.NestedField) + for _, spec := range specs { + partitionType := spec.PartitionType(currentSchema) + for idx, field := range spec.Fields() { + active := true + for _, sourceID := range field.SourceIDs { + if _, ok := currentSchema.FindTypeByID(sourceID); !ok { + active = false + + break + } + } + if !active || idx >= len(partitionType.FieldList) { + continue + } + + if previous, exists := selected[field.FieldID]; exists { + if !slices.Equal(previous.SourceIDs, field.SourceIDs) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible source IDs %v and %v", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.SourceIDs, field.SourceIDs) + } + + previousVoid := isInspectVoidTransform(previous.Transform) + fieldVoid := isInspectVoidTransform(field.Transform) + if previousVoid || fieldVoid { + if previousVoid && !fieldVoid { + selected[field.FieldID] = field + old := fieldsByID[field.FieldID] + old.Type = partitionType.FieldList[idx].Type + fieldsByID[field.FieldID] = old + } + + continue + } + + if !previous.Transform.Equals(field.Transform) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible transforms %q and %q", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.Transform, field.Transform) + } + + continue + } + + selected[field.FieldID] = field + fieldsByID[field.FieldID] = iceberg.NestedField{ + ID: field.FieldID, + Name: field.Name, + Type: partitionType.FieldList[idx].Type, + Required: false, + } + } + } + + fields := make([]iceberg.NestedField, 0, len(fieldsByID)) + for _, field := range fieldsByID { + fields = append(fields, field) + } + sort.Slice(fields, func(left, right int) bool { return fields[left].ID < fields[right].ID }) + + return &iceberg.StructType{FieldList: fields}, nil +} + +func isInspectVoidTransform(transform iceberg.Transform) bool { + switch transform.(type) { + case iceberg.VoidTransform, *iceberg.VoidTransform: Review Comment: `VoidTransform` is always returned by value (`transforms.go`, and every test constructs it that way), so the `*iceberg.VoidTransform` arm never matches. Not harmful today, but if a future parser ever returns the pointer form it'd slip past void-detection and trip the "incompatible transforms" error instead. I'd drop the pointer case unless there's a path that actually produces it. ########## table/inspect_files.go: ########## @@ -0,0 +1,533 @@ +// 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" + "iter" + "slices" + "sort" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/arrow-go/v18/arrow/scalar" + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +// DataFiles returns the live data files in the current snapshot. Deleted +// manifest entries are omitted, matching the data_files metadata table. +func (i InspectTable) DataFiles(ctx context.Context) (array.RecordReader, error) { + partitionType, err := inspectPartitionType(i.tbl.metadata) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + schema := DataFilesSchema(partitionType) + arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false) + if err != nil { + return nil, fmt.Errorf("inspect data files: build arrow schema: %w", err) + } + + rr, err := i.manifestEntryReader(ctx, arrowSchema, true, + func(manifest iceberg.ManifestFile) bool { + return manifest.ManifestContent() == iceberg.ManifestContentData + }, + func(bldr *array.RecordBuilder, entry iceberg.ManifestEntry) error { + return appendContentFileRecord(bldr, partitionType, entry.DataFile()) + }) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + + return rr, nil +} + +const inspectRecordBatchSize = 4096 + +// manifestEntryReader streams manifest entries into bounded Arrow record +// batches. It keeps only the current batch and manifest decoder state in +// memory instead of materializing every entry before returning a reader. +func (i InspectTable) manifestEntryReader( + ctx context.Context, + arrowSchema *arrow.Schema, + discardDeleted bool, + includeManifest func(iceberg.ManifestFile) bool, + appendEntry func(*array.RecordBuilder, iceberg.ManifestEntry) error, +) (array.RecordReader, error) { + snapshot := i.tbl.metadata.CurrentSnapshot() + if snapshot == nil { + return array.ReaderFromIter(arrowSchema, emptyInspectRecordBatch(i.alloc, arrowSchema)), 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 + } + + return array.ReaderFromIter(arrowSchema, func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(i.alloc, arrowSchema) + defer bldr.Release() + + rows := 0 + emitted := false + emit := func() bool { + if rows == 0 { + return true + } + + batch := bldr.NewRecordBatch() + rows = 0 + emitted = true + if yield(batch, nil) { + return true + } + + batch.Release() + + return false + } + emitEmpty := func() { + batch := bldr.NewRecordBatch() + emitted = true + if !yield(batch, nil) { + batch.Release() + } + } + yieldError := func(err error) { + _ = yield(nil, err) + } + + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if includeManifest != nil && !includeManifest(manifest) { + continue + } + + for entry, err := range manifest.Entries(fs, discardDeleted) { + if err != nil { + yieldError(fmt.Errorf("read manifest %s: %w", manifest.FilePath(), err)) + + return + } + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if err := appendEntry(bldr, entry); err != nil { + yieldError(fmt.Errorf("append manifest entry from %s: %w", manifest.FilePath(), err)) + + return + } + rows++ + if rows == inspectRecordBatchSize && !emit() { + return + } + } + } + + if rows > 0 { + _ = emit() + } else if !emitted { + emitEmpty() + } + }), nil +} + +func emptyInspectRecordBatch(alloc memory.Allocator, schema *arrow.Schema) iter.Seq2[arrow.RecordBatch, error] { + return func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(alloc, schema) + defer bldr.Release() + + batch := bldr.NewRecordBatch() + if !yield(batch, nil) { + batch.Release() + } + } +} + +// inspectPartitionType returns the table-wide partition type. It contains the +// union of partition fields from every spec, which lets metadata tables +// represent live files written before partition evolution. +func inspectPartitionType(metadata Metadata) (*iceberg.StructType, error) { + currentSchema := metadata.CurrentSchema() + specs := metadata.PartitionSpecs() + sort.Slice(specs, func(left, right int) bool { + return specs[left].ID() > specs[right].ID() + }) + + selected := make(map[int]iceberg.PartitionField) + fieldsByID := make(map[int]iceberg.NestedField) + for _, spec := range specs { + partitionType := spec.PartitionType(currentSchema) + for idx, field := range spec.Fields() { + active := true + for _, sourceID := range field.SourceIDs { + if _, ok := currentSchema.FindTypeByID(sourceID); !ok { + active = false + + break + } + } + if !active || idx >= len(partitionType.FieldList) { + continue + } + + if previous, exists := selected[field.FieldID]; exists { + if !slices.Equal(previous.SourceIDs, field.SourceIDs) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible source IDs %v and %v", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.SourceIDs, field.SourceIDs) + } + + previousVoid := isInspectVoidTransform(previous.Transform) + fieldVoid := isInspectVoidTransform(field.Transform) + if previousVoid || fieldVoid { + if previousVoid && !fieldVoid { + selected[field.FieldID] = field + old := fieldsByID[field.FieldID] + old.Type = partitionType.FieldList[idx].Type + fieldsByID[field.FieldID] = old + } + + continue + } + + if !previous.Transform.Equals(field.Transform) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible transforms %q and %q", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.Transform, field.Transform) + } + + continue + } + + selected[field.FieldID] = field + fieldsByID[field.FieldID] = iceberg.NestedField{ + ID: field.FieldID, + Name: field.Name, + Type: partitionType.FieldList[idx].Type, + Required: false, + } + } + } + + fields := make([]iceberg.NestedField, 0, len(fieldsByID)) + for _, field := range fieldsByID { + fields = append(fields, field) + } + sort.Slice(fields, func(left, right int) bool { return fields[left].ID < fields[right].ID }) + + return &iceberg.StructType{FieldList: fields}, nil +} + +func isInspectVoidTransform(transform iceberg.Transform) bool { + switch transform.(type) { + case iceberg.VoidTransform, *iceberg.VoidTransform: + return true + default: + return false + } +} + +// DataFilesSchema returns the common content-file schema used by the data_files +// and delete_files metadata tables. The partition field is omitted for an +// unpartitioned table, as required by the Iceberg metadata-table spec. +func DataFilesSchema(partitionType *iceberg.StructType) *iceberg.Schema { + return iceberg.NewSchema(0, inspectContentFileFields(partitionType)...) +} + +func inspectContentFileType(partitionType *iceberg.StructType) *iceberg.StructType { + return &iceberg.StructType{FieldList: inspectContentFileFields(partitionType)} +} + +func inspectContentFileFields(partitionType *iceberg.StructType) []iceberg.NestedField { + fields := []iceberg.NestedField{ + {ID: 134, Name: "content", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + {ID: 100, Name: "file_path", Type: iceberg.PrimitiveTypes.String, Required: true}, + {ID: 101, Name: "file_format", Type: iceberg.PrimitiveTypes.String, Required: true}, + {ID: 141, Name: "spec_id", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + } + if partitionType != nil && len(partitionType.FieldList) > 0 { + fields = append(fields, iceberg.NestedField{ID: 102, Name: "partition", Type: partitionType, Required: true}) + } + fields = append(fields, + iceberg.NestedField{ID: 103, Name: "record_count", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 104, Name: "file_size_in_bytes", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 108, Name: "column_sizes", Type: inspectInt64MapType(117, 118), Required: false}, + iceberg.NestedField{ID: 109, Name: "value_counts", Type: inspectInt64MapType(119, 120), Required: false}, + iceberg.NestedField{ID: 110, Name: "null_value_counts", Type: inspectInt64MapType(121, 122), Required: false}, + iceberg.NestedField{ID: 137, Name: "nan_value_counts", Type: inspectInt64MapType(138, 139), Required: false}, + iceberg.NestedField{ID: 125, Name: "lower_bounds", Type: inspectBinaryMapType(126, 127), Required: false}, + iceberg.NestedField{ID: 128, Name: "upper_bounds", Type: inspectBinaryMapType(129, 130), Required: false}, + iceberg.NestedField{ID: 131, Name: "key_metadata", Type: iceberg.PrimitiveTypes.Binary, Required: false}, + iceberg.NestedField{ID: 132, Name: "split_offsets", Type: &iceberg.ListType{ElementID: 133, Element: iceberg.PrimitiveTypes.Int64, ElementRequired: true}, Required: false}, + iceberg.NestedField{ID: 135, Name: "equality_ids", Type: &iceberg.ListType{ElementID: 136, Element: iceberg.PrimitiveTypes.Int32, ElementRequired: true}, Required: false}, + iceberg.NestedField{ID: 140, Name: "sort_order_id", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + iceberg.NestedField{ID: 142, Name: "first_row_id", Type: iceberg.PrimitiveTypes.Int64, Required: false}, + iceberg.NestedField{ID: 143, Name: "referenced_data_file", Type: iceberg.PrimitiveTypes.String, Required: false}, + iceberg.NestedField{ID: 144, Name: "content_offset", Type: iceberg.PrimitiveTypes.Int64, Required: false}, + iceberg.NestedField{ID: 145, Name: "content_size_in_bytes", Type: iceberg.PrimitiveTypes.Int64, Required: false}, + ) + + return fields +} + +func inspectInt64MapType(keyID, valueID int) *iceberg.MapType { + return &iceberg.MapType{KeyID: keyID, KeyType: iceberg.PrimitiveTypes.Int32, ValueID: valueID, ValueType: iceberg.PrimitiveTypes.Int64, ValueRequired: true} +} + +func inspectBinaryMapType(keyID, valueID int) *iceberg.MapType { + return &iceberg.MapType{KeyID: keyID, KeyType: iceberg.PrimitiveTypes.Int32, ValueID: valueID, ValueType: iceberg.PrimitiveTypes.Binary, ValueRequired: true} +} + +func appendContentFileRecord(bldr *array.RecordBuilder, partitionType *iceberg.StructType, file iceberg.DataFile) error { + return appendContentFileFields(bldr.Field, partitionType, file) +} + +func appendContentFile(builder *array.StructBuilder, partitionType *iceberg.StructType, file iceberg.DataFile) error { + builder.Append(true) + + return appendContentFileFields(builder.FieldBuilder, partitionType, file) +} + +func appendContentFileFields(fieldBuilder func(int) array.Builder, partitionType *iceberg.StructType, file iceberg.DataFile) error { + idx := 0 Review Comment: The schema in `inspectContentFileFields` and this builder loop have to stay in perfect lock-step by hand, every `idx++` here mirroring one field there, including the conditional bump for the optional partition column. There's no guard, so inserting or reordering a single field silently shifts everything downstream: wrong order writes mismatched values, wrong type panics at the assertion, and nothing catches it until runtime. `inspect.go` sidesteps this by naming each builder explicitly. Since you've built this as shared scaffolding for delete_files/entries too, the drift surface only grows. I'd either name the builders like `inspect.go` or look them up by field ID, and failing that at least assert `idx == bldr.NumField()` at the end so a mismatch fails loudly. wdyt? ########## table/inspect_files.go: ########## @@ -0,0 +1,533 @@ +// 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" + "iter" + "slices" + "sort" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/arrow-go/v18/arrow/scalar" + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +// DataFiles returns the live data files in the current snapshot. Deleted +// manifest entries are omitted, matching the data_files metadata table. +func (i InspectTable) DataFiles(ctx context.Context) (array.RecordReader, error) { + partitionType, err := inspectPartitionType(i.tbl.metadata) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + schema := DataFilesSchema(partitionType) + arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false) + if err != nil { + return nil, fmt.Errorf("inspect data files: build arrow schema: %w", err) + } + + rr, err := i.manifestEntryReader(ctx, arrowSchema, true, + func(manifest iceberg.ManifestFile) bool { + return manifest.ManifestContent() == iceberg.ManifestContentData + }, + func(bldr *array.RecordBuilder, entry iceberg.ManifestEntry) error { + return appendContentFileRecord(bldr, partitionType, entry.DataFile()) + }) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + + return rr, nil +} + +const inspectRecordBatchSize = 4096 + +// manifestEntryReader streams manifest entries into bounded Arrow record +// batches. It keeps only the current batch and manifest decoder state in +// memory instead of materializing every entry before returning a reader. +func (i InspectTable) manifestEntryReader( + ctx context.Context, + arrowSchema *arrow.Schema, + discardDeleted bool, + includeManifest func(iceberg.ManifestFile) bool, + appendEntry func(*array.RecordBuilder, iceberg.ManifestEntry) error, +) (array.RecordReader, error) { + snapshot := i.tbl.metadata.CurrentSnapshot() + if snapshot == nil { + return array.ReaderFromIter(arrowSchema, emptyInspectRecordBatch(i.alloc, arrowSchema)), 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 + } + + return array.ReaderFromIter(arrowSchema, func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(i.alloc, arrowSchema) + defer bldr.Release() + + rows := 0 + emitted := false + emit := func() bool { + if rows == 0 { + return true + } + + batch := bldr.NewRecordBatch() + rows = 0 + emitted = true + if yield(batch, nil) { + return true + } + + batch.Release() + + return false + } + emitEmpty := func() { + batch := bldr.NewRecordBatch() + emitted = true + if !yield(batch, nil) { + batch.Release() + } + } + yieldError := func(err error) { + _ = yield(nil, err) + } + + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if includeManifest != nil && !includeManifest(manifest) { + continue + } + + for entry, err := range manifest.Entries(fs, discardDeleted) { + if err != nil { + yieldError(fmt.Errorf("read manifest %s: %w", manifest.FilePath(), err)) + + return + } + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if err := appendEntry(bldr, entry); err != nil { + yieldError(fmt.Errorf("append manifest entry from %s: %w", manifest.FilePath(), err)) + + return + } + rows++ + if rows == inspectRecordBatchSize && !emit() { + return + } + } + } + + if rows > 0 { + _ = emit() + } else if !emitted { + emitEmpty() + } + }), nil +} + +func emptyInspectRecordBatch(alloc memory.Allocator, schema *arrow.Schema) iter.Seq2[arrow.RecordBatch, error] { + return func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(alloc, schema) + defer bldr.Release() + + batch := bldr.NewRecordBatch() + if !yield(batch, nil) { + batch.Release() + } + } +} + +// inspectPartitionType returns the table-wide partition type. It contains the +// union of partition fields from every spec, which lets metadata tables +// represent live files written before partition evolution. +func inspectPartitionType(metadata Metadata) (*iceberg.StructType, error) { + currentSchema := metadata.CurrentSchema() + specs := metadata.PartitionSpecs() + sort.Slice(specs, func(left, right int) bool { + return specs[left].ID() > specs[right].ID() + }) + + selected := make(map[int]iceberg.PartitionField) + fieldsByID := make(map[int]iceberg.NestedField) + for _, spec := range specs { + partitionType := spec.PartitionType(currentSchema) + for idx, field := range spec.Fields() { + active := true + for _, sourceID := range field.SourceIDs { + if _, ok := currentSchema.FindTypeByID(sourceID); !ok { + active = false + + break + } + } + if !active || idx >= len(partitionType.FieldList) { + continue + } + + if previous, exists := selected[field.FieldID]; exists { + if !slices.Equal(previous.SourceIDs, field.SourceIDs) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible source IDs %v and %v", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.SourceIDs, field.SourceIDs) + } + + previousVoid := isInspectVoidTransform(previous.Transform) + fieldVoid := isInspectVoidTransform(field.Transform) + if previousVoid || fieldVoid { + if previousVoid && !fieldVoid { + selected[field.FieldID] = field + old := fieldsByID[field.FieldID] + old.Type = partitionType.FieldList[idx].Type + fieldsByID[field.FieldID] = old + } + + continue + } + + if !previous.Transform.Equals(field.Transform) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible transforms %q and %q", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.Transform, field.Transform) + } + + continue + } + + selected[field.FieldID] = field + fieldsByID[field.FieldID] = iceberg.NestedField{ + ID: field.FieldID, + Name: field.Name, + Type: partitionType.FieldList[idx].Type, + Required: false, + } + } + } + + fields := make([]iceberg.NestedField, 0, len(fieldsByID)) + for _, field := range fieldsByID { + fields = append(fields, field) + } + sort.Slice(fields, func(left, right int) bool { return fields[left].ID < fields[right].ID }) + + return &iceberg.StructType{FieldList: fields}, nil +} + +func isInspectVoidTransform(transform iceberg.Transform) bool { + switch transform.(type) { + case iceberg.VoidTransform, *iceberg.VoidTransform: + return true + default: + return false + } +} + +// DataFilesSchema returns the common content-file schema used by the data_files +// and delete_files metadata tables. The partition field is omitted for an +// unpartitioned table, as required by the Iceberg metadata-table spec. +func DataFilesSchema(partitionType *iceberg.StructType) *iceberg.Schema { + return iceberg.NewSchema(0, inspectContentFileFields(partitionType)...) +} + +func inspectContentFileType(partitionType *iceberg.StructType) *iceberg.StructType { + return &iceberg.StructType{FieldList: inspectContentFileFields(partitionType)} +} + +func inspectContentFileFields(partitionType *iceberg.StructType) []iceberg.NestedField { + fields := []iceberg.NestedField{ + {ID: 134, Name: "content", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + {ID: 100, Name: "file_path", Type: iceberg.PrimitiveTypes.String, Required: true}, + {ID: 101, Name: "file_format", Type: iceberg.PrimitiveTypes.String, Required: true}, + {ID: 141, Name: "spec_id", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + } + if partitionType != nil && len(partitionType.FieldList) > 0 { + fields = append(fields, iceberg.NestedField{ID: 102, Name: "partition", Type: partitionType, Required: true}) + } + fields = append(fields, + iceberg.NestedField{ID: 103, Name: "record_count", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 104, Name: "file_size_in_bytes", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 108, Name: "column_sizes", Type: inspectInt64MapType(117, 118), Required: false}, + iceberg.NestedField{ID: 109, Name: "value_counts", Type: inspectInt64MapType(119, 120), Required: false}, + iceberg.NestedField{ID: 110, Name: "null_value_counts", Type: inspectInt64MapType(121, 122), Required: false}, + iceberg.NestedField{ID: 137, Name: "nan_value_counts", Type: inspectInt64MapType(138, 139), Required: false}, + iceberg.NestedField{ID: 125, Name: "lower_bounds", Type: inspectBinaryMapType(126, 127), Required: false}, + iceberg.NestedField{ID: 128, Name: "upper_bounds", Type: inspectBinaryMapType(129, 130), Required: false}, + iceberg.NestedField{ID: 131, Name: "key_metadata", Type: iceberg.PrimitiveTypes.Binary, Required: false}, + iceberg.NestedField{ID: 132, Name: "split_offsets", Type: &iceberg.ListType{ElementID: 133, Element: iceberg.PrimitiveTypes.Int64, ElementRequired: true}, Required: false}, + iceberg.NestedField{ID: 135, Name: "equality_ids", Type: &iceberg.ListType{ElementID: 136, Element: iceberg.PrimitiveTypes.Int32, ElementRequired: true}, Required: false}, + iceberg.NestedField{ID: 140, Name: "sort_order_id", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + iceberg.NestedField{ID: 142, Name: "first_row_id", Type: iceberg.PrimitiveTypes.Int64, Required: false}, + iceberg.NestedField{ID: 143, Name: "referenced_data_file", Type: iceberg.PrimitiveTypes.String, Required: false}, + iceberg.NestedField{ID: 144, Name: "content_offset", Type: iceberg.PrimitiveTypes.Int64, Required: false}, + iceberg.NestedField{ID: 145, Name: "content_size_in_bytes", Type: iceberg.PrimitiveTypes.Int64, Required: false}, + ) + + return fields +} + +func inspectInt64MapType(keyID, valueID int) *iceberg.MapType { + return &iceberg.MapType{KeyID: keyID, KeyType: iceberg.PrimitiveTypes.Int32, ValueID: valueID, ValueType: iceberg.PrimitiveTypes.Int64, ValueRequired: true} +} + +func inspectBinaryMapType(keyID, valueID int) *iceberg.MapType { + return &iceberg.MapType{KeyID: keyID, KeyType: iceberg.PrimitiveTypes.Int32, ValueID: valueID, ValueType: iceberg.PrimitiveTypes.Binary, ValueRequired: true} +} + +func appendContentFileRecord(bldr *array.RecordBuilder, partitionType *iceberg.StructType, file iceberg.DataFile) error { + return appendContentFileFields(bldr.Field, partitionType, file) +} + +func appendContentFile(builder *array.StructBuilder, partitionType *iceberg.StructType, file iceberg.DataFile) error { + builder.Append(true) + + return appendContentFileFields(builder.FieldBuilder, partitionType, file) +} + +func appendContentFileFields(fieldBuilder func(int) array.Builder, partitionType *iceberg.StructType, file iceberg.DataFile) error { + idx := 0 + fieldBuilder(idx).(*array.Int32Builder).Append(int32(file.ContentType())) + idx++ + fieldBuilder(idx).(*array.StringBuilder).Append(file.FilePath()) + idx++ + fieldBuilder(idx).(*array.StringBuilder).Append(string(file.FileFormat())) + idx++ + fieldBuilder(idx).(*array.Int32Builder).Append(file.SpecID()) + idx++ + + if partitionType != nil && len(partitionType.FieldList) > 0 { + partition := fieldBuilder(idx).(*array.StructBuilder) + if err := appendInspectPartition(partition, partitionType, file.Partition()); err != nil { + return err + } + idx++ + } + + fieldBuilder(idx).(*array.Int64Builder).Append(file.Count()) + idx++ + fieldBuilder(idx).(*array.Int64Builder).Append(file.FileSizeBytes()) + idx++ + appendInspectInt64Map(fieldBuilder(idx).(*array.MapBuilder), file.ColumnSizes()) + idx++ + appendInspectInt64Map(fieldBuilder(idx).(*array.MapBuilder), file.ValueCounts()) + idx++ + appendInspectInt64Map(fieldBuilder(idx).(*array.MapBuilder), file.NullValueCounts()) + idx++ + appendInspectInt64Map(fieldBuilder(idx).(*array.MapBuilder), file.NaNValueCounts()) + idx++ + appendInspectBinaryMap(fieldBuilder(idx).(*array.MapBuilder), file.LowerBoundValues()) + idx++ + appendInspectBinaryMap(fieldBuilder(idx).(*array.MapBuilder), file.UpperBoundValues()) + idx++ + appendInspectBytes(fieldBuilder(idx), file.KeyMetadata()) + idx++ + appendInspectInt64List(fieldBuilder(idx).(*array.ListBuilder), file.SplitOffsets()) + idx++ + appendInspectInt32List(fieldBuilder(idx).(*array.ListBuilder), file.EqualityFieldIDs()) + idx++ + appendInspectOptionalInt32(fieldBuilder(idx).(*array.Int32Builder), file.SortOrderID()) + idx++ + appendInspectOptionalInt64(fieldBuilder(idx).(*array.Int64Builder), file.FirstRowID()) + idx++ + appendInspectOptionalString(fieldBuilder(idx).(*array.StringBuilder), file.ReferencedDataFile()) + idx++ + appendInspectOptionalInt64(fieldBuilder(idx).(*array.Int64Builder), file.ContentOffset()) + idx++ + appendInspectOptionalInt64(fieldBuilder(idx).(*array.Int64Builder), file.ContentSizeInBytes()) + + return nil +} + +func appendInspectPartition(builder *array.StructBuilder, partitionType *iceberg.StructType, values map[int]any) error { + arrowType := builder.Type().(*arrow.StructType) + builder.Append(true) + for idx, field := range partitionType.FieldList { + value := values[field.ID] + if value == nil { + builder.FieldBuilder(idx).AppendNull() + + continue + } + sc, err := inspectValueScalar(value, field.Type, arrowType.Field(idx).Type) + if err != nil { + return fmt.Errorf("partition field %q: %w", field.Name, err) + } + if err := scalar.Append(builder.FieldBuilder(idx), sc); err != nil { + return err + } + } + + return nil +} + +func inspectValueScalar(value any, typ iceberg.Type, arrowType arrow.DataType) (scalar.Scalar, error) { + switch typ.(type) { + case iceberg.DateType: + switch value := value.(type) { + case iceberg.Date: + return scalar.NewDate32Scalar(arrow.Date32(value)), nil + case int32: + return scalar.NewDate32Scalar(arrow.Date32(value)), nil + } + case iceberg.TimeType: + if value, ok := value.(iceberg.Time); ok { + return scalar.NewTime64Scalar(arrow.Time64(value), arrowType), nil + } + case iceberg.TimestampType, iceberg.TimestampTzType: + if value, ok := value.(iceberg.Timestamp); ok { + return scalar.NewTimestampScalar(arrow.Timestamp(value), arrowType), nil + } + case iceberg.TimestampNsType, iceberg.TimestampTzNsType: + if value, ok := value.(iceberg.TimestampNano); ok { + return scalar.NewTimestampScalar(arrow.Timestamp(value), arrowType), nil + } + case iceberg.UUIDType: + if value, ok := value.(uuid.UUID); ok { + return scalar.MakeScalarParam(value[:], arrowType) + } + case iceberg.DecimalType: + switch value := value.(type) { + case iceberg.DecimalLiteral: + return scalar.NewDecimal128Scalar(value.Val, arrowType), nil + case iceberg.Decimal: + return scalar.NewDecimal128Scalar(value.Val, arrowType), nil + default: + return nil, fmt.Errorf("unsupported decimal partition value %T", value) + } + } + + return scalar.MakeScalarParam(value, arrowType) Review Comment: The temporal and UUID cases only handle the happy-path assertion. If the type-switch matches but the value isn't the expected concrete type, we fall through to `scalar.MakeScalarParam(value, arrowType)` here instead of erroring. For a `TimestampType` fed an `int64` that quietly builds an `Int64Scalar` rather than a timestamp (silently wrong data), and for `UUIDType` fed a `uuid.UUID` it hits `MakeScalarParam` with a `[16]byte` and panics. `DecimalType` already does the right thing with an explicit error return. I'd give the temporal and UUID cases the same treatment so a bad value is a clear error, not corrupt output or a panic. ########## table/inspect_files.go: ########## @@ -0,0 +1,533 @@ +// 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" + "iter" + "slices" + "sort" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/arrow-go/v18/arrow/scalar" + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +// DataFiles returns the live data files in the current snapshot. Deleted +// manifest entries are omitted, matching the data_files metadata table. +func (i InspectTable) DataFiles(ctx context.Context) (array.RecordReader, error) { + partitionType, err := inspectPartitionType(i.tbl.metadata) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + schema := DataFilesSchema(partitionType) + arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false) + if err != nil { + return nil, fmt.Errorf("inspect data files: build arrow schema: %w", err) + } + + rr, err := i.manifestEntryReader(ctx, arrowSchema, true, + func(manifest iceberg.ManifestFile) bool { + return manifest.ManifestContent() == iceberg.ManifestContentData + }, + func(bldr *array.RecordBuilder, entry iceberg.ManifestEntry) error { + return appendContentFileRecord(bldr, partitionType, entry.DataFile()) + }) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + + return rr, nil +} + +const inspectRecordBatchSize = 4096 + +// manifestEntryReader streams manifest entries into bounded Arrow record +// batches. It keeps only the current batch and manifest decoder state in +// memory instead of materializing every entry before returning a reader. +func (i InspectTable) manifestEntryReader( + ctx context.Context, + arrowSchema *arrow.Schema, + discardDeleted bool, + includeManifest func(iceberg.ManifestFile) bool, + appendEntry func(*array.RecordBuilder, iceberg.ManifestEntry) error, +) (array.RecordReader, error) { + snapshot := i.tbl.metadata.CurrentSnapshot() + if snapshot == nil { + return array.ReaderFromIter(arrowSchema, emptyInspectRecordBatch(i.alloc, arrowSchema)), 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 + } + + return array.ReaderFromIter(arrowSchema, func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(i.alloc, arrowSchema) + defer bldr.Release() + + rows := 0 + emitted := false + emit := func() bool { + if rows == 0 { + return true + } + + batch := bldr.NewRecordBatch() + rows = 0 + emitted = true + if yield(batch, nil) { + return true + } + + batch.Release() + + return false + } + emitEmpty := func() { + batch := bldr.NewRecordBatch() + emitted = true + if !yield(batch, nil) { + batch.Release() + } + } + yieldError := func(err error) { + _ = yield(nil, err) + } + + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if includeManifest != nil && !includeManifest(manifest) { + continue + } + + for entry, err := range manifest.Entries(fs, discardDeleted) { + if err != nil { + yieldError(fmt.Errorf("read manifest %s: %w", manifest.FilePath(), err)) + + return + } + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if err := appendEntry(bldr, entry); err != nil { + yieldError(fmt.Errorf("append manifest entry from %s: %w", manifest.FilePath(), err)) + + return + } + rows++ + if rows == inspectRecordBatchSize && !emit() { + return + } + } + } + + if rows > 0 { + _ = emit() + } else if !emitted { + emitEmpty() + } + }), nil +} + +func emptyInspectRecordBatch(alloc memory.Allocator, schema *arrow.Schema) iter.Seq2[arrow.RecordBatch, error] { + return func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(alloc, schema) + defer bldr.Release() + + batch := bldr.NewRecordBatch() + if !yield(batch, nil) { + batch.Release() + } + } +} + +// inspectPartitionType returns the table-wide partition type. It contains the +// union of partition fields from every spec, which lets metadata tables +// represent live files written before partition evolution. +func inspectPartitionType(metadata Metadata) (*iceberg.StructType, error) { + currentSchema := metadata.CurrentSchema() + specs := metadata.PartitionSpecs() + sort.Slice(specs, func(left, right int) bool { + return specs[left].ID() > specs[right].ID() + }) + + selected := make(map[int]iceberg.PartitionField) + fieldsByID := make(map[int]iceberg.NestedField) + for _, spec := range specs { + partitionType := spec.PartitionType(currentSchema) + for idx, field := range spec.Fields() { + active := true + for _, sourceID := range field.SourceIDs { + if _, ok := currentSchema.FindTypeByID(sourceID); !ok { + active = false + + break + } + } + if !active || idx >= len(partitionType.FieldList) { + continue + } + + if previous, exists := selected[field.FieldID]; exists { + if !slices.Equal(previous.SourceIDs, field.SourceIDs) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible source IDs %v and %v", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.SourceIDs, field.SourceIDs) + } + + previousVoid := isInspectVoidTransform(previous.Transform) + fieldVoid := isInspectVoidTransform(field.Transform) + if previousVoid || fieldVoid { + if previousVoid && !fieldVoid { + selected[field.FieldID] = field + old := fieldsByID[field.FieldID] + old.Type = partitionType.FieldList[idx].Type + fieldsByID[field.FieldID] = old + } + + continue + } + + if !previous.Transform.Equals(field.Transform) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible transforms %q and %q", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.Transform, field.Transform) + } + + continue + } + + selected[field.FieldID] = field + fieldsByID[field.FieldID] = iceberg.NestedField{ + ID: field.FieldID, + Name: field.Name, + Type: partitionType.FieldList[idx].Type, + Required: false, + } + } + } + + fields := make([]iceberg.NestedField, 0, len(fieldsByID)) + for _, field := range fieldsByID { + fields = append(fields, field) + } + sort.Slice(fields, func(left, right int) bool { return fields[left].ID < fields[right].ID }) + + return &iceberg.StructType{FieldList: fields}, nil +} + +func isInspectVoidTransform(transform iceberg.Transform) bool { + switch transform.(type) { + case iceberg.VoidTransform, *iceberg.VoidTransform: + return true + default: + return false + } +} + +// DataFilesSchema returns the common content-file schema used by the data_files +// and delete_files metadata tables. The partition field is omitted for an +// unpartitioned table, as required by the Iceberg metadata-table spec. +func DataFilesSchema(partitionType *iceberg.StructType) *iceberg.Schema { + return iceberg.NewSchema(0, inspectContentFileFields(partitionType)...) +} + +func inspectContentFileType(partitionType *iceberg.StructType) *iceberg.StructType { Review Comment: `inspectContentFileType` (and `appendContentFile` just below) have no callers or tests, they're scaffolding for the future entries table. The deadcode linter will flag `inspectContentFileType`, and `appendContentFile`'s `Append(true)` + `FieldBuilder` convention differs from `appendContentFileRecord`'s `bldr.Field`, so the first real caller could easily wire it up wrong. I'd drop both and reintroduce them with the entries table where they'll have a test, rather than let them bitrot. wdyt? ########## table/inspect_files.go: ########## @@ -0,0 +1,533 @@ +// 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" + "iter" + "slices" + "sort" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/arrow-go/v18/arrow/scalar" + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +// DataFiles returns the live data files in the current snapshot. Deleted +// manifest entries are omitted, matching the data_files metadata table. +func (i InspectTable) DataFiles(ctx context.Context) (array.RecordReader, error) { + partitionType, err := inspectPartitionType(i.tbl.metadata) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + schema := DataFilesSchema(partitionType) + arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false) + if err != nil { + return nil, fmt.Errorf("inspect data files: build arrow schema: %w", err) + } + + rr, err := i.manifestEntryReader(ctx, arrowSchema, true, + func(manifest iceberg.ManifestFile) bool { + return manifest.ManifestContent() == iceberg.ManifestContentData + }, + func(bldr *array.RecordBuilder, entry iceberg.ManifestEntry) error { + return appendContentFileRecord(bldr, partitionType, entry.DataFile()) + }) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + + return rr, nil +} + +const inspectRecordBatchSize = 4096 + +// manifestEntryReader streams manifest entries into bounded Arrow record +// batches. It keeps only the current batch and manifest decoder state in +// memory instead of materializing every entry before returning a reader. +func (i InspectTable) manifestEntryReader( + ctx context.Context, + arrowSchema *arrow.Schema, + discardDeleted bool, + includeManifest func(iceberg.ManifestFile) bool, + appendEntry func(*array.RecordBuilder, iceberg.ManifestEntry) error, +) (array.RecordReader, error) { + snapshot := i.tbl.metadata.CurrentSnapshot() + if snapshot == nil { + return array.ReaderFromIter(arrowSchema, emptyInspectRecordBatch(i.alloc, arrowSchema)), 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 + } + + return array.ReaderFromIter(arrowSchema, func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(i.alloc, arrowSchema) + defer bldr.Release() + + rows := 0 + emitted := false + emit := func() bool { + if rows == 0 { + return true + } + + batch := bldr.NewRecordBatch() + rows = 0 + emitted = true + if yield(batch, nil) { + return true + } + + batch.Release() + + return false + } + emitEmpty := func() { + batch := bldr.NewRecordBatch() + emitted = true + if !yield(batch, nil) { + batch.Release() + } + } + yieldError := func(err error) { + _ = yield(nil, err) + } + + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if includeManifest != nil && !includeManifest(manifest) { + continue + } + + for entry, err := range manifest.Entries(fs, discardDeleted) { + if err != nil { + yieldError(fmt.Errorf("read manifest %s: %w", manifest.FilePath(), err)) + + return + } + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if err := appendEntry(bldr, entry); err != nil { + yieldError(fmt.Errorf("append manifest entry from %s: %w", manifest.FilePath(), err)) + + return + } + rows++ + if rows == inspectRecordBatchSize && !emit() { + return + } + } + } + + if rows > 0 { + _ = emit() + } else if !emitted { + emitEmpty() + } + }), nil +} + +func emptyInspectRecordBatch(alloc memory.Allocator, schema *arrow.Schema) iter.Seq2[arrow.RecordBatch, error] { + return func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(alloc, schema) + defer bldr.Release() + + batch := bldr.NewRecordBatch() + if !yield(batch, nil) { + batch.Release() + } + } +} + +// inspectPartitionType returns the table-wide partition type. It contains the +// union of partition fields from every spec, which lets metadata tables +// represent live files written before partition evolution. +func inspectPartitionType(metadata Metadata) (*iceberg.StructType, error) { + currentSchema := metadata.CurrentSchema() + specs := metadata.PartitionSpecs() + sort.Slice(specs, func(left, right int) bool { + return specs[left].ID() > specs[right].ID() + }) + + selected := make(map[int]iceberg.PartitionField) + fieldsByID := make(map[int]iceberg.NestedField) + for _, spec := range specs { + partitionType := spec.PartitionType(currentSchema) + for idx, field := range spec.Fields() { + active := true + for _, sourceID := range field.SourceIDs { + if _, ok := currentSchema.FindTypeByID(sourceID); !ok { + active = false + + break + } + } + if !active || idx >= len(partitionType.FieldList) { + continue + } + + if previous, exists := selected[field.FieldID]; exists { + if !slices.Equal(previous.SourceIDs, field.SourceIDs) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible source IDs %v and %v", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.SourceIDs, field.SourceIDs) + } + + previousVoid := isInspectVoidTransform(previous.Transform) + fieldVoid := isInspectVoidTransform(field.Transform) + if previousVoid || fieldVoid { + if previousVoid && !fieldVoid { + selected[field.FieldID] = field + old := fieldsByID[field.FieldID] + old.Type = partitionType.FieldList[idx].Type + fieldsByID[field.FieldID] = old + } + + continue + } + + if !previous.Transform.Equals(field.Transform) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible transforms %q and %q", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.Transform, field.Transform) + } + + continue + } + + selected[field.FieldID] = field + fieldsByID[field.FieldID] = iceberg.NestedField{ + ID: field.FieldID, + Name: field.Name, + Type: partitionType.FieldList[idx].Type, + Required: false, + } + } + } + + fields := make([]iceberg.NestedField, 0, len(fieldsByID)) + for _, field := range fieldsByID { + fields = append(fields, field) + } + sort.Slice(fields, func(left, right int) bool { return fields[left].ID < fields[right].ID }) + + return &iceberg.StructType{FieldList: fields}, nil +} + +func isInspectVoidTransform(transform iceberg.Transform) bool { + switch transform.(type) { + case iceberg.VoidTransform, *iceberg.VoidTransform: + return true + default: + return false + } +} + +// DataFilesSchema returns the common content-file schema used by the data_files +// and delete_files metadata tables. The partition field is omitted for an +// unpartitioned table, as required by the Iceberg metadata-table spec. +func DataFilesSchema(partitionType *iceberg.StructType) *iceberg.Schema { + return iceberg.NewSchema(0, inspectContentFileFields(partitionType)...) +} + +func inspectContentFileType(partitionType *iceberg.StructType) *iceberg.StructType { + return &iceberg.StructType{FieldList: inspectContentFileFields(partitionType)} +} + +func inspectContentFileFields(partitionType *iceberg.StructType) []iceberg.NestedField { + fields := []iceberg.NestedField{ + {ID: 134, Name: "content", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + {ID: 100, Name: "file_path", Type: iceberg.PrimitiveTypes.String, Required: true}, + {ID: 101, Name: "file_format", Type: iceberg.PrimitiveTypes.String, Required: true}, + {ID: 141, Name: "spec_id", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + } + if partitionType != nil && len(partitionType.FieldList) > 0 { + fields = append(fields, iceberg.NestedField{ID: 102, Name: "partition", Type: partitionType, Required: true}) + } + fields = append(fields, + iceberg.NestedField{ID: 103, Name: "record_count", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 104, Name: "file_size_in_bytes", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 108, Name: "column_sizes", Type: inspectInt64MapType(117, 118), Required: false}, + iceberg.NestedField{ID: 109, Name: "value_counts", Type: inspectInt64MapType(119, 120), Required: false}, + iceberg.NestedField{ID: 110, Name: "null_value_counts", Type: inspectInt64MapType(121, 122), Required: false}, + iceberg.NestedField{ID: 137, Name: "nan_value_counts", Type: inspectInt64MapType(138, 139), Required: false}, + iceberg.NestedField{ID: 125, Name: "lower_bounds", Type: inspectBinaryMapType(126, 127), Required: false}, Review Comment: Both Java (`BaseFilesTable` joining `MetricsUtil.readableMetricsSchema`) and PyIceberg (`_get_files_schema`) append a `readable_metrics` struct here, per-column bounds decoded from binary into native types. We only emit the raw `lower_bounds`/`upper_bounds` as `map<int, binary>`, so `SELECT readable_metrics FROM ..._files` against an iceberg-go table comes back column-not-found. Building it means threading the table schema in and decoding each primitive column's bounds, which is real work. Is that intended for a follow-up, or in scope here? Fine either way, I'd just want it called out explicitly rather than silently diverging from the other engines. ########## table/inspect_internal_test.go: ########## @@ -404,6 +409,197 @@ func TestInspectSnapshotsEmpty(t *testing.T) { require.EqualValues(t, 6, rec.NumCols()) } +func TestDataFilesSchema(t *testing.T) { + sc := DataFilesSchema(&iceberg.StructType{FieldList: []iceberg.NestedField{ + {ID: 1000, Name: "bucket", Type: iceberg.PrimitiveTypes.Int32, Required: true}, + }}) + + require.Equal(t, []string{ + "content", "file_path", "file_format", "spec_id", "partition", + "record_count", "file_size_in_bytes", "column_sizes", "value_counts", "null_value_counts", + "nan_value_counts", "lower_bounds", "upper_bounds", "key_metadata", "split_offsets", + "equality_ids", "sort_order_id", "first_row_id", "referenced_data_file", "content_offset", + "content_size_in_bytes", + }, testFieldNames(sc)) + + fields := sc.Fields() + require.Equal(t, 134, fields[0].ID) + require.Equal(t, 100, fields[1].ID) + require.Equal(t, 141, fields[3].ID) + require.Equal(t, 102, fields[4].ID) + require.Equal(t, 145, fields[len(fields)-1].ID) + + unpartitioned := DataFilesSchema(&iceberg.StructType{}) + require.NotContains(t, testFieldNames(unpartitioned), "partition") +} + +func TestInspectDataFilesStreamsBatchesAndSkipsDeleted(t *testing.T) { + const snapshotID = int64(1) + spec := *iceberg.UnpartitionedSpec + txn, memIO := createTestTransactionWithMemIO(t, spec) + schema := simpleSchema() + + entries := make([]iceberg.ManifestEntry, 0, inspectRecordBatchSize+2) + for index := 0; index < inspectRecordBatchSize+1; index++ { + file := newTestDataFile(t, spec, + "mem://default/table-location/data/live-"+strconv.Itoa(index)+".parquet", nil) + sequenceNumber := int64(1) + entries = append(entries, iceberg.NewManifestEntry( + iceberg.EntryStatusADDED, int64Ptr(snapshotID), &sequenceNumber, &sequenceNumber, file)) + } + deletedPath := "mem://default/table-location/data/deleted.parquet" + deleted := newTestDataFile(t, spec, deletedPath, nil) + deletedSequenceNumber := int64(1) + entries = append(entries, iceberg.NewManifestEntry( + iceberg.EntryStatusDELETED, int64Ptr(snapshotID), &deletedSequenceNumber, &deletedSequenceNumber, deleted)) + + manifestPath := "mem://default/table-location/metadata/data-manifest.avro" + manifestListPath := "mem://default/table-location/metadata/snap-1-manifest-list.avro" + var manifestBuf bytes.Buffer + manifest, err := iceberg.WriteManifest(manifestPath, &manifestBuf, 2, spec, schema, snapshotID, entries) + require.NoError(t, err) + require.NoError(t, memIO.WriteFile(manifestPath, manifestBuf.Bytes())) + + var listBuf bytes.Buffer + sequenceNumber := int64(1) + require.NoError(t, iceberg.WriteManifestList(2, &listBuf, snapshotID, nil, &sequenceNumber, 0, + []iceberg.ManifestFile{manifest})) + require.NoError(t, memIO.WriteFile(manifestListPath, listBuf.Bytes())) + + snapID := snapshotID + txn.meta.snapshotList = []Snapshot{{ + SnapshotID: snapshotID, + ManifestList: manifestListPath, + SequenceNumber: sequenceNumber, + }} + txn.meta.currentSnapshotID = &snapID + 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().DataFiles(context.Background()) + require.NoError(t, err) + defer rr.Release() + + var batchRows []int + var paths []string + for rr.Next() { Review Comment: This streaming path has the trickiest ownership in the file, the `emit` closure releasing batches, `NewRecordBatch`, the yield-false early exit, and none of it runs under a checked allocator. `TestInspectAllocatorOption` does `checked.AssertSize(t, 0)` for the Snapshots path; I'd add the equivalent here, with one variant that drains fully and one that abandons mid-stream (`rr.Release()` after a single batch) so the early-exit release path is actually covered. ########## table/inspect_files.go: ########## @@ -0,0 +1,533 @@ +// 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" + "iter" + "slices" + "sort" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/arrow-go/v18/arrow/scalar" + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +// DataFiles returns the live data files in the current snapshot. Deleted +// manifest entries are omitted, matching the data_files metadata table. +func (i InspectTable) DataFiles(ctx context.Context) (array.RecordReader, error) { + partitionType, err := inspectPartitionType(i.tbl.metadata) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + schema := DataFilesSchema(partitionType) + arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false) + if err != nil { + return nil, fmt.Errorf("inspect data files: build arrow schema: %w", err) + } + + rr, err := i.manifestEntryReader(ctx, arrowSchema, true, + func(manifest iceberg.ManifestFile) bool { + return manifest.ManifestContent() == iceberg.ManifestContentData + }, + func(bldr *array.RecordBuilder, entry iceberg.ManifestEntry) error { + return appendContentFileRecord(bldr, partitionType, entry.DataFile()) + }) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + + return rr, nil +} + +const inspectRecordBatchSize = 4096 + +// manifestEntryReader streams manifest entries into bounded Arrow record +// batches. It keeps only the current batch and manifest decoder state in +// memory instead of materializing every entry before returning a reader. +func (i InspectTable) manifestEntryReader( + ctx context.Context, + arrowSchema *arrow.Schema, + discardDeleted bool, + includeManifest func(iceberg.ManifestFile) bool, + appendEntry func(*array.RecordBuilder, iceberg.ManifestEntry) error, +) (array.RecordReader, error) { + snapshot := i.tbl.metadata.CurrentSnapshot() + if snapshot == nil { + return array.ReaderFromIter(arrowSchema, emptyInspectRecordBatch(i.alloc, arrowSchema)), 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 + } + + return array.ReaderFromIter(arrowSchema, func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(i.alloc, arrowSchema) + defer bldr.Release() + + rows := 0 + emitted := false + emit := func() bool { + if rows == 0 { + return true + } + + batch := bldr.NewRecordBatch() + rows = 0 + emitted = true + if yield(batch, nil) { + return true + } + + batch.Release() + + return false + } + emitEmpty := func() { + batch := bldr.NewRecordBatch() + emitted = true + if !yield(batch, nil) { + batch.Release() + } + } + yieldError := func(err error) { + _ = yield(nil, err) + } + + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if includeManifest != nil && !includeManifest(manifest) { + continue + } + + for entry, err := range manifest.Entries(fs, discardDeleted) { + if err != nil { + yieldError(fmt.Errorf("read manifest %s: %w", manifest.FilePath(), err)) + + return + } + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if err := appendEntry(bldr, entry); err != nil { + yieldError(fmt.Errorf("append manifest entry from %s: %w", manifest.FilePath(), err)) + + return + } + rows++ + if rows == inspectRecordBatchSize && !emit() { + return + } + } + } + + if rows > 0 { + _ = emit() + } else if !emitted { + emitEmpty() + } + }), nil +} + +func emptyInspectRecordBatch(alloc memory.Allocator, schema *arrow.Schema) iter.Seq2[arrow.RecordBatch, error] { + return func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(alloc, schema) + defer bldr.Release() + + batch := bldr.NewRecordBatch() + if !yield(batch, nil) { + batch.Release() + } + } +} + +// inspectPartitionType returns the table-wide partition type. It contains the +// union of partition fields from every spec, which lets metadata tables +// represent live files written before partition evolution. +func inspectPartitionType(metadata Metadata) (*iceberg.StructType, error) { + currentSchema := metadata.CurrentSchema() + specs := metadata.PartitionSpecs() + sort.Slice(specs, func(left, right int) bool { + return specs[left].ID() > specs[right].ID() + }) + + selected := make(map[int]iceberg.PartitionField) + fieldsByID := make(map[int]iceberg.NestedField) + for _, spec := range specs { + partitionType := spec.PartitionType(currentSchema) + for idx, field := range spec.Fields() { + active := true + for _, sourceID := range field.SourceIDs { + if _, ok := currentSchema.FindTypeByID(sourceID); !ok { + active = false + + break + } + } + if !active || idx >= len(partitionType.FieldList) { + continue + } + + if previous, exists := selected[field.FieldID]; exists { + if !slices.Equal(previous.SourceIDs, field.SourceIDs) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible source IDs %v and %v", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.SourceIDs, field.SourceIDs) + } + + previousVoid := isInspectVoidTransform(previous.Transform) + fieldVoid := isInspectVoidTransform(field.Transform) + if previousVoid || fieldVoid { + if previousVoid && !fieldVoid { + selected[field.FieldID] = field + old := fieldsByID[field.FieldID] + old.Type = partitionType.FieldList[idx].Type + fieldsByID[field.FieldID] = old + } + + continue + } + + if !previous.Transform.Equals(field.Transform) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible transforms %q and %q", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.Transform, field.Transform) + } + + continue + } + + selected[field.FieldID] = field + fieldsByID[field.FieldID] = iceberg.NestedField{ + ID: field.FieldID, + Name: field.Name, + Type: partitionType.FieldList[idx].Type, + Required: false, + } + } + } + + fields := make([]iceberg.NestedField, 0, len(fieldsByID)) + for _, field := range fieldsByID { + fields = append(fields, field) + } + sort.Slice(fields, func(left, right int) bool { return fields[left].ID < fields[right].ID }) + + return &iceberg.StructType{FieldList: fields}, nil +} + +func isInspectVoidTransform(transform iceberg.Transform) bool { + switch transform.(type) { + case iceberg.VoidTransform, *iceberg.VoidTransform: + return true + default: + return false + } +} + +// DataFilesSchema returns the common content-file schema used by the data_files +// and delete_files metadata tables. The partition field is omitted for an +// unpartitioned table, as required by the Iceberg metadata-table spec. +func DataFilesSchema(partitionType *iceberg.StructType) *iceberg.Schema { + return iceberg.NewSchema(0, inspectContentFileFields(partitionType)...) +} + +func inspectContentFileType(partitionType *iceberg.StructType) *iceberg.StructType { + return &iceberg.StructType{FieldList: inspectContentFileFields(partitionType)} +} + +func inspectContentFileFields(partitionType *iceberg.StructType) []iceberg.NestedField { + fields := []iceberg.NestedField{ + {ID: 134, Name: "content", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + {ID: 100, Name: "file_path", Type: iceberg.PrimitiveTypes.String, Required: true}, + {ID: 101, Name: "file_format", Type: iceberg.PrimitiveTypes.String, Required: true}, + {ID: 141, Name: "spec_id", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + } + if partitionType != nil && len(partitionType.FieldList) > 0 { + fields = append(fields, iceberg.NestedField{ID: 102, Name: "partition", Type: partitionType, Required: true}) + } + fields = append(fields, + iceberg.NestedField{ID: 103, Name: "record_count", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 104, Name: "file_size_in_bytes", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 108, Name: "column_sizes", Type: inspectInt64MapType(117, 118), Required: false}, + iceberg.NestedField{ID: 109, Name: "value_counts", Type: inspectInt64MapType(119, 120), Required: false}, + iceberg.NestedField{ID: 110, Name: "null_value_counts", Type: inspectInt64MapType(121, 122), Required: false}, + iceberg.NestedField{ID: 137, Name: "nan_value_counts", Type: inspectInt64MapType(138, 139), Required: false}, + iceberg.NestedField{ID: 125, Name: "lower_bounds", Type: inspectBinaryMapType(126, 127), Required: false}, + iceberg.NestedField{ID: 128, Name: "upper_bounds", Type: inspectBinaryMapType(129, 130), Required: false}, + iceberg.NestedField{ID: 131, Name: "key_metadata", Type: iceberg.PrimitiveTypes.Binary, Required: false}, + iceberg.NestedField{ID: 132, Name: "split_offsets", Type: &iceberg.ListType{ElementID: 133, Element: iceberg.PrimitiveTypes.Int64, ElementRequired: true}, Required: false}, + iceberg.NestedField{ID: 135, Name: "equality_ids", Type: &iceberg.ListType{ElementID: 136, Element: iceberg.PrimitiveTypes.Int32, ElementRequired: true}, Required: false}, + iceberg.NestedField{ID: 140, Name: "sort_order_id", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + iceberg.NestedField{ID: 142, Name: "first_row_id", Type: iceberg.PrimitiveTypes.Int64, Required: false}, + iceberg.NestedField{ID: 143, Name: "referenced_data_file", Type: iceberg.PrimitiveTypes.String, Required: false}, + iceberg.NestedField{ID: 144, Name: "content_offset", Type: iceberg.PrimitiveTypes.Int64, Required: false}, + iceberg.NestedField{ID: 145, Name: "content_size_in_bytes", Type: iceberg.PrimitiveTypes.Int64, Required: false}, + ) + + return fields +} + +func inspectInt64MapType(keyID, valueID int) *iceberg.MapType { + return &iceberg.MapType{KeyID: keyID, KeyType: iceberg.PrimitiveTypes.Int32, ValueID: valueID, ValueType: iceberg.PrimitiveTypes.Int64, ValueRequired: true} +} + +func inspectBinaryMapType(keyID, valueID int) *iceberg.MapType { + return &iceberg.MapType{KeyID: keyID, KeyType: iceberg.PrimitiveTypes.Int32, ValueID: valueID, ValueType: iceberg.PrimitiveTypes.Binary, ValueRequired: true} +} + +func appendContentFileRecord(bldr *array.RecordBuilder, partitionType *iceberg.StructType, file iceberg.DataFile) error { + return appendContentFileFields(bldr.Field, partitionType, file) +} + +func appendContentFile(builder *array.StructBuilder, partitionType *iceberg.StructType, file iceberg.DataFile) error { + builder.Append(true) + + return appendContentFileFields(builder.FieldBuilder, partitionType, file) +} + +func appendContentFileFields(fieldBuilder func(int) array.Builder, partitionType *iceberg.StructType, file iceberg.DataFile) error { + idx := 0 + fieldBuilder(idx).(*array.Int32Builder).Append(int32(file.ContentType())) + idx++ + fieldBuilder(idx).(*array.StringBuilder).Append(file.FilePath()) + idx++ + fieldBuilder(idx).(*array.StringBuilder).Append(string(file.FileFormat())) + idx++ + fieldBuilder(idx).(*array.Int32Builder).Append(file.SpecID()) + idx++ + + if partitionType != nil && len(partitionType.FieldList) > 0 { + partition := fieldBuilder(idx).(*array.StructBuilder) + if err := appendInspectPartition(partition, partitionType, file.Partition()); err != nil { + return err + } + idx++ + } + + fieldBuilder(idx).(*array.Int64Builder).Append(file.Count()) + idx++ + fieldBuilder(idx).(*array.Int64Builder).Append(file.FileSizeBytes()) + idx++ + appendInspectInt64Map(fieldBuilder(idx).(*array.MapBuilder), file.ColumnSizes()) + idx++ + appendInspectInt64Map(fieldBuilder(idx).(*array.MapBuilder), file.ValueCounts()) + idx++ + appendInspectInt64Map(fieldBuilder(idx).(*array.MapBuilder), file.NullValueCounts()) + idx++ + appendInspectInt64Map(fieldBuilder(idx).(*array.MapBuilder), file.NaNValueCounts()) + idx++ + appendInspectBinaryMap(fieldBuilder(idx).(*array.MapBuilder), file.LowerBoundValues()) + idx++ + appendInspectBinaryMap(fieldBuilder(idx).(*array.MapBuilder), file.UpperBoundValues()) + idx++ + appendInspectBytes(fieldBuilder(idx), file.KeyMetadata()) + idx++ + appendInspectInt64List(fieldBuilder(idx).(*array.ListBuilder), file.SplitOffsets()) + idx++ + appendInspectInt32List(fieldBuilder(idx).(*array.ListBuilder), file.EqualityFieldIDs()) + idx++ + appendInspectOptionalInt32(fieldBuilder(idx).(*array.Int32Builder), file.SortOrderID()) + idx++ + appendInspectOptionalInt64(fieldBuilder(idx).(*array.Int64Builder), file.FirstRowID()) + idx++ + appendInspectOptionalString(fieldBuilder(idx).(*array.StringBuilder), file.ReferencedDataFile()) + idx++ + appendInspectOptionalInt64(fieldBuilder(idx).(*array.Int64Builder), file.ContentOffset()) + idx++ + appendInspectOptionalInt64(fieldBuilder(idx).(*array.Int64Builder), file.ContentSizeInBytes()) + + return nil +} + +func appendInspectPartition(builder *array.StructBuilder, partitionType *iceberg.StructType, values map[int]any) error { + arrowType := builder.Type().(*arrow.StructType) + builder.Append(true) + for idx, field := range partitionType.FieldList { + value := values[field.ID] + if value == nil { + builder.FieldBuilder(idx).AppendNull() + + continue + } + sc, err := inspectValueScalar(value, field.Type, arrowType.Field(idx).Type) + if err != nil { + return fmt.Errorf("partition field %q: %w", field.Name, err) + } + if err := scalar.Append(builder.FieldBuilder(idx), sc); err != nil { + return err + } + } + + return nil +} + +func inspectValueScalar(value any, typ iceberg.Type, arrowType arrow.DataType) (scalar.Scalar, error) { + switch typ.(type) { + case iceberg.DateType: + switch value := value.(type) { + case iceberg.Date: + return scalar.NewDate32Scalar(arrow.Date32(value)), nil + case int32: + return scalar.NewDate32Scalar(arrow.Date32(value)), nil + } + case iceberg.TimeType: + if value, ok := value.(iceberg.Time); ok { + return scalar.NewTime64Scalar(arrow.Time64(value), arrowType), nil + } + case iceberg.TimestampType, iceberg.TimestampTzType: + if value, ok := value.(iceberg.Timestamp); ok { + return scalar.NewTimestampScalar(arrow.Timestamp(value), arrowType), nil + } + case iceberg.TimestampNsType, iceberg.TimestampTzNsType: + if value, ok := value.(iceberg.TimestampNano); ok { + return scalar.NewTimestampScalar(arrow.Timestamp(value), arrowType), nil + } + case iceberg.UUIDType: + if value, ok := value.(uuid.UUID); ok { + return scalar.MakeScalarParam(value[:], arrowType) + } + case iceberg.DecimalType: + switch value := value.(type) { + case iceberg.DecimalLiteral: + return scalar.NewDecimal128Scalar(value.Val, arrowType), nil + case iceberg.Decimal: + return scalar.NewDecimal128Scalar(value.Val, arrowType), nil + default: + return nil, fmt.Errorf("unsupported decimal partition value %T", value) + } + } + + return scalar.MakeScalarParam(value, arrowType) +} + +func appendInspectInt64Map(builder *array.MapBuilder, values map[int]int64) { + if values == nil { + builder.AppendNull() + + return + } + builder.Append(true) + keys := builder.KeyBuilder().(*array.Int32Builder) + items := builder.ItemBuilder().(*array.Int64Builder) + ids := make([]int, 0, len(values)) + for id := range values { + ids = append(ids, id) + } + sort.Ints(ids) + for _, id := range ids { + keys.Append(int32(id)) + items.Append(values[id]) + } +} + +func appendInspectBinaryMap(builder *array.MapBuilder, values map[int][]byte) { + if values == nil { + builder.AppendNull() + + return + } + builder.Append(true) + keys := builder.KeyBuilder().(*array.Int32Builder) + items := builder.ItemBuilder().(*array.BinaryBuilder) + ids := make([]int, 0, len(values)) + for id := range values { + ids = append(ids, id) + } + sort.Ints(ids) + for _, id := range ids { + keys.Append(int32(id)) + items.Append(values[id]) + } +} + +func appendInspectBytes(builder array.Builder, value []byte) { + if value == nil { + builder.AppendNull() + + return + } + builder.(*array.BinaryBuilder).Append(value) Review Comment: This `array.Builder` gets asserted to `*array.BinaryBuilder` here with no comma-ok, so if field order ever drifts from the schema this panics with nothing to recover from. Every other append helper takes a concrete builder type, which moves the assertion to the `fieldBuilder(idx)` call site. I'd have this one take `*array.BinaryBuilder` directly to match, same failure but at least consistent with the rest. ########## table/inspect_internal_test.go: ########## @@ -404,6 +409,197 @@ func TestInspectSnapshotsEmpty(t *testing.T) { require.EqualValues(t, 6, rec.NumCols()) } +func TestDataFilesSchema(t *testing.T) { + sc := DataFilesSchema(&iceberg.StructType{FieldList: []iceberg.NestedField{ + {ID: 1000, Name: "bucket", Type: iceberg.PrimitiveTypes.Int32, Required: true}, + }}) + + require.Equal(t, []string{ + "content", "file_path", "file_format", "spec_id", "partition", + "record_count", "file_size_in_bytes", "column_sizes", "value_counts", "null_value_counts", + "nan_value_counts", "lower_bounds", "upper_bounds", "key_metadata", "split_offsets", + "equality_ids", "sort_order_id", "first_row_id", "referenced_data_file", "content_offset", + "content_size_in_bytes", + }, testFieldNames(sc)) + + fields := sc.Fields() + require.Equal(t, 134, fields[0].ID) + require.Equal(t, 100, fields[1].ID) + require.Equal(t, 141, fields[3].ID) + require.Equal(t, 102, fields[4].ID) + require.Equal(t, 145, fields[len(fields)-1].ID) + + unpartitioned := DataFilesSchema(&iceberg.StructType{}) + require.NotContains(t, testFieldNames(unpartitioned), "partition") +} + +func TestInspectDataFilesStreamsBatchesAndSkipsDeleted(t *testing.T) { + const snapshotID = int64(1) + spec := *iceberg.UnpartitionedSpec Review Comment: This test (and the whole suite) only ever uses `UnpartitionedSpec`, so the partition column branch in `appendContentFileFields` / `appendInspectPartition` / `inspectValueScalar` never runs end-to-end. `TestInspectValueScalarDecimal` and `TestInspectPartitionTypeUsesAllActiveSpecs` exercise those pieces in isolation, but nothing writes a real partitioned data file and reads it back through `DataFiles`, which is exactly the path most likely to panic (see the temporal/UUID fallthrough). I'd add an integration test with at least an identity-partitioned int32 column that verifies the partition values come back correct. ########## table/inspect_files.go: ########## @@ -0,0 +1,533 @@ +// 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" + "iter" + "slices" + "sort" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/arrow-go/v18/arrow/scalar" + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +// DataFiles returns the live data files in the current snapshot. Deleted +// manifest entries are omitted, matching the data_files metadata table. +func (i InspectTable) DataFiles(ctx context.Context) (array.RecordReader, error) { + partitionType, err := inspectPartitionType(i.tbl.metadata) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + schema := DataFilesSchema(partitionType) + arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false) + if err != nil { + return nil, fmt.Errorf("inspect data files: build arrow schema: %w", err) + } + + rr, err := i.manifestEntryReader(ctx, arrowSchema, true, + func(manifest iceberg.ManifestFile) bool { + return manifest.ManifestContent() == iceberg.ManifestContentData + }, + func(bldr *array.RecordBuilder, entry iceberg.ManifestEntry) error { + return appendContentFileRecord(bldr, partitionType, entry.DataFile()) + }) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + + return rr, nil +} + +const inspectRecordBatchSize = 4096 + +// manifestEntryReader streams manifest entries into bounded Arrow record +// batches. It keeps only the current batch and manifest decoder state in +// memory instead of materializing every entry before returning a reader. +func (i InspectTable) manifestEntryReader( + ctx context.Context, + arrowSchema *arrow.Schema, + discardDeleted bool, + includeManifest func(iceberg.ManifestFile) bool, + appendEntry func(*array.RecordBuilder, iceberg.ManifestEntry) error, +) (array.RecordReader, error) { + snapshot := i.tbl.metadata.CurrentSnapshot() + if snapshot == nil { + return array.ReaderFromIter(arrowSchema, emptyInspectRecordBatch(i.alloc, arrowSchema)), 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 + } + + return array.ReaderFromIter(arrowSchema, func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(i.alloc, arrowSchema) + defer bldr.Release() + + rows := 0 + emitted := false + emit := func() bool { + if rows == 0 { + return true + } + + batch := bldr.NewRecordBatch() + rows = 0 + emitted = true + if yield(batch, nil) { + return true + } + + batch.Release() + + return false + } + emitEmpty := func() { Review Comment: Worth noting `DataFiles` always yields at least one batch, a snapshot with only delete manifests skips every entry, `emitted` stays false, and `emitEmpty` fires a zero-row batch, same as the empty-snapshot case but through a different code path. That seems intentional and matches the existing Snapshots/History behavior, so mostly I'd just document the "always >=1 batch" invariant and add a test case for the all-deleted-manifest path so the two routes stay pinned. ########## table/inspect_files.go: ########## @@ -0,0 +1,533 @@ +// 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" + "iter" + "slices" + "sort" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/arrow-go/v18/arrow/scalar" + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +// DataFiles returns the live data files in the current snapshot. Deleted +// manifest entries are omitted, matching the data_files metadata table. +func (i InspectTable) DataFiles(ctx context.Context) (array.RecordReader, error) { + partitionType, err := inspectPartitionType(i.tbl.metadata) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + schema := DataFilesSchema(partitionType) + arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false) + if err != nil { + return nil, fmt.Errorf("inspect data files: build arrow schema: %w", err) + } + + rr, err := i.manifestEntryReader(ctx, arrowSchema, true, + func(manifest iceberg.ManifestFile) bool { + return manifest.ManifestContent() == iceberg.ManifestContentData + }, + func(bldr *array.RecordBuilder, entry iceberg.ManifestEntry) error { + return appendContentFileRecord(bldr, partitionType, entry.DataFile()) + }) + if err != nil { + return nil, fmt.Errorf("inspect data files: %w", err) + } + + return rr, nil +} + +const inspectRecordBatchSize = 4096 + +// manifestEntryReader streams manifest entries into bounded Arrow record +// batches. It keeps only the current batch and manifest decoder state in +// memory instead of materializing every entry before returning a reader. +func (i InspectTable) manifestEntryReader( + ctx context.Context, + arrowSchema *arrow.Schema, + discardDeleted bool, + includeManifest func(iceberg.ManifestFile) bool, + appendEntry func(*array.RecordBuilder, iceberg.ManifestEntry) error, +) (array.RecordReader, error) { + snapshot := i.tbl.metadata.CurrentSnapshot() + if snapshot == nil { + return array.ReaderFromIter(arrowSchema, emptyInspectRecordBatch(i.alloc, arrowSchema)), 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 + } + + return array.ReaderFromIter(arrowSchema, func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(i.alloc, arrowSchema) + defer bldr.Release() + + rows := 0 + emitted := false + emit := func() bool { + if rows == 0 { + return true + } + + batch := bldr.NewRecordBatch() + rows = 0 + emitted = true + if yield(batch, nil) { + return true + } + + batch.Release() + + return false + } + emitEmpty := func() { + batch := bldr.NewRecordBatch() + emitted = true + if !yield(batch, nil) { + batch.Release() + } + } + yieldError := func(err error) { + _ = yield(nil, err) + } + + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if includeManifest != nil && !includeManifest(manifest) { + continue + } + + for entry, err := range manifest.Entries(fs, discardDeleted) { + if err != nil { + yieldError(fmt.Errorf("read manifest %s: %w", manifest.FilePath(), err)) + + return + } + if err := ctx.Err(); err != nil { + yieldError(err) + + return + } + if err := appendEntry(bldr, entry); err != nil { + yieldError(fmt.Errorf("append manifest entry from %s: %w", manifest.FilePath(), err)) + + return + } + rows++ + if rows == inspectRecordBatchSize && !emit() { + return + } + } + } + + if rows > 0 { + _ = emit() + } else if !emitted { + emitEmpty() + } + }), nil +} + +func emptyInspectRecordBatch(alloc memory.Allocator, schema *arrow.Schema) iter.Seq2[arrow.RecordBatch, error] { + return func(yield func(arrow.RecordBatch, error) bool) { + bldr := array.NewRecordBuilder(alloc, schema) + defer bldr.Release() + + batch := bldr.NewRecordBatch() + if !yield(batch, nil) { + batch.Release() + } + } +} + +// inspectPartitionType returns the table-wide partition type. It contains the +// union of partition fields from every spec, which lets metadata tables +// represent live files written before partition evolution. +func inspectPartitionType(metadata Metadata) (*iceberg.StructType, error) { + currentSchema := metadata.CurrentSchema() + specs := metadata.PartitionSpecs() + sort.Slice(specs, func(left, right int) bool { + return specs[left].ID() > specs[right].ID() + }) + + selected := make(map[int]iceberg.PartitionField) + fieldsByID := make(map[int]iceberg.NestedField) + for _, spec := range specs { + partitionType := spec.PartitionType(currentSchema) + for idx, field := range spec.Fields() { + active := true + for _, sourceID := range field.SourceIDs { + if _, ok := currentSchema.FindTypeByID(sourceID); !ok { + active = false + + break + } + } + if !active || idx >= len(partitionType.FieldList) { + continue + } + + if previous, exists := selected[field.FieldID]; exists { + if !slices.Equal(previous.SourceIDs, field.SourceIDs) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible source IDs %v and %v", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.SourceIDs, field.SourceIDs) + } + + previousVoid := isInspectVoidTransform(previous.Transform) + fieldVoid := isInspectVoidTransform(field.Transform) + if previousVoid || fieldVoid { + if previousVoid && !fieldVoid { + selected[field.FieldID] = field + old := fieldsByID[field.FieldID] + old.Type = partitionType.FieldList[idx].Type + fieldsByID[field.FieldID] = old + } + + continue + } + + if !previous.Transform.Equals(field.Transform) { + return nil, fmt.Errorf("%w: partition field ID %d has incompatible transforms %q and %q", + iceberg.ErrInvalidPartitionSpec, field.FieldID, previous.Transform, field.Transform) + } + + continue + } + + selected[field.FieldID] = field + fieldsByID[field.FieldID] = iceberg.NestedField{ + ID: field.FieldID, + Name: field.Name, + Type: partitionType.FieldList[idx].Type, + Required: false, + } + } + } + + fields := make([]iceberg.NestedField, 0, len(fieldsByID)) + for _, field := range fieldsByID { + fields = append(fields, field) + } + sort.Slice(fields, func(left, right int) bool { return fields[left].ID < fields[right].ID }) + + return &iceberg.StructType{FieldList: fields}, nil +} + +func isInspectVoidTransform(transform iceberg.Transform) bool { + switch transform.(type) { + case iceberg.VoidTransform, *iceberg.VoidTransform: + return true + default: + return false + } +} + +// DataFilesSchema returns the common content-file schema used by the data_files +// and delete_files metadata tables. The partition field is omitted for an +// unpartitioned table, as required by the Iceberg metadata-table spec. +func DataFilesSchema(partitionType *iceberg.StructType) *iceberg.Schema { + return iceberg.NewSchema(0, inspectContentFileFields(partitionType)...) +} + +func inspectContentFileType(partitionType *iceberg.StructType) *iceberg.StructType { + return &iceberg.StructType{FieldList: inspectContentFileFields(partitionType)} +} + +func inspectContentFileFields(partitionType *iceberg.StructType) []iceberg.NestedField { + fields := []iceberg.NestedField{ + {ID: 134, Name: "content", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + {ID: 100, Name: "file_path", Type: iceberg.PrimitiveTypes.String, Required: true}, + {ID: 101, Name: "file_format", Type: iceberg.PrimitiveTypes.String, Required: true}, + {ID: 141, Name: "spec_id", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + } + if partitionType != nil && len(partitionType.FieldList) > 0 { + fields = append(fields, iceberg.NestedField{ID: 102, Name: "partition", Type: partitionType, Required: true}) + } + fields = append(fields, + iceberg.NestedField{ID: 103, Name: "record_count", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 104, Name: "file_size_in_bytes", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 108, Name: "column_sizes", Type: inspectInt64MapType(117, 118), Required: false}, + iceberg.NestedField{ID: 109, Name: "value_counts", Type: inspectInt64MapType(119, 120), Required: false}, + iceberg.NestedField{ID: 110, Name: "null_value_counts", Type: inspectInt64MapType(121, 122), Required: false}, + iceberg.NestedField{ID: 137, Name: "nan_value_counts", Type: inspectInt64MapType(138, 139), Required: false}, + iceberg.NestedField{ID: 125, Name: "lower_bounds", Type: inspectBinaryMapType(126, 127), Required: false}, + iceberg.NestedField{ID: 128, Name: "upper_bounds", Type: inspectBinaryMapType(129, 130), Required: false}, + iceberg.NestedField{ID: 131, Name: "key_metadata", Type: iceberg.PrimitiveTypes.Binary, Required: false}, + iceberg.NestedField{ID: 132, Name: "split_offsets", Type: &iceberg.ListType{ElementID: 133, Element: iceberg.PrimitiveTypes.Int64, ElementRequired: true}, Required: false}, + iceberg.NestedField{ID: 135, Name: "equality_ids", Type: &iceberg.ListType{ElementID: 136, Element: iceberg.PrimitiveTypes.Int32, ElementRequired: true}, Required: false}, + iceberg.NestedField{ID: 140, Name: "sort_order_id", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + iceberg.NestedField{ID: 142, Name: "first_row_id", Type: iceberg.PrimitiveTypes.Int64, Required: false}, + iceberg.NestedField{ID: 143, Name: "referenced_data_file", Type: iceberg.PrimitiveTypes.String, Required: false}, + iceberg.NestedField{ID: 144, Name: "content_offset", Type: iceberg.PrimitiveTypes.Int64, Required: false}, + iceberg.NestedField{ID: 145, Name: "content_size_in_bytes", Type: iceberg.PrimitiveTypes.Int64, Required: false}, + ) + + return fields +} + +func inspectInt64MapType(keyID, valueID int) *iceberg.MapType { + return &iceberg.MapType{KeyID: keyID, KeyType: iceberg.PrimitiveTypes.Int32, ValueID: valueID, ValueType: iceberg.PrimitiveTypes.Int64, ValueRequired: true} +} + +func inspectBinaryMapType(keyID, valueID int) *iceberg.MapType { + return &iceberg.MapType{KeyID: keyID, KeyType: iceberg.PrimitiveTypes.Int32, ValueID: valueID, ValueType: iceberg.PrimitiveTypes.Binary, ValueRequired: true} +} + +func appendContentFileRecord(bldr *array.RecordBuilder, partitionType *iceberg.StructType, file iceberg.DataFile) error { + return appendContentFileFields(bldr.Field, partitionType, file) +} + +func appendContentFile(builder *array.StructBuilder, partitionType *iceberg.StructType, file iceberg.DataFile) error { + builder.Append(true) + + return appendContentFileFields(builder.FieldBuilder, partitionType, file) +} + +func appendContentFileFields(fieldBuilder func(int) array.Builder, partitionType *iceberg.StructType, file iceberg.DataFile) error { + idx := 0 + fieldBuilder(idx).(*array.Int32Builder).Append(int32(file.ContentType())) + idx++ + fieldBuilder(idx).(*array.StringBuilder).Append(file.FilePath()) + idx++ + fieldBuilder(idx).(*array.StringBuilder).Append(string(file.FileFormat())) + idx++ + fieldBuilder(idx).(*array.Int32Builder).Append(file.SpecID()) + idx++ + + if partitionType != nil && len(partitionType.FieldList) > 0 { + partition := fieldBuilder(idx).(*array.StructBuilder) + if err := appendInspectPartition(partition, partitionType, file.Partition()); err != nil { + return err + } + idx++ + } + + fieldBuilder(idx).(*array.Int64Builder).Append(file.Count()) + idx++ + fieldBuilder(idx).(*array.Int64Builder).Append(file.FileSizeBytes()) + idx++ + appendInspectInt64Map(fieldBuilder(idx).(*array.MapBuilder), file.ColumnSizes()) + idx++ + appendInspectInt64Map(fieldBuilder(idx).(*array.MapBuilder), file.ValueCounts()) + idx++ + appendInspectInt64Map(fieldBuilder(idx).(*array.MapBuilder), file.NullValueCounts()) + idx++ + appendInspectInt64Map(fieldBuilder(idx).(*array.MapBuilder), file.NaNValueCounts()) + idx++ + appendInspectBinaryMap(fieldBuilder(idx).(*array.MapBuilder), file.LowerBoundValues()) + idx++ + appendInspectBinaryMap(fieldBuilder(idx).(*array.MapBuilder), file.UpperBoundValues()) + idx++ + appendInspectBytes(fieldBuilder(idx), file.KeyMetadata()) + idx++ + appendInspectInt64List(fieldBuilder(idx).(*array.ListBuilder), file.SplitOffsets()) + idx++ + appendInspectInt32List(fieldBuilder(idx).(*array.ListBuilder), file.EqualityFieldIDs()) + idx++ + appendInspectOptionalInt32(fieldBuilder(idx).(*array.Int32Builder), file.SortOrderID()) + idx++ + appendInspectOptionalInt64(fieldBuilder(idx).(*array.Int64Builder), file.FirstRowID()) + idx++ + appendInspectOptionalString(fieldBuilder(idx).(*array.StringBuilder), file.ReferencedDataFile()) + idx++ + appendInspectOptionalInt64(fieldBuilder(idx).(*array.Int64Builder), file.ContentOffset()) + idx++ + appendInspectOptionalInt64(fieldBuilder(idx).(*array.Int64Builder), file.ContentSizeInBytes()) + + return nil +} + +func appendInspectPartition(builder *array.StructBuilder, partitionType *iceberg.StructType, values map[int]any) error { + arrowType := builder.Type().(*arrow.StructType) + builder.Append(true) + for idx, field := range partitionType.FieldList { + value := values[field.ID] + if value == nil { + builder.FieldBuilder(idx).AppendNull() + + continue + } + sc, err := inspectValueScalar(value, field.Type, arrowType.Field(idx).Type) + if err != nil { + return fmt.Errorf("partition field %q: %w", field.Name, err) + } + if err := scalar.Append(builder.FieldBuilder(idx), sc); err != nil { + return err + } + } + + return nil +} + +func inspectValueScalar(value any, typ iceberg.Type, arrowType arrow.DataType) (scalar.Scalar, error) { + switch typ.(type) { + case iceberg.DateType: + switch value := value.(type) { + case iceberg.Date: + return scalar.NewDate32Scalar(arrow.Date32(value)), nil + case int32: + return scalar.NewDate32Scalar(arrow.Date32(value)), nil + } + case iceberg.TimeType: + if value, ok := value.(iceberg.Time); ok { + return scalar.NewTime64Scalar(arrow.Time64(value), arrowType), nil + } + case iceberg.TimestampType, iceberg.TimestampTzType: + if value, ok := value.(iceberg.Timestamp); ok { + return scalar.NewTimestampScalar(arrow.Timestamp(value), arrowType), nil + } + case iceberg.TimestampNsType, iceberg.TimestampTzNsType: + if value, ok := value.(iceberg.TimestampNano); ok { + return scalar.NewTimestampScalar(arrow.Timestamp(value), arrowType), nil + } + case iceberg.UUIDType: + if value, ok := value.(uuid.UUID); ok { + return scalar.MakeScalarParam(value[:], arrowType) + } + case iceberg.DecimalType: + switch value := value.(type) { + case iceberg.DecimalLiteral: + return scalar.NewDecimal128Scalar(value.Val, arrowType), nil + case iceberg.Decimal: + return scalar.NewDecimal128Scalar(value.Val, arrowType), nil + default: + return nil, fmt.Errorf("unsupported decimal partition value %T", value) + } + } + + return scalar.MakeScalarParam(value, arrowType) +} + +func appendInspectInt64Map(builder *array.MapBuilder, values map[int]int64) { + if values == nil { + builder.AppendNull() + + return + } + builder.Append(true) + keys := builder.KeyBuilder().(*array.Int32Builder) + items := builder.ItemBuilder().(*array.Int64Builder) + ids := make([]int, 0, len(values)) + for id := range values { + ids = append(ids, id) + } + sort.Ints(ids) + for _, id := range ids { + keys.Append(int32(id)) Review Comment: These `int32(id)` conversions (and the ones in `appendInspectInt32List` and `appendInspectOptionalInt32` a bit further down) truncate silently, `int` is 64-bit here, so a column or equality-field ID above `MaxInt32` wraps into a corrupted key with no error. Probably can't happen with real field IDs, but a one-line range guard or a comment noting the spec bound would make that assumption explicit. -- 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]
