laskoviymishka commented on code in PR #1214: URL: https://github.com/apache/iceberg-go/pull/1214#discussion_r3481003325
########## table/data_file_meta.go: ########## @@ -0,0 +1,270 @@ +// 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 ( + "errors" + "fmt" + + "github.com/apache/arrow-go/v18/parquet/metadata" + "github.com/apache/iceberg-go" + tblutils "github.com/apache/iceberg-go/table/internal" +) + +// DataFileArgs collects everything needed to compute a fully populated +// [iceberg.DataFile] (data, equality-delete, or positional-delete) from +// a file's in-memory, format-specific metadata object without performing +// any filesystem reads. +type DataFileArgs struct { + // Schema is the Iceberg schema the file's contents conform to. The + // file's columns MUST carry per-leaf field-ID metadata whose values + // match this schema's field IDs (for Parquet this is the + // "PARQUET:field_id" key on each leaf). + Schema *iceberg.Schema + + // Spec is the partition spec the resulting DataFile is bound to — + // typically tbl.Spec(). The zero value [iceberg.PartitionSpec]{} is + // equivalent to *iceberg.UnpartitionedSpec and is accepted for + // unpartitioned tables. + Spec iceberg.PartitionSpec + + // Format identifies the on-disk file format of the file being + // described and selects the internal FileFormat implementation used + // to extract per-column statistics from Metadata. + Format iceberg.FileFormat + + // Metadata is the format-specific, in-memory file metadata object + // produced by the external writer (for example + // *parquet/metadata.FileMetaData when Format is + // [iceberg.ParquetFile]). + Metadata any + + // FilePath is the fully qualified location the file bytes were + // written to (the same path that will be recorded in the manifest entry). + FilePath string + + // FileSize is the byte length of the object at FilePath. Must be + // greater than zero. + FileSize int64 + + // Content selects which manifest-entry kind to produce — one of + // [iceberg.EntryContentData], [iceberg.EntryContentEqDeletes] or + // [iceberg.EntryContentPosDeletes]. + Content iceberg.ManifestEntryContent + + // PartitionValues maps spec field ID → partition value. Required when Spec + // is partitioned; leave nil/empty for unpartitioned tables. + PartitionValues map[int]any + + // SortOrderID is the sort-order ID recorded on the manifest entry, + // typically tbl.SortOrder().OrderID(). Zero means "unsorted". + SortOrderID int + + // Properties are consulted only for write-side metrics-mode keys + // (write.metadata.metrics.default and per-column + // write.metadata.metrics.column.<name> overrides) that decide + // which columns get full vs truncated vs no statistics. Pass + // tbl.Properties() for table-default behavior; nil is safe. + Properties iceberg.Properties + + // EqualityFieldIDs lists the schema field IDs that an equality + // delete file matches on. Required when + // Content == [iceberg.EntryContentEqDeletes] and MUST be empty + // for every other Content value. Every listed ID must be present in Schema. + EqualityFieldIDs []int + + // FirstRowID is the _row_id assigned to the first row of the data file. + // Set it for v3 data files; it is not required on v1/v2 tables, and + // first_row_id does not apply to delete files. + FirstRowID *int64 +} + +// DataFileFromMetadata builds a fully populated [iceberg.DataFile] from a +// file's in-memory, format-specific metadata object, extracting per-column +// statistics (column_sizes, value_counts, null_value_counts, lower_bounds, +// upper_bounds), split offsets, and the manifest-entry shape required for +// the requested content type. The format-specific extraction is dispatched +// through the internal FileFormat interface; no filesystem reads are +// performed. +// +// The returned DataFile is intended to be handed to +// [Transaction.AddDataFiles] (data files) or [Transaction.NewRowDelta] +// (delete files); it is NOT committed automatically. The file bytes +// described by args.Metadata must already have been uploaded to +// args.FilePath. +func DataFileFromMetadata(args DataFileArgs) (df iceberg.DataFile, err error) { + defer func() { + if r := recover(); r != nil { + df = nil + switch e := r.(type) { + case error: + err = fmt.Errorf("error while building an iceberg datafile: %w", e) + default: + err = fmt.Errorf("error while building an iceberg datafile: %v", e) + } + } + }() + + if args.Schema == nil { + return nil, errors.New("schema is required") + } + if args.Metadata == nil { + return nil, errors.New("file metadata is required") + } + pqMeta, ok := args.Metadata.(*metadata.FileMetaData) + if !ok { + return nil, fmt.Errorf( + "unsupported metadata type: expected *metadata.FileMetaData, got %T", + args.Metadata) + } + if pqMeta == nil { + return nil, errors.New("file metadata is required") + } + if args.FilePath == "" { + return nil, errors.New("file path is required") + } + if args.FileSize <= 0 { + return nil, fmt.Errorf("file size must be greater than 0, got %d", args.FileSize) + } + + switch args.Content { + case iceberg.EntryContentData, + iceberg.EntryContentEqDeletes, + iceberg.EntryContentPosDeletes: + default: + return nil, fmt.Errorf("invalid content: %v", args.Content) + } + + if args.Content == iceberg.EntryContentEqDeletes && len(args.EqualityFieldIDs) == 0 { + return nil, errors.New("EqualityFieldIDs is required for equality-delete files") + } + if args.Content != iceberg.EntryContentEqDeletes && len(args.EqualityFieldIDs) > 0 { + return nil, fmt.Errorf("EqualityFieldIDs must be empty for content %v", args.Content) + } + + if args.Content == iceberg.EntryContentPosDeletes && args.SortOrderID != UnsortedSortOrderID { + return nil, fmt.Errorf("position delete file claims sort order id %d; the spec requires unsorted order id (%d)", + args.SortOrderID, UnsortedSortOrderID) + } + + if err := validateContentSchema(args); err != nil { + return nil, err + } + + if err := validatePartitionValues(args.Spec, args.PartitionValues); err != nil { + return nil, err + } + + format := tblutils.GetFileFormat(args.Format) + if format == nil { + return nil, fmt.Errorf("unsupported file format: %s", args.Format) + } + + statsPlan, err := computeStatsPlan(args.Schema, args.Properties) + if err != nil { + return nil, fmt.Errorf("failed to build stats plan: %w", err) + } + + pathMapping, err := format.PathToIDMapping(args.Schema) + if err != nil { + return nil, fmt.Errorf("failed to build path mapping: %w", err) + } + + stats := format.DataFileStatsFromMeta( + args.Metadata, + statsPlan, + pathMapping, + tblutils.VariantFieldIDsFromSchema(args.Schema), + ) + if stats == nil { + return nil, errors.New("failed to build data file stats from metadata") + } + + if len(args.EqualityFieldIDs) > 0 { + stats.EqualityFieldIDs = args.EqualityFieldIDs + } + + partitionValues := args.PartitionValues + if partitionValues == nil { + partitionValues = make(map[int]any) + } + + df = stats.ToDataFile(tblutils.DataFileOpts{ + Schema: args.Schema, + Spec: args.Spec, + Path: args.FilePath, + Format: args.Format, + Content: args.Content, + FileSize: args.FileSize, + PartitionValues: partitionValues, Review Comment: I think there's a real problem here, and the current tests don't catch it because they only exercise files where the partition column is constant. `ToDataFile` (utils.go ~211-237) calls `d.PartitionValue(field, opts.Schema)` and prefers the *stats-inferred* value over the `PartitionValues` we pass in — and that inference panics with "more than one value for partition field" whenever the stats show more than one distinct value for a partition column. So an identity-partitioned file spanning `id=1..10` panics, gets caught by the blanket recover, and comes back as an opaque error — even though the caller supplied the correct partition value. That's exactly the #1186 case: an external writer registering an already-written partitioned file. I'd skip the stats-based `d.PartitionValue` for any field the caller supplied a non-nil value for (or thread a suppress-inference flag through `DataFileOpts`). The existing `fileToDataFile` path supplies no caller values, so it stays unaffected. A test with a non-constant partitioned column would lock this down. wdyt? ########## table/data_file_meta.go: ########## @@ -0,0 +1,270 @@ +// 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 ( + "errors" + "fmt" + + "github.com/apache/arrow-go/v18/parquet/metadata" + "github.com/apache/iceberg-go" + tblutils "github.com/apache/iceberg-go/table/internal" +) + +// DataFileArgs collects everything needed to compute a fully populated +// [iceberg.DataFile] (data, equality-delete, or positional-delete) from +// a file's in-memory, format-specific metadata object without performing +// any filesystem reads. +type DataFileArgs struct { + // Schema is the Iceberg schema the file's contents conform to. The + // file's columns MUST carry per-leaf field-ID metadata whose values + // match this schema's field IDs (for Parquet this is the + // "PARQUET:field_id" key on each leaf). + Schema *iceberg.Schema + + // Spec is the partition spec the resulting DataFile is bound to — + // typically tbl.Spec(). The zero value [iceberg.PartitionSpec]{} is + // equivalent to *iceberg.UnpartitionedSpec and is accepted for + // unpartitioned tables. + Spec iceberg.PartitionSpec + + // Format identifies the on-disk file format of the file being + // described and selects the internal FileFormat implementation used + // to extract per-column statistics from Metadata. + Format iceberg.FileFormat + + // Metadata is the format-specific, in-memory file metadata object + // produced by the external writer (for example + // *parquet/metadata.FileMetaData when Format is + // [iceberg.ParquetFile]). + Metadata any + + // FilePath is the fully qualified location the file bytes were + // written to (the same path that will be recorded in the manifest entry). + FilePath string + + // FileSize is the byte length of the object at FilePath. Must be + // greater than zero. + FileSize int64 + + // Content selects which manifest-entry kind to produce — one of + // [iceberg.EntryContentData], [iceberg.EntryContentEqDeletes] or + // [iceberg.EntryContentPosDeletes]. + Content iceberg.ManifestEntryContent + + // PartitionValues maps spec field ID → partition value. Required when Spec + // is partitioned; leave nil/empty for unpartitioned tables. + PartitionValues map[int]any + + // SortOrderID is the sort-order ID recorded on the manifest entry, + // typically tbl.SortOrder().OrderID(). Zero means "unsorted". + SortOrderID int + + // Properties are consulted only for write-side metrics-mode keys + // (write.metadata.metrics.default and per-column + // write.metadata.metrics.column.<name> overrides) that decide + // which columns get full vs truncated vs no statistics. Pass + // tbl.Properties() for table-default behavior; nil is safe. + Properties iceberg.Properties + + // EqualityFieldIDs lists the schema field IDs that an equality + // delete file matches on. Required when + // Content == [iceberg.EntryContentEqDeletes] and MUST be empty + // for every other Content value. Every listed ID must be present in Schema. + EqualityFieldIDs []int + + // FirstRowID is the _row_id assigned to the first row of the data file. + // Set it for v3 data files; it is not required on v1/v2 tables, and + // first_row_id does not apply to delete files. + FirstRowID *int64 +} + +// DataFileFromMetadata builds a fully populated [iceberg.DataFile] from a +// file's in-memory, format-specific metadata object, extracting per-column +// statistics (column_sizes, value_counts, null_value_counts, lower_bounds, +// upper_bounds), split offsets, and the manifest-entry shape required for +// the requested content type. The format-specific extraction is dispatched +// through the internal FileFormat interface; no filesystem reads are +// performed. +// +// The returned DataFile is intended to be handed to +// [Transaction.AddDataFiles] (data files) or [Transaction.NewRowDelta] +// (delete files); it is NOT committed automatically. The file bytes +// described by args.Metadata must already have been uploaded to +// args.FilePath. +func DataFileFromMetadata(args DataFileArgs) (df iceberg.DataFile, err error) { + defer func() { + if r := recover(); r != nil { + df = nil + switch e := r.(type) { + case error: + err = fmt.Errorf("error while building an iceberg datafile: %w", e) + default: + err = fmt.Errorf("error while building an iceberg datafile: %v", e) + } + } + }() + + if args.Schema == nil { + return nil, errors.New("schema is required") + } + if args.Metadata == nil { + return nil, errors.New("file metadata is required") + } + pqMeta, ok := args.Metadata.(*metadata.FileMetaData) + if !ok { + return nil, fmt.Errorf( + "unsupported metadata type: expected *metadata.FileMetaData, got %T", + args.Metadata) + } + if pqMeta == nil { + return nil, errors.New("file metadata is required") + } + if args.FilePath == "" { + return nil, errors.New("file path is required") + } + if args.FileSize <= 0 { + return nil, fmt.Errorf("file size must be greater than 0, got %d", args.FileSize) + } + + switch args.Content { + case iceberg.EntryContentData, + iceberg.EntryContentEqDeletes, + iceberg.EntryContentPosDeletes: + default: + return nil, fmt.Errorf("invalid content: %v", args.Content) + } + + if args.Content == iceberg.EntryContentEqDeletes && len(args.EqualityFieldIDs) == 0 { + return nil, errors.New("EqualityFieldIDs is required for equality-delete files") + } + if args.Content != iceberg.EntryContentEqDeletes && len(args.EqualityFieldIDs) > 0 { + return nil, fmt.Errorf("EqualityFieldIDs must be empty for content %v", args.Content) + } + + if args.Content == iceberg.EntryContentPosDeletes && args.SortOrderID != UnsortedSortOrderID { + return nil, fmt.Errorf("position delete file claims sort order id %d; the spec requires unsorted order id (%d)", + args.SortOrderID, UnsortedSortOrderID) + } + + if err := validateContentSchema(args); err != nil { + return nil, err + } + + if err := validatePartitionValues(args.Spec, args.PartitionValues); err != nil { + return nil, err + } + + format := tblutils.GetFileFormat(args.Format) + if format == nil { + return nil, fmt.Errorf("unsupported file format: %s", args.Format) + } + + statsPlan, err := computeStatsPlan(args.Schema, args.Properties) + if err != nil { + return nil, fmt.Errorf("failed to build stats plan: %w", err) + } + + pathMapping, err := format.PathToIDMapping(args.Schema) + if err != nil { + return nil, fmt.Errorf("failed to build path mapping: %w", err) + } + + stats := format.DataFileStatsFromMeta( + args.Metadata, + statsPlan, + pathMapping, + tblutils.VariantFieldIDsFromSchema(args.Schema), + ) + if stats == nil { + return nil, errors.New("failed to build data file stats from metadata") + } + + if len(args.EqualityFieldIDs) > 0 { + stats.EqualityFieldIDs = args.EqualityFieldIDs + } + + partitionValues := args.PartitionValues + if partitionValues == nil { + partitionValues = make(map[int]any) + } + + df = stats.ToDataFile(tblutils.DataFileOpts{ + Schema: args.Schema, + Spec: args.Spec, + Path: args.FilePath, + Format: args.Format, + Content: args.Content, + FileSize: args.FileSize, + PartitionValues: partitionValues, + SortOrderID: args.SortOrderID, + FirstRowID: args.FirstRowID, Review Comment: Two spec problems here, both from §First Row ID Inheritance, and the field doc ("Set it for v3 data files") points callers the wrong way. For delete files, `first_row_id` must always be null — and nothing rejects a non-nil `FirstRowID` on eq/pos-delete content; it's plumbed straight through. For *newly written* data files it must also be null: "its first_row_id field is set to null because it is not assigned until the snapshot is successfully committed" — row IDs come from inheritance at commit. Since this helper produces ADDED-status entries for `Transaction.AddDataFiles`, the spec-correct value for the normal case is nil, and hand-supplying one risks drifting from the table's `next-row-id` counter under concurrent commits. I'd gate `bldr.FirstRowID(...)` on `opts.Content == EntryContentData` over in `table/internal/utils.go` (the `if opts.FirstRowID != nil` block), reject a non-nil `FirstRowID` on delete content here, and reword the doc to: "nil for newly-written files — row IDs are assigned by inheritance at commit; set only when re-registering an existing file whose first_row_id was previously assigned; always nil for delete files." Worth asking whether exposing `FirstRowID` at all is right for an append-oriented helper. ########## table/data_file_meta.go: ########## @@ -0,0 +1,270 @@ +// 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 ( + "errors" + "fmt" + + "github.com/apache/arrow-go/v18/parquet/metadata" + "github.com/apache/iceberg-go" + tblutils "github.com/apache/iceberg-go/table/internal" +) + +// DataFileArgs collects everything needed to compute a fully populated +// [iceberg.DataFile] (data, equality-delete, or positional-delete) from +// a file's in-memory, format-specific metadata object without performing +// any filesystem reads. +type DataFileArgs struct { + // Schema is the Iceberg schema the file's contents conform to. The + // file's columns MUST carry per-leaf field-ID metadata whose values + // match this schema's field IDs (for Parquet this is the + // "PARQUET:field_id" key on each leaf). + Schema *iceberg.Schema + + // Spec is the partition spec the resulting DataFile is bound to — + // typically tbl.Spec(). The zero value [iceberg.PartitionSpec]{} is + // equivalent to *iceberg.UnpartitionedSpec and is accepted for + // unpartitioned tables. + Spec iceberg.PartitionSpec + + // Format identifies the on-disk file format of the file being + // described and selects the internal FileFormat implementation used + // to extract per-column statistics from Metadata. + Format iceberg.FileFormat + + // Metadata is the format-specific, in-memory file metadata object + // produced by the external writer (for example + // *parquet/metadata.FileMetaData when Format is + // [iceberg.ParquetFile]). + Metadata any Review Comment: `Format` and `Metadata` are two fields that have to agree, and the only thing enforcing that is the runtime type-assert below — a caller passing `ParquetFile` plus an Avro metadata object compiles fine and fails at call time. Since Parquet is the only supported format today, `any` isn't buying us real generality yet. I'd lean toward typing it now (`Metadata *metadata.FileMetaData`, and drop or default `Format`) while there are no callers to break. If we want `any` for forward-compat, that's defensible, but then I'd document the Format↔Metadata invariant much more loudly right here. wdyt? ########## table/data_file_meta.go: ########## @@ -0,0 +1,270 @@ +// 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 ( + "errors" + "fmt" + + "github.com/apache/arrow-go/v18/parquet/metadata" + "github.com/apache/iceberg-go" + tblutils "github.com/apache/iceberg-go/table/internal" +) + +// DataFileArgs collects everything needed to compute a fully populated +// [iceberg.DataFile] (data, equality-delete, or positional-delete) from +// a file's in-memory, format-specific metadata object without performing +// any filesystem reads. +type DataFileArgs struct { + // Schema is the Iceberg schema the file's contents conform to. The + // file's columns MUST carry per-leaf field-ID metadata whose values + // match this schema's field IDs (for Parquet this is the + // "PARQUET:field_id" key on each leaf). + Schema *iceberg.Schema + + // Spec is the partition spec the resulting DataFile is bound to — + // typically tbl.Spec(). The zero value [iceberg.PartitionSpec]{} is + // equivalent to *iceberg.UnpartitionedSpec and is accepted for + // unpartitioned tables. + Spec iceberg.PartitionSpec + + // Format identifies the on-disk file format of the file being + // described and selects the internal FileFormat implementation used + // to extract per-column statistics from Metadata. + Format iceberg.FileFormat + + // Metadata is the format-specific, in-memory file metadata object + // produced by the external writer (for example + // *parquet/metadata.FileMetaData when Format is + // [iceberg.ParquetFile]). + Metadata any + + // FilePath is the fully qualified location the file bytes were + // written to (the same path that will be recorded in the manifest entry). + FilePath string + + // FileSize is the byte length of the object at FilePath. Must be + // greater than zero. + FileSize int64 + + // Content selects which manifest-entry kind to produce — one of + // [iceberg.EntryContentData], [iceberg.EntryContentEqDeletes] or + // [iceberg.EntryContentPosDeletes]. + Content iceberg.ManifestEntryContent + + // PartitionValues maps spec field ID → partition value. Required when Spec + // is partitioned; leave nil/empty for unpartitioned tables. + PartitionValues map[int]any + + // SortOrderID is the sort-order ID recorded on the manifest entry, + // typically tbl.SortOrder().OrderID(). Zero means "unsorted". + SortOrderID int + + // Properties are consulted only for write-side metrics-mode keys + // (write.metadata.metrics.default and per-column + // write.metadata.metrics.column.<name> overrides) that decide + // which columns get full vs truncated vs no statistics. Pass + // tbl.Properties() for table-default behavior; nil is safe. + Properties iceberg.Properties + + // EqualityFieldIDs lists the schema field IDs that an equality + // delete file matches on. Required when + // Content == [iceberg.EntryContentEqDeletes] and MUST be empty + // for every other Content value. Every listed ID must be present in Schema. + EqualityFieldIDs []int + + // FirstRowID is the _row_id assigned to the first row of the data file. + // Set it for v3 data files; it is not required on v1/v2 tables, and + // first_row_id does not apply to delete files. + FirstRowID *int64 +} + +// DataFileFromMetadata builds a fully populated [iceberg.DataFile] from a +// file's in-memory, format-specific metadata object, extracting per-column +// statistics (column_sizes, value_counts, null_value_counts, lower_bounds, +// upper_bounds), split offsets, and the manifest-entry shape required for +// the requested content type. The format-specific extraction is dispatched +// through the internal FileFormat interface; no filesystem reads are +// performed. +// +// The returned DataFile is intended to be handed to +// [Transaction.AddDataFiles] (data files) or [Transaction.NewRowDelta] +// (delete files); it is NOT committed automatically. The file bytes +// described by args.Metadata must already have been uploaded to +// args.FilePath. +func DataFileFromMetadata(args DataFileArgs) (df iceberg.DataFile, err error) { + defer func() { + if r := recover(); r != nil { + df = nil + switch e := r.(type) { + case error: + err = fmt.Errorf("error while building an iceberg datafile: %w", e) Review Comment: Wrapping the entire function in `recover` means a genuine bug in this package's own code (a nil-map write, an index-out-of-range) gets laundered into a generic "error while building" rather than surfacing as the panic it is. It's also currently masking the two real, reachable input failures (the partition-override panic, and the missing data-schema validation in `validateContentSchema`). Once those are validated up front, I'd either drop the blanket recover or scope it tightly to the `DataFileStatsFromMeta` + `ToDataFile` calls, with a comment naming which call panics and why. Two small things while you're in here: `df = nil` is redundant (the named return is already nil), and the message wording ("error while building an iceberg datafile") diverges from the package's "error encountered during …" pattern in `arrow_utils.go` / `equality_delete_writer.go`. ########## table/data_file_meta.go: ########## @@ -0,0 +1,270 @@ +// 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 ( + "errors" + "fmt" + + "github.com/apache/arrow-go/v18/parquet/metadata" + "github.com/apache/iceberg-go" + tblutils "github.com/apache/iceberg-go/table/internal" +) + +// DataFileArgs collects everything needed to compute a fully populated +// [iceberg.DataFile] (data, equality-delete, or positional-delete) from +// a file's in-memory, format-specific metadata object without performing +// any filesystem reads. +type DataFileArgs struct { + // Schema is the Iceberg schema the file's contents conform to. The + // file's columns MUST carry per-leaf field-ID metadata whose values + // match this schema's field IDs (for Parquet this is the + // "PARQUET:field_id" key on each leaf). + Schema *iceberg.Schema + + // Spec is the partition spec the resulting DataFile is bound to — + // typically tbl.Spec(). The zero value [iceberg.PartitionSpec]{} is + // equivalent to *iceberg.UnpartitionedSpec and is accepted for + // unpartitioned tables. + Spec iceberg.PartitionSpec + + // Format identifies the on-disk file format of the file being + // described and selects the internal FileFormat implementation used + // to extract per-column statistics from Metadata. + Format iceberg.FileFormat + + // Metadata is the format-specific, in-memory file metadata object + // produced by the external writer (for example + // *parquet/metadata.FileMetaData when Format is + // [iceberg.ParquetFile]). + Metadata any + + // FilePath is the fully qualified location the file bytes were + // written to (the same path that will be recorded in the manifest entry). + FilePath string + + // FileSize is the byte length of the object at FilePath. Must be + // greater than zero. + FileSize int64 + + // Content selects which manifest-entry kind to produce — one of + // [iceberg.EntryContentData], [iceberg.EntryContentEqDeletes] or + // [iceberg.EntryContentPosDeletes]. + Content iceberg.ManifestEntryContent + + // PartitionValues maps spec field ID → partition value. Required when Spec + // is partitioned; leave nil/empty for unpartitioned tables. + PartitionValues map[int]any + + // SortOrderID is the sort-order ID recorded on the manifest entry, + // typically tbl.SortOrder().OrderID(). Zero means "unsorted". + SortOrderID int + + // Properties are consulted only for write-side metrics-mode keys + // (write.metadata.metrics.default and per-column + // write.metadata.metrics.column.<name> overrides) that decide + // which columns get full vs truncated vs no statistics. Pass + // tbl.Properties() for table-default behavior; nil is safe. + Properties iceberg.Properties + + // EqualityFieldIDs lists the schema field IDs that an equality + // delete file matches on. Required when + // Content == [iceberg.EntryContentEqDeletes] and MUST be empty + // for every other Content value. Every listed ID must be present in Schema. + EqualityFieldIDs []int + + // FirstRowID is the _row_id assigned to the first row of the data file. + // Set it for v3 data files; it is not required on v1/v2 tables, and + // first_row_id does not apply to delete files. + FirstRowID *int64 +} + +// DataFileFromMetadata builds a fully populated [iceberg.DataFile] from a +// file's in-memory, format-specific metadata object, extracting per-column +// statistics (column_sizes, value_counts, null_value_counts, lower_bounds, +// upper_bounds), split offsets, and the manifest-entry shape required for +// the requested content type. The format-specific extraction is dispatched +// through the internal FileFormat interface; no filesystem reads are +// performed. +// +// The returned DataFile is intended to be handed to +// [Transaction.AddDataFiles] (data files) or [Transaction.NewRowDelta] +// (delete files); it is NOT committed automatically. The file bytes +// described by args.Metadata must already have been uploaded to +// args.FilePath. +func DataFileFromMetadata(args DataFileArgs) (df iceberg.DataFile, err error) { + defer func() { + if r := recover(); r != nil { + df = nil + switch e := r.(type) { + case error: + err = fmt.Errorf("error while building an iceberg datafile: %w", e) + default: + err = fmt.Errorf("error while building an iceberg datafile: %v", e) + } + } + }() + + if args.Schema == nil { + return nil, errors.New("schema is required") + } + if args.Metadata == nil { + return nil, errors.New("file metadata is required") + } + pqMeta, ok := args.Metadata.(*metadata.FileMetaData) + if !ok { + return nil, fmt.Errorf( + "unsupported metadata type: expected *metadata.FileMetaData, got %T", + args.Metadata) + } + if pqMeta == nil { + return nil, errors.New("file metadata is required") + } + if args.FilePath == "" { + return nil, errors.New("file path is required") + } + if args.FileSize <= 0 { + return nil, fmt.Errorf("file size must be greater than 0, got %d", args.FileSize) + } + + switch args.Content { + case iceberg.EntryContentData, + iceberg.EntryContentEqDeletes, + iceberg.EntryContentPosDeletes: + default: + return nil, fmt.Errorf("invalid content: %v", args.Content) + } + + if args.Content == iceberg.EntryContentEqDeletes && len(args.EqualityFieldIDs) == 0 { + return nil, errors.New("EqualityFieldIDs is required for equality-delete files") + } + if args.Content != iceberg.EntryContentEqDeletes && len(args.EqualityFieldIDs) > 0 { + return nil, fmt.Errorf("EqualityFieldIDs must be empty for content %v", args.Content) + } + + if args.Content == iceberg.EntryContentPosDeletes && args.SortOrderID != UnsortedSortOrderID { + return nil, fmt.Errorf("position delete file claims sort order id %d; the spec requires unsorted order id (%d)", + args.SortOrderID, UnsortedSortOrderID) + } + + if err := validateContentSchema(args); err != nil { + return nil, err + } + + if err := validatePartitionValues(args.Spec, args.PartitionValues); err != nil { + return nil, err + } + + format := tblutils.GetFileFormat(args.Format) + if format == nil { + return nil, fmt.Errorf("unsupported file format: %s", args.Format) + } + + statsPlan, err := computeStatsPlan(args.Schema, args.Properties) + if err != nil { + return nil, fmt.Errorf("failed to build stats plan: %w", err) + } + + pathMapping, err := format.PathToIDMapping(args.Schema) + if err != nil { + return nil, fmt.Errorf("failed to build path mapping: %w", err) + } + + stats := format.DataFileStatsFromMeta( + args.Metadata, + statsPlan, + pathMapping, + tblutils.VariantFieldIDsFromSchema(args.Schema), + ) + if stats == nil { + return nil, errors.New("failed to build data file stats from metadata") + } + + if len(args.EqualityFieldIDs) > 0 { + stats.EqualityFieldIDs = args.EqualityFieldIDs + } + + partitionValues := args.PartitionValues + if partitionValues == nil { + partitionValues = make(map[int]any) + } + + df = stats.ToDataFile(tblutils.DataFileOpts{ + Schema: args.Schema, + Spec: args.Spec, + Path: args.FilePath, + Format: args.Format, + Content: args.Content, + FileSize: args.FileSize, + PartitionValues: partitionValues, + SortOrderID: args.SortOrderID, + FirstRowID: args.FirstRowID, + }) + + return df, nil +} + +func validateContentSchema(args DataFileArgs) error { + switch args.Content { + case iceberg.EntryContentEqDeletes: + for _, id := range args.EqualityFieldIDs { + if _, ok := args.Schema.FindFieldByID(id); !ok { Review Comment: This checks the ID exists but not that it resolves to a leaf primitive. You can't equality-delete on a struct/list/map field, and Java's writers enforce primitivity here. A caller passing the ID of a nested struct field would pass this check and produce an entry other clients reject. I'd tighten this to also assert the resolved field's type is primitive. ########## table/data_file_meta.go: ########## @@ -0,0 +1,270 @@ +// 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 ( + "errors" + "fmt" + + "github.com/apache/arrow-go/v18/parquet/metadata" + "github.com/apache/iceberg-go" + tblutils "github.com/apache/iceberg-go/table/internal" +) + +// DataFileArgs collects everything needed to compute a fully populated +// [iceberg.DataFile] (data, equality-delete, or positional-delete) from +// a file's in-memory, format-specific metadata object without performing +// any filesystem reads. +type DataFileArgs struct { + // Schema is the Iceberg schema the file's contents conform to. The + // file's columns MUST carry per-leaf field-ID metadata whose values + // match this schema's field IDs (for Parquet this is the + // "PARQUET:field_id" key on each leaf). + Schema *iceberg.Schema + + // Spec is the partition spec the resulting DataFile is bound to — + // typically tbl.Spec(). The zero value [iceberg.PartitionSpec]{} is + // equivalent to *iceberg.UnpartitionedSpec and is accepted for + // unpartitioned tables. + Spec iceberg.PartitionSpec + + // Format identifies the on-disk file format of the file being + // described and selects the internal FileFormat implementation used + // to extract per-column statistics from Metadata. + Format iceberg.FileFormat + + // Metadata is the format-specific, in-memory file metadata object + // produced by the external writer (for example + // *parquet/metadata.FileMetaData when Format is + // [iceberg.ParquetFile]). + Metadata any + + // FilePath is the fully qualified location the file bytes were + // written to (the same path that will be recorded in the manifest entry). + FilePath string + + // FileSize is the byte length of the object at FilePath. Must be + // greater than zero. + FileSize int64 + + // Content selects which manifest-entry kind to produce — one of + // [iceberg.EntryContentData], [iceberg.EntryContentEqDeletes] or + // [iceberg.EntryContentPosDeletes]. + Content iceberg.ManifestEntryContent + + // PartitionValues maps spec field ID → partition value. Required when Spec + // is partitioned; leave nil/empty for unpartitioned tables. + PartitionValues map[int]any + + // SortOrderID is the sort-order ID recorded on the manifest entry, + // typically tbl.SortOrder().OrderID(). Zero means "unsorted". + SortOrderID int + + // Properties are consulted only for write-side metrics-mode keys + // (write.metadata.metrics.default and per-column + // write.metadata.metrics.column.<name> overrides) that decide + // which columns get full vs truncated vs no statistics. Pass + // tbl.Properties() for table-default behavior; nil is safe. + Properties iceberg.Properties + + // EqualityFieldIDs lists the schema field IDs that an equality + // delete file matches on. Required when + // Content == [iceberg.EntryContentEqDeletes] and MUST be empty + // for every other Content value. Every listed ID must be present in Schema. + EqualityFieldIDs []int + + // FirstRowID is the _row_id assigned to the first row of the data file. + // Set it for v3 data files; it is not required on v1/v2 tables, and + // first_row_id does not apply to delete files. + FirstRowID *int64 +} + +// DataFileFromMetadata builds a fully populated [iceberg.DataFile] from a +// file's in-memory, format-specific metadata object, extracting per-column +// statistics (column_sizes, value_counts, null_value_counts, lower_bounds, +// upper_bounds), split offsets, and the manifest-entry shape required for +// the requested content type. The format-specific extraction is dispatched +// through the internal FileFormat interface; no filesystem reads are +// performed. +// +// The returned DataFile is intended to be handed to +// [Transaction.AddDataFiles] (data files) or [Transaction.NewRowDelta] +// (delete files); it is NOT committed automatically. The file bytes +// described by args.Metadata must already have been uploaded to +// args.FilePath. +func DataFileFromMetadata(args DataFileArgs) (df iceberg.DataFile, err error) { + defer func() { + if r := recover(); r != nil { + df = nil + switch e := r.(type) { + case error: + err = fmt.Errorf("error while building an iceberg datafile: %w", e) + default: + err = fmt.Errorf("error while building an iceberg datafile: %v", e) + } + } + }() + + if args.Schema == nil { + return nil, errors.New("schema is required") + } + if args.Metadata == nil { + return nil, errors.New("file metadata is required") + } + pqMeta, ok := args.Metadata.(*metadata.FileMetaData) + if !ok { + return nil, fmt.Errorf( + "unsupported metadata type: expected *metadata.FileMetaData, got %T", + args.Metadata) + } + if pqMeta == nil { + return nil, errors.New("file metadata is required") + } + if args.FilePath == "" { + return nil, errors.New("file path is required") + } + if args.FileSize <= 0 { + return nil, fmt.Errorf("file size must be greater than 0, got %d", args.FileSize) + } + + switch args.Content { + case iceberg.EntryContentData, + iceberg.EntryContentEqDeletes, + iceberg.EntryContentPosDeletes: + default: + return nil, fmt.Errorf("invalid content: %v", args.Content) + } + + if args.Content == iceberg.EntryContentEqDeletes && len(args.EqualityFieldIDs) == 0 { + return nil, errors.New("EqualityFieldIDs is required for equality-delete files") + } + if args.Content != iceberg.EntryContentEqDeletes && len(args.EqualityFieldIDs) > 0 { + return nil, fmt.Errorf("EqualityFieldIDs must be empty for content %v", args.Content) + } + + if args.Content == iceberg.EntryContentPosDeletes && args.SortOrderID != UnsortedSortOrderID { + return nil, fmt.Errorf("position delete file claims sort order id %d; the spec requires unsorted order id (%d)", + args.SortOrderID, UnsortedSortOrderID) + } + + if err := validateContentSchema(args); err != nil { + return nil, err + } + + if err := validatePartitionValues(args.Spec, args.PartitionValues); err != nil { + return nil, err + } + + format := tblutils.GetFileFormat(args.Format) + if format == nil { + return nil, fmt.Errorf("unsupported file format: %s", args.Format) + } + + statsPlan, err := computeStatsPlan(args.Schema, args.Properties) + if err != nil { + return nil, fmt.Errorf("failed to build stats plan: %w", err) + } + + pathMapping, err := format.PathToIDMapping(args.Schema) + if err != nil { + return nil, fmt.Errorf("failed to build path mapping: %w", err) + } + + stats := format.DataFileStatsFromMeta( + args.Metadata, + statsPlan, + pathMapping, + tblutils.VariantFieldIDsFromSchema(args.Schema), + ) + if stats == nil { + return nil, errors.New("failed to build data file stats from metadata") + } + + if len(args.EqualityFieldIDs) > 0 { + stats.EqualityFieldIDs = args.EqualityFieldIDs + } + + partitionValues := args.PartitionValues + if partitionValues == nil { + partitionValues = make(map[int]any) + } + + df = stats.ToDataFile(tblutils.DataFileOpts{ + Schema: args.Schema, + Spec: args.Spec, + Path: args.FilePath, + Format: args.Format, + Content: args.Content, + FileSize: args.FileSize, + PartitionValues: partitionValues, + SortOrderID: args.SortOrderID, + FirstRowID: args.FirstRowID, + }) + + return df, nil +} + +func validateContentSchema(args DataFileArgs) error { + switch args.Content { + case iceberg.EntryContentEqDeletes: + for _, id := range args.EqualityFieldIDs { + if _, ok := args.Schema.FindFieldByID(id); !ok { + return fmt.Errorf( + "EqualityFieldIDs contains field id %d which is not present in the supplied schema", id) + } + } + case iceberg.EntryContentPosDeletes: Review Comment: There's no way to set `referenced_data_file` here, and no path to the V3 `content_offset` / `content_size_in_bytes` (spec notes 4-5 require these for deletion vectors). So a positional-delete file built through this helper lands without the data-file reference — legal-but-suboptimal for V2 (engines lose the data-file scoping), and not enough if anyone reaches for V3 DV semantics. The PR mentions this was exercised with pos-deletes via Trino — do the entries it produced actually carry `referenced_data_file`, or is Trino just tolerating its absence? I'd either add a `ReferencedDataFile` field plumbed to the builder, or scope this PR to data + equality deletes and split pos-deletes into a follow-up. At minimum the DV scope limit deserves a line in the doc comment. Shipping a half-wired pos-delete path on a public API is the part that's hard to undo. ########## table/data_file_meta.go: ########## @@ -0,0 +1,270 @@ +// 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 ( + "errors" + "fmt" + + "github.com/apache/arrow-go/v18/parquet/metadata" + "github.com/apache/iceberg-go" + tblutils "github.com/apache/iceberg-go/table/internal" +) + +// DataFileArgs collects everything needed to compute a fully populated +// [iceberg.DataFile] (data, equality-delete, or positional-delete) from +// a file's in-memory, format-specific metadata object without performing +// any filesystem reads. +type DataFileArgs struct { + // Schema is the Iceberg schema the file's contents conform to. The + // file's columns MUST carry per-leaf field-ID metadata whose values + // match this schema's field IDs (for Parquet this is the + // "PARQUET:field_id" key on each leaf). + Schema *iceberg.Schema + + // Spec is the partition spec the resulting DataFile is bound to — + // typically tbl.Spec(). The zero value [iceberg.PartitionSpec]{} is + // equivalent to *iceberg.UnpartitionedSpec and is accepted for + // unpartitioned tables. + Spec iceberg.PartitionSpec + + // Format identifies the on-disk file format of the file being + // described and selects the internal FileFormat implementation used + // to extract per-column statistics from Metadata. + Format iceberg.FileFormat + + // Metadata is the format-specific, in-memory file metadata object + // produced by the external writer (for example + // *parquet/metadata.FileMetaData when Format is + // [iceberg.ParquetFile]). + Metadata any + + // FilePath is the fully qualified location the file bytes were + // written to (the same path that will be recorded in the manifest entry). + FilePath string + + // FileSize is the byte length of the object at FilePath. Must be + // greater than zero. + FileSize int64 + + // Content selects which manifest-entry kind to produce — one of + // [iceberg.EntryContentData], [iceberg.EntryContentEqDeletes] or + // [iceberg.EntryContentPosDeletes]. + Content iceberg.ManifestEntryContent + + // PartitionValues maps spec field ID → partition value. Required when Spec + // is partitioned; leave nil/empty for unpartitioned tables. + PartitionValues map[int]any + + // SortOrderID is the sort-order ID recorded on the manifest entry, + // typically tbl.SortOrder().OrderID(). Zero means "unsorted". + SortOrderID int + + // Properties are consulted only for write-side metrics-mode keys + // (write.metadata.metrics.default and per-column + // write.metadata.metrics.column.<name> overrides) that decide + // which columns get full vs truncated vs no statistics. Pass + // tbl.Properties() for table-default behavior; nil is safe. + Properties iceberg.Properties + + // EqualityFieldIDs lists the schema field IDs that an equality + // delete file matches on. Required when + // Content == [iceberg.EntryContentEqDeletes] and MUST be empty + // for every other Content value. Every listed ID must be present in Schema. + EqualityFieldIDs []int + + // FirstRowID is the _row_id assigned to the first row of the data file. + // Set it for v3 data files; it is not required on v1/v2 tables, and + // first_row_id does not apply to delete files. + FirstRowID *int64 +} + +// DataFileFromMetadata builds a fully populated [iceberg.DataFile] from a +// file's in-memory, format-specific metadata object, extracting per-column +// statistics (column_sizes, value_counts, null_value_counts, lower_bounds, +// upper_bounds), split offsets, and the manifest-entry shape required for +// the requested content type. The format-specific extraction is dispatched +// through the internal FileFormat interface; no filesystem reads are +// performed. +// +// The returned DataFile is intended to be handed to +// [Transaction.AddDataFiles] (data files) or [Transaction.NewRowDelta] +// (delete files); it is NOT committed automatically. The file bytes +// described by args.Metadata must already have been uploaded to +// args.FilePath. +func DataFileFromMetadata(args DataFileArgs) (df iceberg.DataFile, err error) { + defer func() { + if r := recover(); r != nil { + df = nil + switch e := r.(type) { + case error: + err = fmt.Errorf("error while building an iceberg datafile: %w", e) + default: + err = fmt.Errorf("error while building an iceberg datafile: %v", e) + } + } + }() + + if args.Schema == nil { + return nil, errors.New("schema is required") + } + if args.Metadata == nil { + return nil, errors.New("file metadata is required") + } + pqMeta, ok := args.Metadata.(*metadata.FileMetaData) + if !ok { + return nil, fmt.Errorf( + "unsupported metadata type: expected *metadata.FileMetaData, got %T", + args.Metadata) + } + if pqMeta == nil { + return nil, errors.New("file metadata is required") + } + if args.FilePath == "" { + return nil, errors.New("file path is required") + } + if args.FileSize <= 0 { + return nil, fmt.Errorf("file size must be greater than 0, got %d", args.FileSize) + } + + switch args.Content { + case iceberg.EntryContentData, + iceberg.EntryContentEqDeletes, + iceberg.EntryContentPosDeletes: + default: + return nil, fmt.Errorf("invalid content: %v", args.Content) + } + + if args.Content == iceberg.EntryContentEqDeletes && len(args.EqualityFieldIDs) == 0 { + return nil, errors.New("EqualityFieldIDs is required for equality-delete files") + } + if args.Content != iceberg.EntryContentEqDeletes && len(args.EqualityFieldIDs) > 0 { + return nil, fmt.Errorf("EqualityFieldIDs must be empty for content %v", args.Content) + } + + if args.Content == iceberg.EntryContentPosDeletes && args.SortOrderID != UnsortedSortOrderID { + return nil, fmt.Errorf("position delete file claims sort order id %d; the spec requires unsorted order id (%d)", + args.SortOrderID, UnsortedSortOrderID) + } + + if err := validateContentSchema(args); err != nil { + return nil, err + } + + if err := validatePartitionValues(args.Spec, args.PartitionValues); err != nil { + return nil, err + } + + format := tblutils.GetFileFormat(args.Format) + if format == nil { + return nil, fmt.Errorf("unsupported file format: %s", args.Format) + } + + statsPlan, err := computeStatsPlan(args.Schema, args.Properties) + if err != nil { + return nil, fmt.Errorf("failed to build stats plan: %w", err) + } + + pathMapping, err := format.PathToIDMapping(args.Schema) + if err != nil { + return nil, fmt.Errorf("failed to build path mapping: %w", err) + } + + stats := format.DataFileStatsFromMeta( + args.Metadata, + statsPlan, + pathMapping, + tblutils.VariantFieldIDsFromSchema(args.Schema), + ) + if stats == nil { + return nil, errors.New("failed to build data file stats from metadata") + } + + if len(args.EqualityFieldIDs) > 0 { + stats.EqualityFieldIDs = args.EqualityFieldIDs + } + + partitionValues := args.PartitionValues + if partitionValues == nil { + partitionValues = make(map[int]any) + } + + df = stats.ToDataFile(tblutils.DataFileOpts{ + Schema: args.Schema, + Spec: args.Spec, + Path: args.FilePath, + Format: args.Format, + Content: args.Content, + FileSize: args.FileSize, + PartitionValues: partitionValues, + SortOrderID: args.SortOrderID, + FirstRowID: args.FirstRowID, + }) + + return df, nil +} + +func validateContentSchema(args DataFileArgs) error { + switch args.Content { + case iceberg.EntryContentEqDeletes: + for _, id := range args.EqualityFieldIDs { + if _, ok := args.Schema.FindFieldByID(id); !ok { + return fmt.Errorf( + "EqualityFieldIDs contains field id %d which is not present in the supplied schema", id) + } + } + case iceberg.EntryContentPosDeletes: + for _, want := range iceberg.PositionalDeleteSchema.Fields() { + got, ok := args.Schema.FindFieldByID(want.ID) + if !ok { + return fmt.Errorf( + "positional delete schema requires reserved field id %d (%s); schema is missing it", want.ID, want.Name) + } + if !got.Type.Equals(want.Type) { + return fmt.Errorf( + "positional delete schema field id %d (%s) must have type %s, got %s", want.ID, want.Name, want.Type, got.Type) + } + } + } + + return nil +} + +func validatePartitionValues(spec iceberg.PartitionSpec, values map[int]any) error { + if spec.Equals(*iceberg.UnpartitionedSpec) { + return nil + } + + expected := make(map[int]string, spec.NumFields()) + for _, field := range spec.Fields() { Review Comment: Separate from the inference panic: once caller-supplied values actually get used, the type of each value starts to matter — an identity transform over an int column wants `int32`, a bucket transform wants the `int32` bucket index, and a void-transform field must be nil. This loop validates presence and rejects extras but doesn't check any of that. Is type-checking the partition tuple something we want to own here, or is that the caller's responsibility? If the latter, a line in the `PartitionValues` doc (including the void→nil rule) would help. wdyt? ########## table/data_file_meta_test.go: ########## @@ -0,0 +1,477 @@ +// 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_test + +import ( + "bytes" + "strings" + "testing" + + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/arrow-go/v18/parquet" + "github.com/apache/arrow-go/v18/parquet/file" + "github.com/apache/arrow-go/v18/parquet/metadata" + "github.com/apache/arrow-go/v18/parquet/pqarrow" + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func parquetMetaFromSchema(t *testing.T, sch *iceberg.Schema, jsonRows string) ([]byte, *metadata.FileMetaData) { + t.Helper() + + arrowSch, err := table.SchemaToArrowSchema(sch, nil, true, false) + require.NoError(t, err) + + rec, _, err := array.RecordFromJSON(memory.DefaultAllocator, arrowSch, strings.NewReader(jsonRows)) + require.NoError(t, err) + defer rec.Release() + + var buf bytes.Buffer + wr, err := pqarrow.NewFileWriter(arrowSch, &buf, + parquet.NewWriterProperties(parquet.WithStats(true)), + pqarrow.DefaultWriterProps()) + require.NoError(t, err) + require.NoError(t, wr.Write(rec)) + require.NoError(t, wr.Close()) + + rdr, err := file.NewParquetReader(bytes.NewReader(buf.Bytes())) + require.NoError(t, err) + defer rdr.Close() + + return buf.Bytes(), rdr.MetaData() +} + +var dataSchema = iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + iceberg.NestedField{ID: 2, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: false}, +) + +const dataRows = `[ + {"id": 1, "name": "alpha"}, + {"id": 2, "name": "beta"}, + {"id": 3, "name": "gamma"} +]` + +func decodeInt32Bound(t *testing.T, raw []byte) int32 { + t.Helper() + lit, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Int32, raw) + require.NoError(t, err) + v, ok := lit.(iceberg.Int32Literal) + require.True(t, ok, "expected Int32Literal, got %T", lit) + + return v.Value() +} + +func decodeStringBound(t *testing.T, raw []byte) string { + t.Helper() + lit, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.String, raw) + require.NoError(t, err) + v, ok := lit.(iceberg.StringLiteral) + require.True(t, ok, "expected StringLiteral, got %T", lit) + + return string(v) +} + +func TestDataFileFromMetadata(t *testing.T) { + unpartitioned := iceberg.NewPartitionSpec() + + type want struct { + content iceberg.ManifestEntryContent + recordCount int64 + equalityFieldIDs []int + boundIDs []int + partition map[int]any + } + + type subtest struct { + name string + sch *iceberg.Schema + rows string + args func(sch *iceberg.Schema, pqBytes []byte, pqMeta *metadata.FileMetaData) table.DataFileArgs + want want + verify func(t *testing.T, df iceberg.DataFile) + } + + cases := []subtest{ + { + name: "data file: full column stats with real bounds", + sch: dataSchema, + rows: dataRows, + args: func(sch *iceberg.Schema, pqBytes []byte, pqMeta *metadata.FileMetaData) table.DataFileArgs { + return table.DataFileArgs{ + Schema: sch, + Spec: unpartitioned, + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/test-data-001.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + } + }, + want: want{ + content: iceberg.EntryContentData, + recordCount: 3, + boundIDs: []int{1, 2}, + }, + verify: func(t *testing.T, df iceberg.DataFile) { + // id column (field 1) lower/upper from the JSON rows. + assert.EqualValues(t, 1, decodeInt32Bound(t, df.LowerBoundValues()[1])) + assert.EqualValues(t, 3, decodeInt32Bound(t, df.UpperBoundValues()[1])) + // name column (field 2) lexicographic min/max are "alpha" / "gamma". + assert.Equal(t, "alpha", decodeStringBound(t, df.LowerBoundValues()[2])) + assert.Equal(t, "gamma", decodeStringBound(t, df.UpperBoundValues()[2])) + assert.Nil(t, df.EqualityFieldIDs(), "data files must not have equality field IDs") + }, + }, + { + name: "equality-delete file: EqualityFieldIDs round-trips", + sch: iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true}, + ), + rows: `[{"id": 10}, {"id": 20}]`, + args: func(sch *iceberg.Schema, pqBytes []byte, pqMeta *metadata.FileMetaData) table.DataFileArgs { + return table.DataFileArgs{ + Schema: sch, + Spec: unpartitioned, + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/eq-del-001.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentEqDeletes, + EqualityFieldIDs: []int{1}, + } + }, + want: want{ + content: iceberg.EntryContentEqDeletes, + recordCount: 2, + equalityFieldIDs: []int{1}, + boundIDs: []int{1}, + }, + verify: func(t *testing.T, df iceberg.DataFile) { + assert.EqualValues(t, 10, decodeInt32Bound(t, df.LowerBoundValues()[1])) + assert.EqualValues(t, 20, decodeInt32Bound(t, df.UpperBoundValues()[1])) + }, + }, + { + name: "partitioned data file: identity transform plumbs partition values", + sch: dataSchema, + rows: `[ + {"id": 7, "name": "alpha"}, + {"id": 7, "name": "beta"} + ]`, + args: func(sch *iceberg.Schema, pqBytes []byte, pqMeta *metadata.FileMetaData) table.DataFileArgs { + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, + FieldID: 1000, + Name: "id", + Transform: iceberg.IdentityTransform{}, + }) + + return table.DataFileArgs{ + Schema: sch, + Spec: spec, + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/id=7/test-data-001.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + PartitionValues: map[int]any{1000: int32(7)}, + } + }, + want: want{ + content: iceberg.EntryContentData, + recordCount: 2, + boundIDs: []int{1, 2}, + partition: map[int]any{1000: int32(7)}, + }, + verify: func(t *testing.T, df iceberg.DataFile) { + assert.EqualValues(t, 7, decodeInt32Bound(t, df.LowerBoundValues()[1])) + assert.EqualValues(t, 7, decodeInt32Bound(t, df.UpperBoundValues()[1])) + }, + }, + { + name: "positional-delete file: reserved field IDs in schema", + sch: iceberg.PositionalDeleteSchema, + rows: `[ + {"file_path": "s3://bucket/data/data-001.parquet", "pos": 0}, + {"file_path": "s3://bucket/data/data-001.parquet", "pos": 5} + ]`, + args: func(sch *iceberg.Schema, pqBytes []byte, pqMeta *metadata.FileMetaData) table.DataFileArgs { + return table.DataFileArgs{ + Schema: sch, + Spec: unpartitioned, + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/pos-del-001.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentPosDeletes, + } + }, + want: want{ + content: iceberg.EntryContentPosDeletes, + recordCount: 2, + boundIDs: []int{2147483545, 2147483546}, + }, + verify: func(t *testing.T, df iceberg.DataFile) { + assert.Nil(t, df.EqualityFieldIDs(), + "positional delete files must not have equality field IDs") + }, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + pqBytes, pqMeta := parquetMetaFromSchema(t, tc.sch, tc.rows) + args := tc.args(tc.sch, pqBytes, pqMeta) + + df, err := table.DataFileFromMetadata(args) + require.NoError(t, err) + require.NotNil(t, df) + + assert.Equal(t, tc.want.content, df.ContentType()) + assert.Equal(t, iceberg.ParquetFile, df.FileFormat()) + assert.Equal(t, args.FilePath, df.FilePath()) + assert.Equal(t, int64(len(pqBytes)), df.FileSizeBytes()) + assert.EqualValues(t, tc.want.recordCount, df.Count()) + assert.Equal(t, tc.want.equalityFieldIDs, df.EqualityFieldIDs()) + + if tc.want.partition != nil { + assert.Equal(t, tc.want.partition, df.Partition(), + "partition data must round-trip from PartitionValues onto DataFile.Partition()") + } + + assert.Len(t, df.ValueCounts(), len(tc.want.boundIDs)) + assert.Len(t, df.LowerBoundValues(), len(tc.want.boundIDs)) + assert.Len(t, df.UpperBoundValues(), len(tc.want.boundIDs)) + for _, id := range tc.want.boundIDs { + assert.Contains(t, df.LowerBoundValues(), id, + "missing lower bound for field id %d", id) + assert.Contains(t, df.UpperBoundValues(), id, + "missing upper bound for field id %d", id) + } + + if tc.verify != nil { + tc.verify(t, df) + } + }) + } +} + +func TestDataFileFromMetadata_FirstRowID(t *testing.T) { + pqBytes, pqMeta := parquetMetaFromSchema(t, dataSchema, dataRows) + firstRowID := int64(42) + + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: dataSchema, + Spec: iceberg.NewPartitionSpec(), + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/v3-data.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + FirstRowID: &firstRowID, + }) + require.NoError(t, err) + require.NotNil(t, df.FirstRowID(), "FirstRowID must be plumbed through for v3 callers") + assert.Equal(t, firstRowID, *df.FirstRowID()) +} + +func TestDataFileFromMetadata_Errors(t *testing.T) { + pqBytes, pqMeta := parquetMetaFromSchema(t, dataSchema, dataRows) + baseSize := int64(len(pqBytes)) + + baseDataArgs := func() table.DataFileArgs { + return table.DataFileArgs{ + Schema: dataSchema, + Spec: iceberg.NewPartitionSpec(), + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/err.parquet", + FileSize: baseSize, + Content: iceberg.EntryContentData, + } + } + + cases := []struct { + name string + mutate func(a *table.DataFileArgs) + wantSub string + }{ + { + name: "nil schema", + mutate: func(a *table.DataFileArgs) { a.Schema = nil }, + wantSub: "schema is required", + }, + { + name: "nil metadata", + mutate: func(a *table.DataFileArgs) { a.Metadata = nil }, + wantSub: "file metadata is required", + }, + { + name: "empty file path", + mutate: func(a *table.DataFileArgs) { a.FilePath = "" }, + wantSub: "file path is required", + }, + { + name: "non-positive file size", + mutate: func(a *table.DataFileArgs) { a.FileSize = 0 }, + wantSub: "file size must be greater than 0", + }, + { + name: "invalid content value", + mutate: func(a *table.DataFileArgs) { a.Content = iceberg.ManifestEntryContent(99) }, + wantSub: "invalid content", + }, + { + name: "equality-delete file with empty EqualityFieldIDs", + mutate: func(a *table.DataFileArgs) { + a.Content = iceberg.EntryContentEqDeletes + a.EqualityFieldIDs = nil + }, + wantSub: "EqualityFieldIDs is required", + }, + { + name: "data file with non-empty EqualityFieldIDs", + mutate: func(a *table.DataFileArgs) { + a.EqualityFieldIDs = []int{1} + }, + wantSub: "EqualityFieldIDs must be empty", + }, + { + name: "positional-delete file with nonzero SortOrderID", + mutate: func(a *table.DataFileArgs) { + a.Schema = iceberg.PositionalDeleteSchema + a.Content = iceberg.EntryContentPosDeletes + a.SortOrderID = 1 + _, posMeta := parquetMetaFromSchema(t, iceberg.PositionalDeleteSchema, + `[{"file_path": "s3://b/d.parquet", "pos": 0}]`) + a.Metadata = posMeta + }, + wantSub: "position delete file claims sort order id 1", + }, + { + name: "equality-delete with unknown field id", + mutate: func(a *table.DataFileArgs) { + a.Content = iceberg.EntryContentEqDeletes + a.EqualityFieldIDs = []int{999} + }, + wantSub: "field id 999", + }, + { + name: "positional-delete schema missing reserved field IDs", + mutate: func(a *table.DataFileArgs) { + a.Content = iceberg.EntryContentPosDeletes + }, + wantSub: "positional delete schema requires reserved field id", + }, + { + name: "partitioned spec missing partition values", + mutate: func(a *table.DataFileArgs) { + a.Spec = iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, + FieldID: 1000, + Name: "id_bucket", + Transform: iceberg.BucketTransform{NumBuckets: 4}, + }) + }, + wantSub: "missing partition value", + }, + { + name: "partition value for unknown field id", + mutate: func(a *table.DataFileArgs) { + a.Spec = iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, + FieldID: 1000, + Name: "id_bucket", + Transform: iceberg.BucketTransform{NumBuckets: 4}, + }) + a.PartitionValues = map[int]any{1000: 0, 1001: 0} + }, + wantSub: "field id 1001", + }, + { + name: "multi-value partition field is recovered into an error", + mutate: func(a *table.DataFileArgs) { + a.Spec = iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, + FieldID: 1000, + Name: "id", + Transform: iceberg.IdentityTransform{}, + }) + a.PartitionValues = map[int]any{1000: int32(1)} + }, + wantSub: "more than one value for partition field", Review Comment: This case is green only because of the blanket recover — it's pinning the recovered-panic string as if it were the contract, and it's the same underlying panic as the partition-override issue. Once that's fixed and partition handling is validated up front, this should flip to asserting on a clean validation error, or become a success case where the caller-supplied value is used without inference. ########## table/data_file_meta.go: ########## @@ -0,0 +1,270 @@ +// 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 ( + "errors" + "fmt" + + "github.com/apache/arrow-go/v18/parquet/metadata" + "github.com/apache/iceberg-go" + tblutils "github.com/apache/iceberg-go/table/internal" +) + +// DataFileArgs collects everything needed to compute a fully populated +// [iceberg.DataFile] (data, equality-delete, or positional-delete) from +// a file's in-memory, format-specific metadata object without performing +// any filesystem reads. +type DataFileArgs struct { + // Schema is the Iceberg schema the file's contents conform to. The + // file's columns MUST carry per-leaf field-ID metadata whose values + // match this schema's field IDs (for Parquet this is the + // "PARQUET:field_id" key on each leaf). + Schema *iceberg.Schema + + // Spec is the partition spec the resulting DataFile is bound to — + // typically tbl.Spec(). The zero value [iceberg.PartitionSpec]{} is + // equivalent to *iceberg.UnpartitionedSpec and is accepted for + // unpartitioned tables. + Spec iceberg.PartitionSpec + + // Format identifies the on-disk file format of the file being + // described and selects the internal FileFormat implementation used + // to extract per-column statistics from Metadata. + Format iceberg.FileFormat + + // Metadata is the format-specific, in-memory file metadata object + // produced by the external writer (for example + // *parquet/metadata.FileMetaData when Format is + // [iceberg.ParquetFile]). + Metadata any + + // FilePath is the fully qualified location the file bytes were + // written to (the same path that will be recorded in the manifest entry). + FilePath string + + // FileSize is the byte length of the object at FilePath. Must be + // greater than zero. + FileSize int64 + + // Content selects which manifest-entry kind to produce — one of + // [iceberg.EntryContentData], [iceberg.EntryContentEqDeletes] or + // [iceberg.EntryContentPosDeletes]. + Content iceberg.ManifestEntryContent + + // PartitionValues maps spec field ID → partition value. Required when Spec + // is partitioned; leave nil/empty for unpartitioned tables. + PartitionValues map[int]any + + // SortOrderID is the sort-order ID recorded on the manifest entry, + // typically tbl.SortOrder().OrderID(). Zero means "unsorted". + SortOrderID int + + // Properties are consulted only for write-side metrics-mode keys + // (write.metadata.metrics.default and per-column + // write.metadata.metrics.column.<name> overrides) that decide + // which columns get full vs truncated vs no statistics. Pass + // tbl.Properties() for table-default behavior; nil is safe. + Properties iceberg.Properties + + // EqualityFieldIDs lists the schema field IDs that an equality + // delete file matches on. Required when + // Content == [iceberg.EntryContentEqDeletes] and MUST be empty + // for every other Content value. Every listed ID must be present in Schema. + EqualityFieldIDs []int + + // FirstRowID is the _row_id assigned to the first row of the data file. + // Set it for v3 data files; it is not required on v1/v2 tables, and + // first_row_id does not apply to delete files. + FirstRowID *int64 +} + +// DataFileFromMetadata builds a fully populated [iceberg.DataFile] from a +// file's in-memory, format-specific metadata object, extracting per-column +// statistics (column_sizes, value_counts, null_value_counts, lower_bounds, +// upper_bounds), split offsets, and the manifest-entry shape required for +// the requested content type. The format-specific extraction is dispatched +// through the internal FileFormat interface; no filesystem reads are +// performed. +// +// The returned DataFile is intended to be handed to +// [Transaction.AddDataFiles] (data files) or [Transaction.NewRowDelta] +// (delete files); it is NOT committed automatically. The file bytes +// described by args.Metadata must already have been uploaded to +// args.FilePath. +func DataFileFromMetadata(args DataFileArgs) (df iceberg.DataFile, err error) { + defer func() { + if r := recover(); r != nil { + df = nil + switch e := r.(type) { + case error: + err = fmt.Errorf("error while building an iceberg datafile: %w", e) + default: + err = fmt.Errorf("error while building an iceberg datafile: %v", e) + } + } + }() + + if args.Schema == nil { + return nil, errors.New("schema is required") + } + if args.Metadata == nil { + return nil, errors.New("file metadata is required") + } + pqMeta, ok := args.Metadata.(*metadata.FileMetaData) + if !ok { + return nil, fmt.Errorf( + "unsupported metadata type: expected *metadata.FileMetaData, got %T", + args.Metadata) + } + if pqMeta == nil { + return nil, errors.New("file metadata is required") + } + if args.FilePath == "" { + return nil, errors.New("file path is required") + } + if args.FileSize <= 0 { + return nil, fmt.Errorf("file size must be greater than 0, got %d", args.FileSize) + } + + switch args.Content { + case iceberg.EntryContentData, + iceberg.EntryContentEqDeletes, + iceberg.EntryContentPosDeletes: + default: + return nil, fmt.Errorf("invalid content: %v", args.Content) + } + + if args.Content == iceberg.EntryContentEqDeletes && len(args.EqualityFieldIDs) == 0 { + return nil, errors.New("EqualityFieldIDs is required for equality-delete files") + } + if args.Content != iceberg.EntryContentEqDeletes && len(args.EqualityFieldIDs) > 0 { + return nil, fmt.Errorf("EqualityFieldIDs must be empty for content %v", args.Content) + } + + if args.Content == iceberg.EntryContentPosDeletes && args.SortOrderID != UnsortedSortOrderID { + return nil, fmt.Errorf("position delete file claims sort order id %d; the spec requires unsorted order id (%d)", + args.SortOrderID, UnsortedSortOrderID) + } + + if err := validateContentSchema(args); err != nil { + return nil, err + } + + if err := validatePartitionValues(args.Spec, args.PartitionValues); err != nil { + return nil, err + } + + format := tblutils.GetFileFormat(args.Format) + if format == nil { + return nil, fmt.Errorf("unsupported file format: %s", args.Format) + } + + statsPlan, err := computeStatsPlan(args.Schema, args.Properties) + if err != nil { + return nil, fmt.Errorf("failed to build stats plan: %w", err) + } + + pathMapping, err := format.PathToIDMapping(args.Schema) + if err != nil { + return nil, fmt.Errorf("failed to build path mapping: %w", err) + } + + stats := format.DataFileStatsFromMeta( + args.Metadata, + statsPlan, + pathMapping, + tblutils.VariantFieldIDsFromSchema(args.Schema), + ) + if stats == nil { + return nil, errors.New("failed to build data file stats from metadata") + } + + if len(args.EqualityFieldIDs) > 0 { + stats.EqualityFieldIDs = args.EqualityFieldIDs + } + + partitionValues := args.PartitionValues + if partitionValues == nil { + partitionValues = make(map[int]any) + } + + df = stats.ToDataFile(tblutils.DataFileOpts{ + Schema: args.Schema, + Spec: args.Spec, + Path: args.FilePath, + Format: args.Format, + Content: args.Content, + FileSize: args.FileSize, + PartitionValues: partitionValues, + SortOrderID: args.SortOrderID, + FirstRowID: args.FirstRowID, + }) + + return df, nil +} + +func validateContentSchema(args DataFileArgs) error { Review Comment: The eq-delete and pos-delete branches validate their schema up front, but there's no `case iceberg.EntryContentData`, so a data file with an empty or mismatched schema slips through to `computeStatsPlan` + `DataFileStatsFromMeta` and panics there ("column chunk %q not found in column mapping" in parquet_files.go) — again laundered through the recover into an opaque error. A schema/Parquet mismatch is a likely caller mistake for this new API, and it's currently untested. I'd add a superset check here (Parquet field IDs ⊇ Iceberg schema field IDs) so it returns a clean error. If the asymmetry is intentional, an explicit `case iceberg.EntryContentData: // no schema constraints` with a one-line comment would at least make it read as deliberate. ########## table/data_file_meta.go: ########## @@ -0,0 +1,270 @@ +// 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 ( + "errors" + "fmt" + + "github.com/apache/arrow-go/v18/parquet/metadata" + "github.com/apache/iceberg-go" + tblutils "github.com/apache/iceberg-go/table/internal" +) + +// DataFileArgs collects everything needed to compute a fully populated +// [iceberg.DataFile] (data, equality-delete, or positional-delete) from +// a file's in-memory, format-specific metadata object without performing +// any filesystem reads. +type DataFileArgs struct { + // Schema is the Iceberg schema the file's contents conform to. The + // file's columns MUST carry per-leaf field-ID metadata whose values + // match this schema's field IDs (for Parquet this is the + // "PARQUET:field_id" key on each leaf). + Schema *iceberg.Schema + + // Spec is the partition spec the resulting DataFile is bound to — + // typically tbl.Spec(). The zero value [iceberg.PartitionSpec]{} is + // equivalent to *iceberg.UnpartitionedSpec and is accepted for + // unpartitioned tables. + Spec iceberg.PartitionSpec + + // Format identifies the on-disk file format of the file being + // described and selects the internal FileFormat implementation used + // to extract per-column statistics from Metadata. + Format iceberg.FileFormat + + // Metadata is the format-specific, in-memory file metadata object + // produced by the external writer (for example + // *parquet/metadata.FileMetaData when Format is + // [iceberg.ParquetFile]). + Metadata any + + // FilePath is the fully qualified location the file bytes were + // written to (the same path that will be recorded in the manifest entry). + FilePath string + + // FileSize is the byte length of the object at FilePath. Must be + // greater than zero. + FileSize int64 + + // Content selects which manifest-entry kind to produce — one of + // [iceberg.EntryContentData], [iceberg.EntryContentEqDeletes] or + // [iceberg.EntryContentPosDeletes]. + Content iceberg.ManifestEntryContent + + // PartitionValues maps spec field ID → partition value. Required when Spec + // is partitioned; leave nil/empty for unpartitioned tables. + PartitionValues map[int]any + + // SortOrderID is the sort-order ID recorded on the manifest entry, + // typically tbl.SortOrder().OrderID(). Zero means "unsorted". + SortOrderID int + + // Properties are consulted only for write-side metrics-mode keys + // (write.metadata.metrics.default and per-column + // write.metadata.metrics.column.<name> overrides) that decide + // which columns get full vs truncated vs no statistics. Pass + // tbl.Properties() for table-default behavior; nil is safe. + Properties iceberg.Properties + + // EqualityFieldIDs lists the schema field IDs that an equality + // delete file matches on. Required when + // Content == [iceberg.EntryContentEqDeletes] and MUST be empty + // for every other Content value. Every listed ID must be present in Schema. + EqualityFieldIDs []int + + // FirstRowID is the _row_id assigned to the first row of the data file. + // Set it for v3 data files; it is not required on v1/v2 tables, and + // first_row_id does not apply to delete files. + FirstRowID *int64 +} + +// DataFileFromMetadata builds a fully populated [iceberg.DataFile] from a +// file's in-memory, format-specific metadata object, extracting per-column +// statistics (column_sizes, value_counts, null_value_counts, lower_bounds, +// upper_bounds), split offsets, and the manifest-entry shape required for +// the requested content type. The format-specific extraction is dispatched +// through the internal FileFormat interface; no filesystem reads are +// performed. +// +// The returned DataFile is intended to be handed to +// [Transaction.AddDataFiles] (data files) or [Transaction.NewRowDelta] +// (delete files); it is NOT committed automatically. The file bytes +// described by args.Metadata must already have been uploaded to +// args.FilePath. +func DataFileFromMetadata(args DataFileArgs) (df iceberg.DataFile, err error) { + defer func() { + if r := recover(); r != nil { + df = nil + switch e := r.(type) { + case error: + err = fmt.Errorf("error while building an iceberg datafile: %w", e) + default: + err = fmt.Errorf("error while building an iceberg datafile: %v", e) + } + } + }() + + if args.Schema == nil { + return nil, errors.New("schema is required") + } + if args.Metadata == nil { + return nil, errors.New("file metadata is required") + } + pqMeta, ok := args.Metadata.(*metadata.FileMetaData) + if !ok { + return nil, fmt.Errorf( + "unsupported metadata type: expected *metadata.FileMetaData, got %T", + args.Metadata) + } + if pqMeta == nil { Review Comment: Both nil-checks are correct and both are reachable: the `args.Metadata == nil` check above only catches a literally-unset field, while a typed-nil `(*metadata.FileMetaData)(nil)` slips past it and gets caught here. Nice that the test already locks this in. I'd add a one-line comment explaining why both exist, otherwise someone will "simplify" one away later. One small thing: both paths return the same "file metadata is required" — a distinct message for the typed-nil case ("metadata pointer is nil") would tell a confused caller which mistake they made. ########## table/data_file_meta_test.go: ########## @@ -0,0 +1,477 @@ +// 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_test + +import ( + "bytes" + "strings" + "testing" + + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/arrow-go/v18/parquet" + "github.com/apache/arrow-go/v18/parquet/file" + "github.com/apache/arrow-go/v18/parquet/metadata" + "github.com/apache/arrow-go/v18/parquet/pqarrow" + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func parquetMetaFromSchema(t *testing.T, sch *iceberg.Schema, jsonRows string) ([]byte, *metadata.FileMetaData) { + t.Helper() + + arrowSch, err := table.SchemaToArrowSchema(sch, nil, true, false) + require.NoError(t, err) + + rec, _, err := array.RecordFromJSON(memory.DefaultAllocator, arrowSch, strings.NewReader(jsonRows)) + require.NoError(t, err) + defer rec.Release() + + var buf bytes.Buffer + wr, err := pqarrow.NewFileWriter(arrowSch, &buf, + parquet.NewWriterProperties(parquet.WithStats(true)), + pqarrow.DefaultWriterProps()) + require.NoError(t, err) + require.NoError(t, wr.Write(rec)) + require.NoError(t, wr.Close()) + + rdr, err := file.NewParquetReader(bytes.NewReader(buf.Bytes())) + require.NoError(t, err) + defer rdr.Close() + + return buf.Bytes(), rdr.MetaData() +} + +var dataSchema = iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: false}, + iceberg.NestedField{ID: 2, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: false}, +) + +const dataRows = `[ + {"id": 1, "name": "alpha"}, + {"id": 2, "name": "beta"}, + {"id": 3, "name": "gamma"} +]` + +func decodeInt32Bound(t *testing.T, raw []byte) int32 { + t.Helper() + lit, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Int32, raw) + require.NoError(t, err) + v, ok := lit.(iceberg.Int32Literal) + require.True(t, ok, "expected Int32Literal, got %T", lit) + + return v.Value() +} + +func decodeStringBound(t *testing.T, raw []byte) string { + t.Helper() + lit, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.String, raw) + require.NoError(t, err) + v, ok := lit.(iceberg.StringLiteral) + require.True(t, ok, "expected StringLiteral, got %T", lit) + + return string(v) +} + +func TestDataFileFromMetadata(t *testing.T) { + unpartitioned := iceberg.NewPartitionSpec() + + type want struct { + content iceberg.ManifestEntryContent + recordCount int64 + equalityFieldIDs []int + boundIDs []int + partition map[int]any + } + + type subtest struct { + name string + sch *iceberg.Schema + rows string + args func(sch *iceberg.Schema, pqBytes []byte, pqMeta *metadata.FileMetaData) table.DataFileArgs + want want + verify func(t *testing.T, df iceberg.DataFile) + } + + cases := []subtest{ + { + name: "data file: full column stats with real bounds", + sch: dataSchema, + rows: dataRows, + args: func(sch *iceberg.Schema, pqBytes []byte, pqMeta *metadata.FileMetaData) table.DataFileArgs { + return table.DataFileArgs{ + Schema: sch, + Spec: unpartitioned, + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/test-data-001.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + } + }, + want: want{ + content: iceberg.EntryContentData, + recordCount: 3, + boundIDs: []int{1, 2}, + }, + verify: func(t *testing.T, df iceberg.DataFile) { + // id column (field 1) lower/upper from the JSON rows. + assert.EqualValues(t, 1, decodeInt32Bound(t, df.LowerBoundValues()[1])) + assert.EqualValues(t, 3, decodeInt32Bound(t, df.UpperBoundValues()[1])) + // name column (field 2) lexicographic min/max are "alpha" / "gamma". + assert.Equal(t, "alpha", decodeStringBound(t, df.LowerBoundValues()[2])) + assert.Equal(t, "gamma", decodeStringBound(t, df.UpperBoundValues()[2])) + assert.Nil(t, df.EqualityFieldIDs(), "data files must not have equality field IDs") + }, + }, + { + name: "equality-delete file: EqualityFieldIDs round-trips", + sch: iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true}, + ), + rows: `[{"id": 10}, {"id": 20}]`, + args: func(sch *iceberg.Schema, pqBytes []byte, pqMeta *metadata.FileMetaData) table.DataFileArgs { + return table.DataFileArgs{ + Schema: sch, + Spec: unpartitioned, + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/eq-del-001.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentEqDeletes, + EqualityFieldIDs: []int{1}, + } + }, + want: want{ + content: iceberg.EntryContentEqDeletes, + recordCount: 2, + equalityFieldIDs: []int{1}, + boundIDs: []int{1}, + }, + verify: func(t *testing.T, df iceberg.DataFile) { + assert.EqualValues(t, 10, decodeInt32Bound(t, df.LowerBoundValues()[1])) + assert.EqualValues(t, 20, decodeInt32Bound(t, df.UpperBoundValues()[1])) + }, + }, + { + name: "partitioned data file: identity transform plumbs partition values", + sch: dataSchema, + rows: `[ + {"id": 7, "name": "alpha"}, + {"id": 7, "name": "beta"} + ]`, + args: func(sch *iceberg.Schema, pqBytes []byte, pqMeta *metadata.FileMetaData) table.DataFileArgs { + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, + FieldID: 1000, + Name: "id", + Transform: iceberg.IdentityTransform{}, + }) + + return table.DataFileArgs{ + Schema: sch, + Spec: spec, + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/id=7/test-data-001.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + PartitionValues: map[int]any{1000: int32(7)}, + } + }, + want: want{ + content: iceberg.EntryContentData, + recordCount: 2, + boundIDs: []int{1, 2}, + partition: map[int]any{1000: int32(7)}, + }, + verify: func(t *testing.T, df iceberg.DataFile) { + assert.EqualValues(t, 7, decodeInt32Bound(t, df.LowerBoundValues()[1])) + assert.EqualValues(t, 7, decodeInt32Bound(t, df.UpperBoundValues()[1])) + }, + }, + { + name: "positional-delete file: reserved field IDs in schema", + sch: iceberg.PositionalDeleteSchema, + rows: `[ + {"file_path": "s3://bucket/data/data-001.parquet", "pos": 0}, + {"file_path": "s3://bucket/data/data-001.parquet", "pos": 5} + ]`, + args: func(sch *iceberg.Schema, pqBytes []byte, pqMeta *metadata.FileMetaData) table.DataFileArgs { + return table.DataFileArgs{ + Schema: sch, + Spec: unpartitioned, + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/pos-del-001.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentPosDeletes, + } + }, + want: want{ + content: iceberg.EntryContentPosDeletes, + recordCount: 2, + boundIDs: []int{2147483545, 2147483546}, + }, + verify: func(t *testing.T, df iceberg.DataFile) { + assert.Nil(t, df.EqualityFieldIDs(), + "positional delete files must not have equality field IDs") + }, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + pqBytes, pqMeta := parquetMetaFromSchema(t, tc.sch, tc.rows) + args := tc.args(tc.sch, pqBytes, pqMeta) + + df, err := table.DataFileFromMetadata(args) + require.NoError(t, err) + require.NotNil(t, df) + + assert.Equal(t, tc.want.content, df.ContentType()) + assert.Equal(t, iceberg.ParquetFile, df.FileFormat()) + assert.Equal(t, args.FilePath, df.FilePath()) + assert.Equal(t, int64(len(pqBytes)), df.FileSizeBytes()) + assert.EqualValues(t, tc.want.recordCount, df.Count()) + assert.Equal(t, tc.want.equalityFieldIDs, df.EqualityFieldIDs()) + + if tc.want.partition != nil { + assert.Equal(t, tc.want.partition, df.Partition(), + "partition data must round-trip from PartitionValues onto DataFile.Partition()") + } + + assert.Len(t, df.ValueCounts(), len(tc.want.boundIDs)) + assert.Len(t, df.LowerBoundValues(), len(tc.want.boundIDs)) + assert.Len(t, df.UpperBoundValues(), len(tc.want.boundIDs)) + for _, id := range tc.want.boundIDs { + assert.Contains(t, df.LowerBoundValues(), id, + "missing lower bound for field id %d", id) + assert.Contains(t, df.UpperBoundValues(), id, + "missing upper bound for field id %d", id) + } + + if tc.verify != nil { + tc.verify(t, df) + } + }) + } +} + +func TestDataFileFromMetadata_FirstRowID(t *testing.T) { + pqBytes, pqMeta := parquetMetaFromSchema(t, dataSchema, dataRows) + firstRowID := int64(42) + + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: dataSchema, + Spec: iceberg.NewPartitionSpec(), + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/v3-data.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + FirstRowID: &firstRowID, + }) + require.NoError(t, err) + require.NotNil(t, df.FirstRowID(), "FirstRowID must be plumbed through for v3 callers") + assert.Equal(t, firstRowID, *df.FirstRowID()) +} + +func TestDataFileFromMetadata_Errors(t *testing.T) { + pqBytes, pqMeta := parquetMetaFromSchema(t, dataSchema, dataRows) + baseSize := int64(len(pqBytes)) + + baseDataArgs := func() table.DataFileArgs { + return table.DataFileArgs{ + Schema: dataSchema, + Spec: iceberg.NewPartitionSpec(), + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/err.parquet", + FileSize: baseSize, + Content: iceberg.EntryContentData, + } + } + + cases := []struct { + name string + mutate func(a *table.DataFileArgs) + wantSub string + }{ + { + name: "nil schema", + mutate: func(a *table.DataFileArgs) { a.Schema = nil }, + wantSub: "schema is required", + }, + { + name: "nil metadata", + mutate: func(a *table.DataFileArgs) { a.Metadata = nil }, + wantSub: "file metadata is required", + }, + { + name: "empty file path", + mutate: func(a *table.DataFileArgs) { a.FilePath = "" }, + wantSub: "file path is required", + }, + { + name: "non-positive file size", + mutate: func(a *table.DataFileArgs) { a.FileSize = 0 }, + wantSub: "file size must be greater than 0", + }, + { + name: "invalid content value", + mutate: func(a *table.DataFileArgs) { a.Content = iceberg.ManifestEntryContent(99) }, + wantSub: "invalid content", + }, + { + name: "equality-delete file with empty EqualityFieldIDs", + mutate: func(a *table.DataFileArgs) { + a.Content = iceberg.EntryContentEqDeletes + a.EqualityFieldIDs = nil + }, + wantSub: "EqualityFieldIDs is required", + }, + { + name: "data file with non-empty EqualityFieldIDs", + mutate: func(a *table.DataFileArgs) { + a.EqualityFieldIDs = []int{1} + }, + wantSub: "EqualityFieldIDs must be empty", + }, + { + name: "positional-delete file with nonzero SortOrderID", + mutate: func(a *table.DataFileArgs) { + a.Schema = iceberg.PositionalDeleteSchema + a.Content = iceberg.EntryContentPosDeletes + a.SortOrderID = 1 + _, posMeta := parquetMetaFromSchema(t, iceberg.PositionalDeleteSchema, + `[{"file_path": "s3://b/d.parquet", "pos": 0}]`) + a.Metadata = posMeta + }, + wantSub: "position delete file claims sort order id 1", + }, + { + name: "equality-delete with unknown field id", + mutate: func(a *table.DataFileArgs) { + a.Content = iceberg.EntryContentEqDeletes + a.EqualityFieldIDs = []int{999} + }, + wantSub: "field id 999", + }, + { + name: "positional-delete schema missing reserved field IDs", + mutate: func(a *table.DataFileArgs) { + a.Content = iceberg.EntryContentPosDeletes + }, + wantSub: "positional delete schema requires reserved field id", + }, + { + name: "partitioned spec missing partition values", + mutate: func(a *table.DataFileArgs) { + a.Spec = iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, + FieldID: 1000, + Name: "id_bucket", + Transform: iceberg.BucketTransform{NumBuckets: 4}, + }) + }, + wantSub: "missing partition value", + }, + { + name: "partition value for unknown field id", + mutate: func(a *table.DataFileArgs) { + a.Spec = iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, + FieldID: 1000, + Name: "id_bucket", + Transform: iceberg.BucketTransform{NumBuckets: 4}, + }) + a.PartitionValues = map[int]any{1000: 0, 1001: 0} + }, + wantSub: "field id 1001", + }, + { + name: "multi-value partition field is recovered into an error", + mutate: func(a *table.DataFileArgs) { + a.Spec = iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, + FieldID: 1000, + Name: "id", + Transform: iceberg.IdentityTransform{}, + }) + a.PartitionValues = map[int]any{1000: int32(1)} + }, + wantSub: "more than one value for partition field", + }, + { + name: "positional-delete schema reserved field has wrong type", + mutate: func(a *table.DataFileArgs) { + a.Content = iceberg.EntryContentPosDeletes + a.Schema = iceberg.NewSchema(0, + iceberg.NestedField{ID: 2147483546, Name: "file_path", Type: iceberg.PrimitiveTypes.String, Required: true}, + iceberg.NestedField{ID: 2147483545, Name: "pos", Type: iceberg.PrimitiveTypes.String, Required: true}, + ) + }, + wantSub: "must have type", + }, + { + name: "invalid metrics mode property fails stats plan", + mutate: func(a *table.DataFileArgs) { + a.Properties = iceberg.Properties{ + table.DefaultWriteMetricsModeKey: "truncate(0)", Review Comment: A few gaps worth closing while the API is new. The only metrics-mode coverage is this error path (`truncate(0)`) — there's no positive case for `truncate(N)` truncating bounds or `none` suppressing stats, and that's exactly where cross-client bound parity lives (Java/PyIceberg default to `truncate(16)`). The happy-path cases also never assert `NullValueCounts` or `ColumnSizes`, and null counts drive pruning. `TestDataFileFromMetadata_FirstRowID` only checks the set case — the v1/v2 nil case (`df.FirstRowID()` should be nil) is worth pinning. There's no multi-row-group fixture, where split offsets and aggregated counts can break silently, and no schema-mismatch case to cover the panic path above. If it's not too much, an end-to-end case that hands the result to `txn.AddDataFiles` / `NewRowDelta` would be the strongest signal the output is actually consumable. (Minor: the substring checks could use `require.ErrorContains`, the repo idiom.) -- 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]
