laskoviymishka commented on code in PR #1214: URL: https://github.com/apache/iceberg-go/pull/1214#discussion_r3529354229
########## table/data_file_meta.go: ########## @@ -0,0 +1,417 @@ +// 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" + "maps" + "runtime/debug" + + "github.com/apache/arrow-go/v18/parquet/metadata" + "github.com/apache/iceberg-go" + tblutils "github.com/apache/iceberg-go/table/internal" +) + +const referencedDataFilePathFieldID = 2147483546 + +// 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. + // Per-column statistics are matched by leaf-column path: each file + // column's dotted path MUST equal the path derived from this schema's + // field names. Embedded field-id metadata in the file (e.g. the Parquet + // "PARQUET:field_id" key) is not consulted, so a column renamed relative + // to the file will not resolve. + 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 and selects the internal + // FileFormat implementation used to extract per-column statistics from + // Metadata. It must match the concrete type of Metadata: the only supported + // combination today is [iceberg.ParquetFile] with a *metadata.FileMetaData. + Format iceberg.FileFormat + + // Metadata is the format-specific, in-memory file metadata object produced + // by the external writer. + // + // Today only Format == [iceberg.ParquetFile] with *metadata.FileMetaData is supported + 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. Leave the map + // nil/empty for unpartitioned tables, where any supplied value is rejected. + // + // Values must already be the transform results in the expected Go types. + // A field is resolved as follows: + // - key present with a non-nil value: that value is recorded verbatim; + // - key present with a nil value, or key absent: treated as "not supplied" + // and, for order-preserving transforms (identity, truncate, the date/time + // transforms), inferred from the file's column statistics; for + // non-order-preserving transforms (bucket, void) it stays nil. + // + // Void-transform fields may be omitted (they always resolve to nil). + // + // This helper validates only that any supplied field IDs belong to the spec + // (no stray IDs); it does not coerce or type-check the values themselves. + // Callers are responsible for supplying values of the correct type. + 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 + + // FormatVersion is the table's format version + FormatVersion int + + // FirstRowID sets the data file's first_row_id (requires FormatVersion >= 3). + // It must always be nil for delete files, and for v1/v2 data files (the + // field is absent from those schemas). + // + // For v3 data files, whether it is required depends on which consuming API + // the resulting DataFile is handed to: + // - [Transaction.AddDataFiles]: REQUIRED. These are externally-written + // files, so the library cannot fabricate row IDs for them; a nil + // first_row_id is rejected at commit time. + // - [Transaction.ReplaceDataFilesWithDataFiles]: REQUIRED for the added + // files, except when the caller opts into rewrite/compaction semantics + // (WithRewriteSemantics), where the manifest-list writer assigns it by + // inheritance and it may be left nil. + // - [Transaction.NewRowDelta] (via RowDelta.AddRows): NOT required — the + // value is assigned by inheritance from the manifest at commit time, so + // leaving it nil is correct. + // + // In short: set it explicitly whenever the file goes through AddDataFiles + // (or a non-rewrite replace); otherwise leave it nil and let inheritance + // assign it. + FirstRowID *int64 + + // ReferencedDataFile, when non-nil, records the single data file that a + // file-scoped position-delete file applies to. + // + // It is only valid for position delete files. Leave it nil for data files, equality-delete files, + // and partition-scoped position deletes. + // + // For partition-scoped position deletes whose file_path column references more + // than one data file (cardinality > 1) — such a file has no single referenced data file. + ReferencedDataFile *string +} + +// 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. +// +// Scope and limitations: +// - Only Parquet is supported today (args.Format must be +// [iceberg.ParquetFile] and args.Metadata a *metadata.FileMetaData). +// - All three manifest-entry contents are supported end to end: data, +// equality deletes, and positional deletes. +// - Row lineage is supported: a data file's first_row_id can be set via +// DataFileArgs.FirstRowID for format-version-3+ tables. +// - Iceberg v3 deletion vectors (Puffin DVs) are not supported in this +// helper function; it emits only Parquet position-delete files. A +// file-scoped position delete may still record its target via +// DataFileArgs.ReferencedDataFile; partition-scoped deletes leave it nil, +// in which case engines may infer scoping from the file_path column bounds. +func DataFileFromMetadata(args DataFileArgs) (iceberg.DataFile, error) { + if args.Schema == nil { + return nil, errors.New("schema is required") + } + if args.Metadata == nil { + return nil, errors.New("file metadata is required") + } + + format := tblutils.GetFileFormat(args.Format) + if format == nil { + return nil, errors.New("unsupported file format: " + string(args.Format)) + } + + // Both nil checks are intentional: + // - args.Metadata == nil (above) catches an unset interface, while + // - pqMeta == nil below catches a typed-nil (*metadata.FileMetaData)(nil) + pqMeta, ok := args.Metadata.(*metadata.FileMetaData) + if !ok { + return nil, fmt.Errorf( + "unsupported metadata type for format %s: expected *metadata.FileMetaData, got %T", + args.Format, args.Metadata) + } + if pqMeta == nil { + return nil, errors.New("metadata pointer is nil") + } + 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("equality_ids is required for equality-delete files") + } + + if args.Content != iceberg.EntryContentEqDeletes && len(args.EqualityFieldIDs) > 0 { + return nil, fmt.Errorf("equality_ids must be empty for content %v", args.Content) + } + + if args.Content != iceberg.EntryContentData && args.FirstRowID != nil { + return nil, fmt.Errorf("first_row_id must be nil for delete files, got a non-nil value for content %v", args.Content) + } + + if args.FirstRowID != nil && args.FormatVersion < 3 { + return nil, fmt.Errorf( + "first_row_id requires table format version >= 3, got v%d", + args.FormatVersion) + } + + if args.ReferencedDataFile != nil { + if args.Content != iceberg.EntryContentPosDeletes { + return nil, fmt.Errorf("referenced_data_file may only be set for position-delete files, got content %v", args.Content) + } + if *args.ReferencedDataFile == "" { + return nil, errors.New("referenced_data_file must be non-empty when set") + } + } + + if err := validateContentSchema(args); err != nil { + return nil, err + } + + if err := validatePartitionValues(args.Spec, args.PartitionValues); err != nil { + return nil, err + } + + statsProps := args.Properties + if args.ReferencedDataFile != nil { + // A file-scoped position delete must carry exact (untruncated) file_path + // The default truncate(16) metrics mode would clip long paths, thus, forcing + // full metrics for that column. + statsProps = maps.Clone(args.Properties) + if statsProps == nil { + statsProps = iceberg.Properties{} + } + statsProps[MetricsModeColumnConfPrefix+".file_path"] = string(tblutils.MetricModeFull) + } + + statsPlan, err := computeStatsPlan(args.Schema, statsProps) + 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) + } + + partitionValues := args.PartitionValues + if partitionValues == nil { + partitionValues = make(map[int]any) + } + + var df iceberg.DataFile + // DataFileStatsFromMeta and ToDataFile are the only calls wrapped in recover because they may panic on invalid input + // (missing field IDs or builder validation failures) + if err := func() (err error) { + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf( + "error encountered during extracting stats and building the data file: %v\n%s", + r, debug.Stack()) + } + }() + + stats := format.DataFileStatsFromMeta( + args.Metadata, + statsPlan, Review Comment: Non-blocking nit: `debug.Stack()` bakes a raw goroutine stack into an error string that reaches public callers — every other recover site in the package (`arrow_utils.go`, `equality_delete_writer.go`) wraps a clean message without the trace, so I'd drop it for consistency. Separately, `r` here can itself be an error (the `PartitionValue` panic is a `fmt.Errorf` value), so a type-assert + `%w` would preserve `errors.Is/As` for callers. Take it or leave it. ########## table/data_file_meta.go: ########## @@ -0,0 +1,417 @@ +// 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" + "maps" + "runtime/debug" + + "github.com/apache/arrow-go/v18/parquet/metadata" + "github.com/apache/iceberg-go" + tblutils "github.com/apache/iceberg-go/table/internal" +) + +const referencedDataFilePathFieldID = 2147483546 + +// 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. + // Per-column statistics are matched by leaf-column path: each file + // column's dotted path MUST equal the path derived from this schema's + // field names. Embedded field-id metadata in the file (e.g. the Parquet + // "PARQUET:field_id" key) is not consulted, so a column renamed relative + // to the file will not resolve. + 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 and selects the internal + // FileFormat implementation used to extract per-column statistics from + // Metadata. It must match the concrete type of Metadata: the only supported + // combination today is [iceberg.ParquetFile] with a *metadata.FileMetaData. + Format iceberg.FileFormat + + // Metadata is the format-specific, in-memory file metadata object produced + // by the external writer. + // + // Today only Format == [iceberg.ParquetFile] with *metadata.FileMetaData is supported + 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. Leave the map + // nil/empty for unpartitioned tables, where any supplied value is rejected. + // + // Values must already be the transform results in the expected Go types. + // A field is resolved as follows: + // - key present with a non-nil value: that value is recorded verbatim; + // - key present with a nil value, or key absent: treated as "not supplied" + // and, for order-preserving transforms (identity, truncate, the date/time + // transforms), inferred from the file's column statistics; for + // non-order-preserving transforms (bucket, void) it stays nil. + // + // Void-transform fields may be omitted (they always resolve to nil). + // + // This helper validates only that any supplied field IDs belong to the spec + // (no stray IDs); it does not coerce or type-check the values themselves. + // Callers are responsible for supplying values of the correct type. + 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 + + // FormatVersion is the table's format version + FormatVersion int + + // FirstRowID sets the data file's first_row_id (requires FormatVersion >= 3). + // It must always be nil for delete files, and for v1/v2 data files (the + // field is absent from those schemas). + // + // For v3 data files, whether it is required depends on which consuming API + // the resulting DataFile is handed to: + // - [Transaction.AddDataFiles]: REQUIRED. These are externally-written + // files, so the library cannot fabricate row IDs for them; a nil + // first_row_id is rejected at commit time. + // - [Transaction.ReplaceDataFilesWithDataFiles]: REQUIRED for the added + // files, except when the caller opts into rewrite/compaction semantics + // (WithRewriteSemantics), where the manifest-list writer assigns it by + // inheritance and it may be left nil. + // - [Transaction.NewRowDelta] (via RowDelta.AddRows): NOT required — the + // value is assigned by inheritance from the manifest at commit time, so + // leaving it nil is correct. + // + // In short: set it explicitly whenever the file goes through AddDataFiles + // (or a non-rewrite replace); otherwise leave it nil and let inheritance + // assign it. + FirstRowID *int64 + + // ReferencedDataFile, when non-nil, records the single data file that a + // file-scoped position-delete file applies to. + // + // It is only valid for position delete files. Leave it nil for data files, equality-delete files, + // and partition-scoped position deletes. + // + // For partition-scoped position deletes whose file_path column references more + // than one data file (cardinality > 1) — such a file has no single referenced data file. + ReferencedDataFile *string +} + +// 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. +// +// Scope and limitations: +// - Only Parquet is supported today (args.Format must be +// [iceberg.ParquetFile] and args.Metadata a *metadata.FileMetaData). +// - All three manifest-entry contents are supported end to end: data, +// equality deletes, and positional deletes. +// - Row lineage is supported: a data file's first_row_id can be set via +// DataFileArgs.FirstRowID for format-version-3+ tables. +// - Iceberg v3 deletion vectors (Puffin DVs) are not supported in this +// helper function; it emits only Parquet position-delete files. A +// file-scoped position delete may still record its target via +// DataFileArgs.ReferencedDataFile; partition-scoped deletes leave it nil, +// in which case engines may infer scoping from the file_path column bounds. +func DataFileFromMetadata(args DataFileArgs) (iceberg.DataFile, error) { + if args.Schema == nil { + return nil, errors.New("schema is required") + } + if args.Metadata == nil { + return nil, errors.New("file metadata is required") + } + + format := tblutils.GetFileFormat(args.Format) + if format == nil { + return nil, errors.New("unsupported file format: " + string(args.Format)) + } + + // Both nil checks are intentional: + // - args.Metadata == nil (above) catches an unset interface, while + // - pqMeta == nil below catches a typed-nil (*metadata.FileMetaData)(nil) + pqMeta, ok := args.Metadata.(*metadata.FileMetaData) + if !ok { + return nil, fmt.Errorf( + "unsupported metadata type for format %s: expected *metadata.FileMetaData, got %T", + args.Format, args.Metadata) + } + if pqMeta == nil { + return nil, errors.New("metadata pointer is nil") + } + 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("equality_ids is required for equality-delete files") + } + + if args.Content != iceberg.EntryContentEqDeletes && len(args.EqualityFieldIDs) > 0 { + return nil, fmt.Errorf("equality_ids must be empty for content %v", args.Content) + } + + if args.Content != iceberg.EntryContentData && args.FirstRowID != nil { + return nil, fmt.Errorf("first_row_id must be nil for delete files, got a non-nil value for content %v", args.Content) + } + + if args.FirstRowID != nil && args.FormatVersion < 3 { + return nil, fmt.Errorf( + "first_row_id requires table format version >= 3, got v%d", + args.FormatVersion) + } + + if args.ReferencedDataFile != nil { + if args.Content != iceberg.EntryContentPosDeletes { + return nil, fmt.Errorf("referenced_data_file may only be set for position-delete files, got content %v", args.Content) + } + if *args.ReferencedDataFile == "" { + return nil, errors.New("referenced_data_file must be non-empty when set") + } + } + + if err := validateContentSchema(args); err != nil { + return nil, err + } + + if err := validatePartitionValues(args.Spec, args.PartitionValues); err != nil { + return nil, err + } + + statsProps := args.Properties + if args.ReferencedDataFile != nil { + // A file-scoped position delete must carry exact (untruncated) file_path + // The default truncate(16) metrics mode would clip long paths, thus, forcing + // full metrics for that column. + statsProps = maps.Clone(args.Properties) + if statsProps == nil { + statsProps = iceberg.Properties{} + } + statsProps[MetricsModeColumnConfPrefix+".file_path"] = string(tblutils.MetricModeFull) + } + + statsPlan, err := computeStatsPlan(args.Schema, statsProps) + if err != nil { + return nil, fmt.Errorf("failed to build stats plan: %w", err) + } Review Comment: Non-blocking, but cheap to foreclose: this override key is hardcoded to `"file_path"` while the metrics plan resolves the same column via `schema.FindColumnName(fieldID)`. The spec only pins the reserved id and type for the pos-delete path column, not its name — so a schema that names that field anything else silently misses the full-metrics override, falls back to `truncate(16)`, and then `verifyReferencedDataFile` fails for any path over 16 bytes (or passes spuriously for short ones). Resolving via `args.Schema.FindColumnName(referencedDataFilePathFieldID)` sidesteps it. Fine as a follow-up if you'd rather not touch it here. ########## table/data_file_meta_test.go: ########## @@ -0,0 +1,1244 @@ +// 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" + iceio "github.com/apache/iceberg-go/io" + "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") + }, + }, + { + name: "partitioned data file: caller value wins over a non-constant column", + sch: dataSchema, + rows: `[ + {"id": 1, "name": "alpha"}, + {"id": 5, "name": "beta"}, + {"id": 10, "name": "gamma"} + ]`, + 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=42/range.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + PartitionValues: map[int]any{1000: int32(42)}, + } + }, + want: want{ + content: iceberg.EntryContentData, + recordCount: 3, + boundIDs: []int{1, 2}, + partition: map[int]any{1000: int32(42)}, + }, + verify: func(t *testing.T, df iceberg.DataFile) { + assert.EqualValues(t, 1, decodeInt32Bound(t, df.LowerBoundValues()[1])) + assert.EqualValues(t, 10, decodeInt32Bound(t, df.UpperBoundValues()[1])) + }, + }, + } + + 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, + FormatVersion: 3, + 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_FirstRowIDBelowV3Errors(t *testing.T) { + pqBytes, pqMeta := parquetMetaFromSchema(t, dataSchema, dataRows) + firstRowID := int64(42) + + for _, version := range []int{0, 1, 2} { + t.Run("v"+formatVersionStr(version), func(t *testing.T) { + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: dataSchema, + Spec: iceberg.NewPartitionSpec(), + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/pre-v3-data.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + FormatVersion: version, + FirstRowID: &firstRowID, + }) + require.Nil(t, df, "FirstRowID must not be silently dropped below v3") + require.ErrorContains(t, err, "requires table format version >= 3") + }) + } +} + +func TestDataFileFromMetadata_PosDeleteSortOrderAccepted(t *testing.T) { + _, posMeta := parquetMetaFromSchema(t, iceberg.PositionalDeleteSchema, + `[{"file_path": "s3://b/d.parquet", "pos": 0}]`) + + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: iceberg.PositionalDeleteSchema, + Spec: iceberg.NewPartitionSpec(), + Format: iceberg.ParquetFile, + Metadata: posMeta, + FilePath: "s3://bucket/data/pos-del-sorted.parquet", + FileSize: 1024, + Content: iceberg.EntryContentPosDeletes, + SortOrderID: 1, + }) + require.NoError(t, err) + require.NotNil(t, df.SortOrderID()) + assert.Equal(t, 1, *df.SortOrderID()) +} + +func TestDataFileFromMetadata_EqualityFieldFloatRejected(t *testing.T) { + sch := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true}, + iceberg.NestedField{ID: 2, Name: "amount", Type: iceberg.PrimitiveTypes.Float64, Required: false}, + ) + pqBytes, pqMeta := parquetMetaFromSchema(t, sch, `[{"id": 1, "amount": 1.5}]`) + + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: sch, + Spec: iceberg.NewPartitionSpec(), + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/eq-del-float.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentEqDeletes, + EqualityFieldIDs: []int{2}, + }) + require.Nil(t, df, "floating-point equality field must be rejected, matching the writer path") + require.ErrorContains(t, err, "floating-point") +} + +func TestDataFileFromMetadata_VoidPartitionAllowsNil(t *testing.T) { + pqBytes, pqMeta := parquetMetaFromSchema(t, dataSchema, dataRows) + + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, + FieldID: 1000, + Name: "id_void", + Transform: iceberg.VoidTransform{}, + }) + + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: dataSchema, + Spec: spec, + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/void.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + PartitionValues: map[int]any{1000: nil}, + }) + require.NoError(t, err) + require.NotNil(t, df) + + part := df.Partition() + require.Contains(t, part, 1000, "void partition field must be recorded on the DataFile") + assert.Nil(t, part[1000], "void transform partition value must be nil") +} + +func TestDataFileFromMetadata_AbsentPartitionKeyEqualsNil(t *testing.T) { + spec := iceberg.NewPartitionSpec( + iceberg.PartitionField{SourceIDs: []int{1}, FieldID: 1000, Name: "id", Transform: iceberg.IdentityTransform{}}, + iceberg.PartitionField{SourceIDs: []int{1}, FieldID: 1001, Name: "id_bucket", Transform: iceberg.BucketTransform{NumBuckets: 4}}, + ) + + pqBytes, pqMeta := parquetMetaFromSchema(t, dataSchema, `[ + {"id": 7, "name": "alpha"}, + {"id": 7, "name": "beta"} + ]`) + + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: dataSchema, + Spec: spec, + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/id=7/absent.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + PartitionValues: map[int]any{1000: nil}, + }) + require.NoError(t, err, "a present-nil value and an absent key must both be accepted") + require.NotNil(t, df) + + part := df.Partition() + assert.EqualValues(t, 7, part[1000], + "a present-with-nil order-preserving partition key must be inferred from stats") + assert.Nil(t, part[1001], + "an absent non-order-preserving (bucket) partition key must stay nil") +} + +func TestDataFileFromMetadata_NilPartitionValueInferredFromStats(t *testing.T) { + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, + FieldID: 1000, + Name: "id", + Transform: iceberg.IdentityTransform{}, + }) + + t.Run("single-valued column infers the partition value", func(t *testing.T) { + pqBytes, pqMeta := parquetMetaFromSchema(t, dataSchema, `[ + {"id": 7, "name": "alpha"}, + {"id": 7, "name": "beta"} + ]`) + + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: dataSchema, + Spec: spec, + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/id=7/inferred.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + PartitionValues: map[int]any{1000: nil}, + }) + require.NoError(t, err) + assert.EqualValues(t, 7, df.Partition()[1000], + "a nil partition value must be inferred from the file's column statistics when the column is single-valued") + }) + + t.Run("multi-valued column cannot infer and errors", func(t *testing.T) { + pqBytes, pqMeta := parquetMetaFromSchema(t, dataSchema, `[ + {"id": 1, "name": "alpha"}, + {"id": 9, "name": "beta"} + ]`) + + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: dataSchema, + Spec: spec, + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/range.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + PartitionValues: map[int]any{1000: nil}, + }) + require.Nil(t, df, "no DataFile must be returned when inference is ambiguous") + require.ErrorContains(t, err, "cannot infer partition value") + }) +} + +func TestDataFileFromMetadata_ReferencedDataFile(t *testing.T) { + const dataPath = "s3://bucket/data/data-001.parquet" + + _, fileScopedMeta := parquetMetaFromSchema(t, iceberg.PositionalDeleteSchema, + `[{"file_path": "`+dataPath+`", "pos": 0}, + {"file_path": "`+dataPath+`", "pos": 5}]`) + + _, partitionScopedMeta := parquetMetaFromSchema(t, iceberg.PositionalDeleteSchema, + `[{"file_path": "`+dataPath+`", "pos": 0}, + {"file_path": "s3://bucket/data/data-002.parquet", "pos": 3}]`) + + t.Run("file-scoped sets referenced_data_file", func(t *testing.T) { + ref := dataPath + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: iceberg.PositionalDeleteSchema, + Spec: iceberg.NewPartitionSpec(), + Format: iceberg.ParquetFile, + Metadata: fileScopedMeta, + FilePath: "s3://bucket/data/pos-del-001.parquet", + FileSize: 1024, + Content: iceberg.EntryContentPosDeletes, + ReferencedDataFile: &ref, + }) + require.NoError(t, err) + require.NotNil(t, df.ReferencedDataFile()) + assert.Equal(t, dataPath, *df.ReferencedDataFile()) + }) + + t.Run("partition-scoped leaves referenced_data_file nil", func(t *testing.T) { + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: iceberg.PositionalDeleteSchema, + Spec: iceberg.NewPartitionSpec(), + Format: iceberg.ParquetFile, + Metadata: partitionScopedMeta, + FilePath: "s3://bucket/data/pos-del-002.parquet", + FileSize: 1024, + Content: iceberg.EntryContentPosDeletes, + // ReferencedDataFile intentionally nil. + }) + require.NoError(t, err) + assert.Nil(t, df.ReferencedDataFile(), + "a partition-scoped position delete must not record a referenced data file") + }) + + t.Run("ReferencedDataFile that does not scope to a single file is rejected", func(t *testing.T) { + // partitionScopedMeta spans two data files, so its file_path bounds are + // not equal; claiming a single referenced data file must fail. + ref := dataPath + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: iceberg.PositionalDeleteSchema, + Spec: iceberg.NewPartitionSpec(), + Format: iceberg.ParquetFile, + Metadata: partitionScopedMeta, + FilePath: "s3://bucket/data/pos-del-003.parquet", + FileSize: 1024, + Content: iceberg.EntryContentPosDeletes, + ReferencedDataFile: &ref, + }) + require.Nil(t, df, "a delete file spanning multiple data files must not be accepted as file-scoped") + require.ErrorContains(t, err, "does not match the position-delete file_path bounds") + }) + + t.Run("ReferencedDataFile pointing at the wrong data file is rejected", func(t *testing.T) { + // The delete file scopes to dataPath, but the caller claims a different file. + ref := "s3://bucket/data/some-other-file.parquet" + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: iceberg.PositionalDeleteSchema, + Spec: iceberg.NewPartitionSpec(), + Format: iceberg.ParquetFile, + Metadata: fileScopedMeta, + FilePath: "s3://bucket/data/pos-del-004.parquet", + FileSize: 1024, + Content: iceberg.EntryContentPosDeletes, + ReferencedDataFile: &ref, + }) + require.Nil(t, df) + require.ErrorContains(t, err, "does not match the position-delete file_path bounds") + }) +} + +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: "equality_ids is required", + }, + { + name: "data file with non-empty EqualityFieldIDs", + mutate: func(a *table.DataFileArgs) { + a.EqualityFieldIDs = []int{1} + }, + wantSub: "equality_ids must be empty", + }, + { + name: "equality-delete with unknown field id", + mutate: func(a *table.DataFileArgs) { + a.Content = iceberg.EntryContentEqDeletes + a.EqualityFieldIDs = []int{999} + }, + wantSub: "field ID 999 not found in table schema", + }, + { + 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: "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: "partition values supplied for an unpartitioned spec", + mutate: func(a *table.DataFileArgs) { + a.PartitionValues = map[int]any{1000: int32(7)} + }, + wantSub: "partition values supplied for an unpartitioned spec", + }, + { + name: "FirstRowID set on an equality-delete file", + mutate: func(a *table.DataFileArgs) { + a.Content = iceberg.EntryContentEqDeletes + a.EqualityFieldIDs = []int{1} + v := int64(7) + a.FirstRowID = &v + }, + wantSub: "first_row_id must be nil for delete files", + }, + { + name: "ReferencedDataFile set on a data file", + mutate: func(a *table.DataFileArgs) { + ref := "s3://bucket/data/data-001.parquet" + a.ReferencedDataFile = &ref + }, + wantSub: "referenced_data_file may only be set for position-delete files", + }, + { + name: "ReferencedDataFile empty on a positional-delete file", + mutate: func(a *table.DataFileArgs) { + a.Schema = iceberg.PositionalDeleteSchema + a.Content = iceberg.EntryContentPosDeletes + _, posMeta := parquetMetaFromSchema(t, iceberg.PositionalDeleteSchema, + `[{"file_path": "s3://b/d.parquet", "pos": 0}]`) + a.Metadata = posMeta + empty := "" + a.ReferencedDataFile = &empty + }, + wantSub: "referenced_data_file must be non-empty when set", + }, + { + 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)", + } + }, + wantSub: "failed to build stats plan", + }, + { + name: "wrong metadata concrete type", + mutate: func(a *table.DataFileArgs) { + a.Metadata = "not a *metadata.FileMetaData" + }, + wantSub: "unsupported metadata type", + }, + { + name: "typed-nil metadata pointer", + mutate: func(a *table.DataFileArgs) { + a.Metadata = (*metadata.FileMetaData)(nil) + }, + wantSub: "metadata pointer is nil", + }, + { + name: "unsupported file format", + mutate: func(a *table.DataFileArgs) { + a.Format = iceberg.AvroFile + }, + wantSub: "unsupported file format", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + args := baseDataArgs() + tc.mutate(&args) + + df, err := table.DataFileFromMetadata(args) + require.Nil(t, df, "expected no DataFile on error, got %+v", df) + require.ErrorContains(t, err, tc.wantSub) + }) + } +} + +func parquetMetaMultiRowGroup(t *testing.T, sch *iceberg.Schema, jsonRows string, maxRowGroupLen int64) ([]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), parquet.WithMaxRowGroupLength(maxRowGroupLen)), + 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() +} + +func TestDataFileFromMetadata_FirstRowIDNil(t *testing.T) { + pqBytes, pqMeta := parquetMetaFromSchema(t, dataSchema, dataRows) + + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: dataSchema, + Spec: iceberg.NewPartitionSpec(), + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/no-first-row-id.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + }) + require.NoError(t, err) + assert.Nil(t, df.FirstRowID(), + "FirstRowID must be nil when the caller does not supply one (v1/v2 default)") +} + +func TestDataFileFromMetadata_MetricsModes(t *testing.T) { + pqBytes, pqMeta := parquetMetaFromSchema(t, dataSchema, dataRows) + + base := func() table.DataFileArgs { + return table.DataFileArgs{ + Schema: dataSchema, + Spec: iceberg.NewPartitionSpec(), + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/metrics.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + } + } + + t.Run("none suppresses all column stats", func(t *testing.T) { + a := base() + a.Properties = iceberg.Properties{table.DefaultWriteMetricsModeKey: "none"} + df, err := table.DataFileFromMetadata(a) + require.NoError(t, err) + assert.Empty(t, df.LowerBoundValues()) + assert.Empty(t, df.UpperBoundValues()) + assert.Empty(t, df.ValueCounts()) + assert.Empty(t, df.NullValueCounts()) + assert.EqualValues(t, 3, df.Count(), "record count is independent of metrics mode") + }) + + t.Run("counts keeps counts but drops bounds", func(t *testing.T) { + a := base() + a.Properties = iceberg.Properties{table.DefaultWriteMetricsModeKey: "counts"} + df, err := table.DataFileFromMetadata(a) + require.NoError(t, err) + assert.Empty(t, df.LowerBoundValues()) + assert.Empty(t, df.UpperBoundValues()) + assert.NotEmpty(t, df.ValueCounts()) + }) + + t.Run("truncate(2) truncates string bounds, keeps numeric bounds", func(t *testing.T) { + a := base() + a.Properties = iceberg.Properties{table.DefaultWriteMetricsModeKey: "truncate(2)"} + df, err := table.DataFileFromMetadata(a) + require.NoError(t, err) + assert.EqualValues(t, 1, decodeInt32Bound(t, df.LowerBoundValues()[1])) + assert.EqualValues(t, 3, decodeInt32Bound(t, df.UpperBoundValues()[1])) + assert.Equal(t, "al", decodeStringBound(t, df.LowerBoundValues()[2])) + assert.LessOrEqual(t, len(df.LowerBoundValues()[2]), 2, "string lower bound must be truncated to <= 2 bytes") + }) +} + +func TestDataFileFromMetadata_NullCountsAndColumnSizes(t *testing.T) { + rows := `[ + {"id": 1, "name": "a"}, + {"id": 2, "name": null}, + {"id": 3, "name": null} + ]` + pqBytes, pqMeta := parquetMetaFromSchema(t, dataSchema, rows) + + df, err := table.DataFileFromMetadata(table.DataFileArgs{ + Schema: dataSchema, + Spec: iceberg.NewPartitionSpec(), + Format: iceberg.ParquetFile, + Metadata: pqMeta, + FilePath: "s3://bucket/data/nulls.parquet", + FileSize: int64(len(pqBytes)), + Content: iceberg.EntryContentData, + }) + require.NoError(t, err) + + assert.EqualValues(t, 2, df.NullValueCounts()[2], "name column has 2 nulls") + assert.EqualValues(t, 0, df.NullValueCounts()[1], "id column has no nulls") + + require.NotEmpty(t, df.ColumnSizes(), "column sizes must be populated") + assert.Contains(t, df.ColumnSizes(), 1) + assert.Contains(t, df.ColumnSizes(), 2) +} + Review Comment: Optional coverage: the new `PreservesOrder()` infer branch also fires for truncate (it preserves order), but there's no truncate case here. A sibling to this test with a truncate-partitioned column — shared prefix infers, mixed prefix errors cleanly rather than panicking through recover — would pin that path. Not blocking. -- 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]
