zeroshade commented on code in PR #1747: URL: https://github.com/apache/iceberg-go/pull/1747#discussion_r3823072284
########## table/inspect_position_deletes.go: ########## @@ -0,0 +1,694 @@ +// 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" + "math" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/scalar" + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + "github.com/apache/iceberg-go/table/dv" + tblutils "github.com/apache/iceberg-go/table/internal" +) + +const ( + positionDeleteFilePathID = math.MaxInt32 - 101 + positionDeletePosID = math.MaxInt32 - 102 + positionDeleteRowID = math.MaxInt32 - 103 + positionDeletePartitionID = math.MaxInt32 - 5 + positionDeleteSpecID = math.MaxInt32 - 4 + positionDeletePhysicalPathID = math.MaxInt32 - 1 + positionDeleteContentOffsetID = math.MaxInt32 - 6 + positionDeleteContentSizeID = math.MaxInt32 - 7 + positionDeletePhysicalPathName = "delete_file_path" +) + +// PositionDeletes returns the individual position-delete records referenced +// by the current snapshot. Parquet position-delete files and V3 deletion +// vectors are exposed through the same schema. +func (i InspectTable) PositionDeletes(ctx context.Context) (array.RecordReader, error) { + partitionType, partitionIDs, err := positionDeletesPartitionType(i.tbl.metadata) + if err != nil { + return nil, fmt.Errorf("inspect position deletes: %w", err) + } + schema := PositionDeletesSchema(i.tbl.metadata.CurrentSchema(), partitionType, i.tbl.metadata.Version()) + arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false) + if err != nil { + return nil, fmt.Errorf("inspect position deletes: build arrow schema: %w", err) + } + + fs, files, err := i.currentPositionDeleteFiles(ctx) + if err != nil { + return nil, fmt.Errorf("inspect position deletes: %w", err) + } + ctx = compute.WithAllocator(ctx, i.alloc) + + return i.positionDeleteRecordReader( + ctx, arrowSchema, fs, files, partitionType, partitionIDs, i.tbl.metadata.Version()), nil +} + +func (i InspectTable) currentPositionDeleteFiles( + ctx context.Context, +) (iceio.IO, []iceberg.DataFile, error) { + snapshot := i.tbl.metadata.CurrentSnapshot() + if snapshot == nil { + return nil, nil, nil + } + if i.tbl.fsF == nil { + return nil, nil, errors.New("table file IO is not configured") + } + fs, err := i.tbl.fsF(ctx) + if err != nil { + return nil, nil, err + } + manifests, err := snapshot.Manifests(fs) + if err != nil { + return nil, nil, err + } + + files := make([]iceberg.DataFile, 0) + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + return nil, nil, err + } + if manifest.ManifestContent() != iceberg.ManifestContentDeletes { + continue + } + for entry, err := range manifest.Entries(fs, true) { + if err != nil { + return nil, nil, fmt.Errorf("read manifest %s: %w", manifest.FilePath(), err) + } + if err := ctx.Err(); err != nil { + return nil, nil, err + } + if entry.DataFile().ContentType() == iceberg.EntryContentPosDeletes { + files = append(files, entry.DataFile()) + } + } + } + + return fs, files, nil +} + +type positionDeleteRecordAppender struct { + filePath *array.StringBuilder + pos *array.Int64Builder + row *array.StructBuilder + partition *inspectPartitionBuilder + specID *array.Int32Builder + deleteFilePath *array.StringBuilder + contentOffset *array.Int64Builder + contentSize *array.Int64Builder + partitionType *iceberg.StructType + partitionIDByOld map[int]int + formatVersion int +} + +func newPositionDeleteRecordAppender( + bldr *array.RecordBuilder, + partitionType *iceberg.StructType, + partitionIDByOld map[int]int, + formatVersion int, +) (positionDeleteRecordAppender, error) { + nextField := 3 + out := positionDeleteRecordAppender{ + filePath: bldr.Field(0).(*array.StringBuilder), + pos: bldr.Field(1).(*array.Int64Builder), + row: bldr.Field(2).(*array.StructBuilder), + partitionType: partitionType, + partitionIDByOld: partitionIDByOld, + formatVersion: formatVersion, + } + if len(partitionType.FieldList) > 0 { + partitionBuilder, err := newInspectPartitionBuilder( + bldr.Field(nextField).(*array.StructBuilder), partitionType) + if err != nil { + return out, err + } + out.partition = partitionBuilder + nextField++ + } + out.specID = bldr.Field(nextField).(*array.Int32Builder) + out.deleteFilePath = bldr.Field(nextField + 1).(*array.StringBuilder) + if formatVersion >= 3 { + out.contentOffset = bldr.Field(nextField + 2).(*array.Int64Builder) + out.contentSize = bldr.Field(nextField + 3).(*array.Int64Builder) + } + + return out, nil +} + +func (a positionDeleteRecordAppender) append( + file iceberg.DataFile, + dataFilePath string, + pos int64, + deletedRow scalar.Scalar, +) error { + a.filePath.Append(dataFilePath) + a.pos.Append(pos) + if err := appendProjectedPositionDeleteRow(a.row, deletedRow); err != nil { + return fmt.Errorf("append deleted row: %w", err) + } + + if a.partition != nil { + partition := make(map[int]any, len(file.Partition())) + for oldID, value := range file.Partition() { + if newID, ok := a.partitionIDByOld[oldID]; ok { + partition[newID] = value + } + } + if err := a.partition.append(partition); err != nil { + return err + } + } + a.specID.Append(file.SpecID()) + a.deleteFilePath.Append(file.FilePath()) + if a.formatVersion >= 3 { + appendInspectOptionalInt64(a.contentOffset, file.ContentOffset()) + appendInspectOptionalInt64(a.contentSize, file.ContentSizeInBytes()) + } + + return nil +} + +// appendPositionDeleteRow projects a row from a position-delete file onto the +// current table schema. Position-delete row structs may contain only a subset +// of the table fields, so matching by Arrow field position or exact type is +// not sufficient. Field IDs are authoritative when the source schema carries +// them; names are used only for readers that do not preserve Parquet metadata. +func appendProjectedPositionDeleteRow(builder *array.StructBuilder, deletedRow scalar.Scalar) error { + if deletedRow == nil || !deletedRow.IsValid() { + builder.AppendNull() + + return nil + } + + row, ok := deletedRow.(*scalar.Struct) + if !ok { + return fmt.Errorf("%w: row has type %s, want struct", iceberg.ErrInvalidSchema, deletedRow.DataType()) + } + + return appendPositionDeleteStruct(builder, row) +} + +func appendPositionDeleteStruct(builder *array.StructBuilder, source *scalar.Struct) error { + sourceType, ok := source.DataType().(*arrow.StructType) + if !ok { + return fmt.Errorf("%w: row has type %s, want struct", iceberg.ErrInvalidSchema, source.DataType()) + } + if len(source.Value) != sourceType.NumFields() { + return fmt.Errorf("%w: row has %d values for %d fields", + iceberg.ErrInvalidSchema, len(source.Value), sourceType.NumFields()) + } + + sourceFields, err := newPositionDeleteFieldLookup(sourceType) + if err != nil { + return err + } + destinationType, ok := builder.Type().(*arrow.StructType) + if !ok { + return fmt.Errorf("%w: destination row has type %s, want struct", iceberg.ErrInvalidSchema, builder.Type()) + } + + if !source.IsValid() { + builder.AppendNull() + + return nil + } + builder.Append(true) + + for index, destinationField := range destinationType.Fields() { + sourceIndex, found := sourceFields.index(destinationField) + if !found { + builder.FieldBuilder(index).AppendNull() + + continue + } + + value := source.Value[sourceIndex] + if value == nil || !value.IsValid() { + builder.FieldBuilder(index).AppendNull() + + continue + } + + fieldBuilder := builder.FieldBuilder(index) + if nestedBuilder, ok := fieldBuilder.(*array.StructBuilder); ok { + nestedValue, ok := value.(*scalar.Struct) + if !ok { + return fmt.Errorf("%w: field %q has type %s, want struct", + iceberg.ErrInvalidSchema, destinationField.Name, value.DataType()) + } + if err := appendPositionDeleteStruct(nestedBuilder, nestedValue); err != nil { + return fmt.Errorf("field %q: %w", destinationField.Name, err) + } + + continue + } + if err := appendPositionDeleteValue(fieldBuilder, value); err != nil { + return fmt.Errorf("field %q: %w", destinationField.Name, err) + } + } + + return nil +} + +func appendPositionDeleteValue(builder array.Builder, value scalar.Scalar) error { + if arrow.TypeEqual(builder.Type(), value.DataType()) { + return scalar.Append(builder, value) + } + if !canPromotePositionDeleteValue(value.DataType(), builder.Type()) { + return scalar.Append(builder, value) Review Comment: This fallback makes valid Iceberg schema evolution fail whenever the promoted value is nested in a list/map, because `scalar.Append` requires exact Arrow type equality. For example, projecting a deleted row written as `list<int>` into a current `list<long>` field reaches this branch and returns `cannot append scalar of type list<int32> to builder for type list<int64>`. The same issue affects promoted map values and evolution inside container structs; decimal precision widening is also valid but is not handled by `canPromotePositionDeleteValue`. Could this projection recurse through list/map children and support all Iceberg-compatible primitive promotions? Please add coverage for at least nested `list<int>` → `list<long>` and decimal precision widening. -- 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]
