From 222e96d6abb464bdfc6e880617ad706aafb2cbb8 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 20 Oct 2026 23:13:43 -0700 Subject: [PATCH] iceberg maintenance: merge bin files across parquet schema spellings (#11875) * iceberg maintenance: compare parquet schemas by field id in bin-pack merge parquet.EqualNodes ignores Iceberg field ids, so a column dropped and\nre-added under the same name and type compared equal and was merged,\nfiling the new column's values under the dropped field id. Require the\nfield ids to agree wherever both sides carry one; the v3 layout's\nper-field check uses the same test.\n\nGenerated with [Devin](https://devin.ai)\n\nCo-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * iceberg maintenance: merge bin files whose schemas differ by field id A bin used to take the first file's Parquet schema as the writer's and\nrefuse every file that was not structurally identical, so a table never\ncompacted once files had a different column set or REQUIRED-vs-OPTIONAL\nrepetition. A file whose flat columns the writer's schema covers by\nfield id is converted into it (a column the file lacks reads as null);\na file that covers the writer's schema restarts the bin with it, so no\ncolumn is dropped. Equality-delete columns resolve on each file's own\nschema. Renames, type changes, nested columns and files without field\nids stay refused. The sorted and spatial merges are unchanged.\n\nGenerated with [Devin](https://devin.ai)\n\nCo-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * iceberg maintenance: test bin-pack merge across parquet schema spellings Regression coverage for the field-id merge: an added column in either\norder, REQUIRED vs OPTIONAL, equality deletes resolved per file, the\nre-added column refused on v1/v2 and v3, and the refusals that stay\n(rename, type change, one-sided columns, an added required column, no\nfield ids).\n\nGenerated with [Devin](https://devin.ai)\n\nCo-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * iceberg maintenance: resolve missing delete columns as null in bin-pack merge An equality delete on an optional column added after a data file was\nwritten no longer refuses the bin: the file reads the column as null,\nso a real delete key cannot match its rows and a null key deletes them.\n\nGenerated with [Devin](https://devin.ai)\n\nCo-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * iceberg maintenance: write bin-pack merges under the files' union schema The writer schema is now the union of the bin's file schemas by field\nid, computed in a schema-only pass before any row is written: a column\nstays REQUIRED only where every file has it and spells it so, and the\nmerge refuses only true name/type/id conflicts. Files that need a union\nneither side covers (an added column plus a REQUIRED/OPTIONAL\ndifference, disjoint added columns) now merge in a single pass instead\nof refusing, and progressively wider bins no longer reread and rewrite\nevery earlier file per restart.\n\nGenerated with [Devin](https://devin.ai)\n\nCo-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * iceberg maintenance: spell the re-added column's test file for codespell Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * iceberg maintenance: resolve equality-delete columns by field id A column renamed after its data and delete files were written still carries its field id there; resolving it by id instead of the current name keeps a rename from turning every equality key null. The missing-column null-key fallback is now for data files only: an equality-delete file that lacks its equality column is refused, as the spec requires it to carry them. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * iceberg maintenance: read only the parquet footer choosing a bin's writer schema The union-schema pass downloaded every input in full, then the merge downloaded each again. The footer suffices for the schema, and a chunked file's ReadAt pulls only the chunks the footer spans. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/worker/tasks/iceberg/compact.go | 197 +++++- .../tasks/iceberg/compact_schema_test.go | 644 ++++++++++++++++++ weed/worker/tasks/iceberg/filer_io.go | 58 +- weed/worker/tasks/iceberg/row_lineage.go | 3 +- weed/worker/tasks/iceberg/schema_evolution.go | 184 +++++ 5 files changed, 1061 insertions(+), 25 deletions(-) create mode 100644 weed/worker/tasks/iceberg/compact_schema_test.go create mode 100644 weed/worker/tasks/iceberg/schema_evolution.go diff --git a/weed/worker/tasks/iceberg/compact.go b/weed/worker/tasks/iceberg/compact.go index bdf548eaa7..e63e9e4a4a 100644 --- a/weed/worker/tasks/iceberg/compact.go +++ b/weed/worker/tasks/iceberg/compact.go @@ -962,7 +962,7 @@ func readEqualityDeleteFile( reader := parquet.NewReader(bytes.NewReader(data)) defer reader.Close() - colIndices, err := resolveEqualityColIndices(reader.Schema(), fieldIDs, icebergSchema) + colIndices, err := resolveEqualityColIndices(reader.Schema(), fieldIDs, icebergSchema, false) if err != nil { return nil, fmt.Errorf("resolve columns in %s: %w", filePath, err) } @@ -993,24 +993,37 @@ func readEqualityDeleteFile( // buildEqualityKey builds a composite string key from specific column values // in a row. Each value is serialized as "kind:length:value" to avoid ambiguity // between types (e.g., int 123 vs string "123") and to prevent collisions from -// values containing separator characters. +// values containing separator characters. A negative column index is a field +// the file lacks: the value there is null. func buildEqualityKey(row parquet.Row, colIndices []int) string { if len(colIndices) == 1 { - v := row[colIndices[0]] + v := equalityKeyValue(row, colIndices[0]) s := v.String() return fmt.Sprintf("%d:%d:%s", v.Kind(), len(s), s) } var b strings.Builder for _, idx := range colIndices { - v := row[idx] + v := equalityKeyValue(row, idx) s := v.String() fmt.Fprintf(&b, "%d:%d:%s", v.Kind(), len(s), s) } return b.String() } +func equalityKeyValue(row parquet.Row, idx int) parquet.Value { + if idx < 0 { + return parquet.NullValue() + } + return row[idx] +} + // resolveEqualityColIndices maps Iceberg field IDs to Parquet column indices. -func resolveEqualityColIndices(pqSchema *parquet.Schema, fieldIDs []int, icebergSchema *iceberg.Schema) ([]int, error) { +// A file carrying field ids resolves by id, so a column renamed since the +// file was written still matches; one without ids falls back to the Iceberg +// name. allowMissing reads a column the file lacks as null (-1 marks it in +// the key): for data files only — an equality-delete file always carries +// its equality columns, so a missing one stays an error there. +func resolveEqualityColIndices(pqSchema *parquet.Schema, fieldIDs []int, icebergSchema *iceberg.Schema, allowMissing bool) ([]int, error) { if len(fieldIDs) == 0 { return nil, nil } @@ -1019,6 +1032,14 @@ func resolveEqualityColIndices(pqSchema *parquet.Schema, fieldIDs []int, iceberg for i, col := range pqSchema.Columns() { colNameToIdx[strings.Join(col, ".")] = i } + idToIdx := make(map[int]int) + for _, f := range pqSchema.Fields() { + if f.Leaf() && f.ID() > 0 { + if idx, ok := colNameToIdx[f.Name()]; ok { + idToIdx[f.ID()] = idx + } + } + } indices := make([]int, len(fieldIDs)) for i, fid := range fieldIDs { @@ -1026,8 +1047,17 @@ func resolveEqualityColIndices(pqSchema *parquet.Schema, fieldIDs []int, iceberg if !ok { return nil, fmt.Errorf("field ID %d not found in iceberg schema", fid) } - idx, ok := colNameToIdx[field.Name] + idx, ok := idToIdx[fid] + if !ok && len(idToIdx) == 0 { + idx, ok = colNameToIdx[field.Name] + } if !ok { + // An optional column added to the table after this file was + // written reads as null there (-1 marks it in the key). + if allowMissing && !field.Required { + indices[i] = -1 + continue + } return nil, fmt.Errorf("column %q (field ID %d) not found in parquet schema", field.Name, fid) } indices[i] = idx @@ -1089,6 +1119,86 @@ func mergeParquetFiles( icebergSchema *iceberg.Schema, rowGroupRows int64, lineage bool, +) ([]byte, int64, error) { + // On a v1/v2 table the writer's schema is chosen before any row is + // written: the union of the bin's file schemas by field id, so a bin of + // progressively wider files merges in one pass instead of restarting + // (and rewriting every earlier file) on each wider one. The v3 merge + // keeps the first file's columns plus the lineage columns. + var target *parquet.Schema + if !lineage { + var err error + if target, err = binWriterSchema(ctx, filerClient, bucketName, dataPath, entries); err != nil { + return nil, 0, err + } + } + for attempt := 0; ; attempt++ { + out, rows, err := mergeParquetFilesInto(ctx, filerClient, bucketName, dataPath, entries, positionDeletes, + eqDeleteGroups, icebergSchema, rowGroupRows, lineage, target) + var wider *widerSchemaError + if errors.As(err, &wider) && attempt < len(entries) { + target = wider.schema + continue + } + return out, rows, err + } +} + +// binWriterSchema is the writer schema for a v1/v2 bin-pack merge: the union +// of the files' schemas by field id. Only each file's footer is read to learn +// its schema; the merge then downloads each file once. Files whose schemas +// cannot union (no field ids, nested or repeated columns, a field id or name +// carrying different types) merge only under the old rule — every file's +// schema equal to the first — and the bin is refused as before otherwise. +func binWriterSchema( + ctx context.Context, + filerClient filer_pb.SeaweedFilerClient, + bucketName, dataPath string, + entries []iceberg.ManifestEntry, +) (*parquet.Schema, error) { + var union schemaUnion + var first *parquet.Schema + unionable, allEqual := true, true + for _, entry := range entries { + source := entry.DataFile().FilePath() + schema, err := parquetSchemaByIcebergPath(ctx, filerClient, bucketName, dataPath, source) + if err != nil { + return nil, fmt.Errorf("schema of %s: %w", source, err) + } + if schema == nil { + return nil, fmt.Errorf("no parquet schema found in %s", source) + } + if first == nil { + first = schema + } else if !schemasEqual(first, schema) { + allEqual = false + } + if unionable { + unionable = union.fold(schema, source) == nil + } + if !unionable && !allEqual { + return nil, fmt.Errorf("schema mismatch in %s: cannot merge files with different schemas", source) + } + } + if !unionable || first == nil { + return nil, nil // every file spells the first file's schema + } + return union.schema(first.Name(), first), nil +} + +// mergeParquetFilesInto merges with target as the writer's schema on a v1/v2 +// table (nil: the first file's). +func mergeParquetFilesInto( + ctx context.Context, + filerClient filer_pb.SeaweedFilerClient, + bucketName, dataPath string, + entries []iceberg.ManifestEntry, + positionDeletes map[string][]int64, + eqDeleteGroups []equalityDeleteGroup, + icebergSchema *iceberg.Schema, + rowGroupRows int64, + lineage bool, + target *parquet.Schema, ) ([]byte, int64, error) { if len(entries) == 0 { return nil, 0, fmt.Errorf("no entries to merge") @@ -1122,6 +1232,30 @@ func mergeParquetFiles( return nil, 0, fmt.Errorf("schema of %s: %w", entries[0].DataFile().FilePath(), err) } outputSchema = layout.schema + } else if target != nil { + outputSchema = target + } + + // joins is how one file enters the merge on a v1/v2 table: converted into + // the writer's schema when that covers it, with its equality-delete + // columns resolved on its own schema. + joins := func(reader *parquet.Reader, entry iceberg.ManifestEntry) (parquet.Conversion, []resolvedEqDeleteGroup, error) { + if layout != nil { + return nil, resolvedEqGroups, nil // a v3 merge matches files with their lineage columns set aside + } + source := entry.DataFile().FilePath() + conv, err := evolvedFileConversion(outputSchema, reader.Schema(), source) + if err != nil { + return nil, nil, err + } + if schemasEqual(parquetSchema, reader.Schema()) { + return conv, resolvedEqGroups, nil + } + groups, err := resolveEqualityDeleteGroupsForSchema(reader.Schema(), eqDeleteGroups, icebergSchema) + if err != nil { + return nil, nil, fmt.Errorf("resolve equality columns in %s: %w", source, err) + } + return conv, groups, nil } var outputBuf bytes.Buffer @@ -1129,7 +1263,17 @@ func mergeParquetFiles( drainReader := func(reader *parquet.Reader, entry iceberg.ManifestEntry) (int64, error) { source := entry.DataFile().FilePath() - return copyFileRows(ctx, reader, entry, layout, bucketName, dataPath, positionDeletes, resolvedEqGroups, eqDeleteGroups, icebergSchema, func(filtered []parquet.Row) error { + conv, groups, err := joins(reader, entry) + if err != nil { + reader.Close() + return 0, err + } + return copyFileRows(ctx, reader, entry, layout, bucketName, dataPath, positionDeletes, groups, eqDeleteGroups, icebergSchema, func(filtered []parquet.Row) error { + if conv != nil { + if _, err := conv.Convert(filtered); err != nil { + return fmt.Errorf("convert rows from %s: %w", source, err) + } + } if _, err := writer.WriteRows(filtered); err != nil { return fmt.Errorf("write rows from %s: %w", source, err) } @@ -1161,13 +1305,6 @@ func mergeParquetFiles( } reader := parquet.NewReader(bytes.NewReader(data)) - // A v3 merge matches files with their lineage columns set aside. - if layout == nil && !schemasEqual(parquetSchema, reader.Schema()) { - reader.Close() - writer.Close() - return nil, 0, fmt.Errorf("schema mismatch in %s: cannot merge files with different schemas", entry.DataFile().FilePath()) - } - rowsWritten, err := drainReader(reader, entry) if err != nil { writer.Close() @@ -1200,7 +1337,7 @@ func resolveEqualityDeleteGroupsForSchema( resolved := make([]resolvedEqDeleteGroup, 0, len(eqDeleteGroups)) for _, g := range eqDeleteGroups { - indices, err := resolveEqualityColIndices(parquetSchema, g.FieldIDs, icebergSchema) + indices, err := resolveEqualityColIndices(parquetSchema, g.FieldIDs, icebergSchema, true) if err != nil { return nil, err } @@ -1429,7 +1566,8 @@ func compactRandomSuffix() string { return hex.EncodeToString(b) } -// schemasEqual compares two parquet schemas structurally. +// schemasEqual compares two parquet schemas structurally, and their Iceberg +// field ids wherever both carry them. func schemasEqual(a, b *parquet.Schema) bool { if a == b { return true @@ -1437,7 +1575,32 @@ func schemasEqual(a, b *parquet.Schema) bool { if a == nil || b == nil { return false } - return parquet.EqualNodes(a, b) + return parquet.EqualNodes(a, b) && fieldIDsAgree(a, b) +} + +// fieldIDsAgree reports whether two structurally equal nodes carry the same +// Iceberg field id at every position where both carry one. parquet-go's +// EqualNodes ignores field ids, so a column dropped and re-added under the +// same name and type (a new field id) looked like the old one, and a merge +// filed the old values under the new field, or the live values under the +// dropped one. +func fieldIDsAgree(a, b parquet.Node) bool { + if a.ID() > 0 && b.ID() > 0 && a.ID() != b.ID() { + return false + } + if a.Leaf() || b.Leaf() { + return a.Leaf() == b.Leaf() + } + fa, fb := a.Fields(), b.Fields() + if len(fa) != len(fb) { + return false + } + for i := range fa { + if fa[i].Name() != fb[i].Name() || !fieldIDsAgree(fa[i], fb[i]) { + return false + } + } + return true } // ensureFilerDir ensures a directory exists in the filer. diff --git a/weed/worker/tasks/iceberg/compact_schema_test.go b/weed/worker/tasks/iceberg/compact_schema_test.go new file mode 100644 index 0000000000..3d119037da --- /dev/null +++ b/weed/worker/tasks/iceberg/compact_schema_test.go @@ -0,0 +1,644 @@ +package iceberg + +import ( + "bytes" + "context" + "io" + "strings" + "testing" + + "github.com/apache/iceberg-go" + "github.com/parquet-go/parquet-go" + + filer_pb "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +// Files of one Iceberg table may spell its schema differently in Parquet: +// a column added by schema evolution, or the same schema written REQUIRED +// by one engine and OPTIONAL by another (PyIceberg writes REQUIRED, DuckDB +// OPTIONAL). The bin-pack merge joins such files by Iceberg field id, +// reading a column a file lacks as null, and refuses files whose same-name +// columns carry different field ids (a column dropped and re-added). + +type narrowSchemaRow struct { // id REQUIRED id=1, name OPTIONAL id=2 + ID int64 `parquet:"id,id(1)"` + Name string `parquet:"name,id(2),optional"` +} + +type wideSchemaRow struct { // plus extra OPTIONAL id=3 + ID int64 `parquet:"id,id(1)"` + Name string `parquet:"name,id(2),optional"` + Extra string `parquet:"extra,id(3),optional"` +} + +type reorderedSchemaRow struct { // the wide schema's columns in another order + Extra string `parquet:"extra,id(3),optional"` + ID int64 `parquet:"id,id(1)"` + Name string `parquet:"name,id(2),optional"` +} + +type optionalIDSchemaRow struct { // id OPTIONAL id=1 + ID int64 `parquet:"id,id(1),optional"` + Name string `parquet:"name,id(2),optional"` +} + +type readdedNameSchemaRow struct { // name re-added under field id 3 + ID int64 `parquet:"id,id(1)"` + Name string `parquet:"name,id(3),optional"` +} + +type nullableExtraSchemaRow struct { // the wide schema, extra writable as null + ID int64 `parquet:"id,id(1)"` + Name string `parquet:"name,id(2),optional"` + Extra *string `parquet:"extra,id(3),optional"` +} + +func newAddedColumnSchema() *iceberg.Schema { // newTestSchema plus extra id=3 + return iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Type: iceberg.PrimitiveTypes.Int64, Name: "id", Required: true}, + iceberg.NestedField{ID: 2, Type: iceberg.PrimitiveTypes.String, Name: "name", Required: false}, + iceberg.NestedField{ID: 3, Type: iceberg.PrimitiveTypes.String, Name: "extra", Required: false}, + ) +} + +func schemaFileEntry(t *testing.T, path string, recordCount int64) iceberg.ManifestEntry { + t.Helper() + dfb, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec, iceberg.EntryContentData, + path, iceberg.ParquetFile, map[int]any{}, nil, nil, recordCount, 1024) + if err != nil { + t.Fatalf("build data file: %v", err) + } + snapID := int64(1) + return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build()) +} + +func mergeNamedFiles(t *testing.T, client filer_pb.SeaweedFilerClient, names []string, counts map[string]int64, + eqGroups []equalityDeleteGroup, schema *iceberg.Schema, lineage bool) ([]byte, int64, error) { + t.Helper() + var entries []iceberg.ManifestEntry + for _, name := range names { + entries = append(entries, schemaFileEntry(t, "data/"+name, counts[name])) + } + return mergeParquetFiles(context.Background(), client, "test-bucket", "ns/tbl", + entries, nil, eqGroups, schema, 0, lineage) +} + +// mergedValues returns the merged file's schema and each row's leaf values +// in column order (nil for nulls). +func mergedValues(t *testing.T, data []byte) (*parquet.Schema, [][]any) { + t.Helper() + f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data))) + if err != nil { + t.Fatalf("open merged: %v", err) + } + var out [][]any + for _, rg := range f.RowGroups() { + rows := rg.Rows() + buf := make([]parquet.Row, 16) + for { + n, err := rows.ReadRows(buf) + for _, r := range buf[:n] { + vals := make([]any, len(r)) + for i, v := range r { + if v.IsNull() { + continue + } + if v.Kind() == parquet.Int64 { + vals[i] = v.Int64() + } else { + vals[i] = v.String() + } + } + out = append(out, vals) + } + if err == io.EOF { + break + } + if err != nil { + t.Fatalf("read merged rows: %v", err) + } + } + rows.Close() + } + return f.Schema(), out +} + +func checkMergedRows(t *testing.T, data []byte, want [][]any) *parquet.Schema { + t.Helper() + schema, got := mergedValues(t, data) + if len(got) != len(want) { + t.Fatalf("merged rows = %v, want %v", got, want) + } + for i := range want { + if len(got[i]) != len(want[i]) { + t.Fatalf("row %d = %v, want %v", i, got[i], want[i]) + } + for j := range want[i] { + if got[i][j] != want[i][j] { + t.Fatalf("row %d column %d = %v, want %v (row %v)", i, j, got[i][j], want[i][j], got[i]) + } + } + } + return schema +} + +// A column added to the table after some files were written: the merge +// joins the files by field id and reads the new column as null for the +// old file's rows, whichever file leads the bin. +func TestMergeParquetFilesAddedColumn(t *testing.T) { + put := func(fs *fakeFilerServer) map[string]int64 { + putDataFileRows(t, fs, "narrow.parquet", []narrowSchemaRow{{1, "a"}, {2, "b"}, {3, "c"}}) + putDataFileRows(t, fs, "wide.parquet", []wideSchemaRow{{4, "d", "e4"}, {5, "e", "e5"}}) + return map[string]int64{"narrow.parquet": 3, "wide.parquet": 2} + } + checkFields := func(t *testing.T, schema *parquet.Schema) { + fields := schema.Fields() + if len(fields) != 3 { + t.Fatalf("merged fields = %d, want 3", len(fields)) + } + names := []string{fields[0].Name(), fields[1].Name(), fields[2].Name()} + if names[0] != "id" || names[1] != "name" || names[2] != "extra" { + t.Fatalf("merged fields = %v, want [id name extra]", names) + } + if fields[2].ID() != 3 { + t.Fatalf("extra field id = %d, want 3", fields[2].ID()) + } + } + + t.Run("narrow first", func(t *testing.T) { + fs, client := startFakeFiler(t) + counts := put(fs) + merged, count, err := mergeNamedFiles(t, client, + []string{"narrow.parquet", "wide.parquet"}, counts, nil, nil, false) + if err != nil { + t.Fatalf("mergeParquetFiles: %v", err) + } + if count != 5 { + t.Fatalf("merged %d rows, want 5", count) + } + checkFields(t, checkMergedRows(t, merged, [][]any{ + {int64(1), "a", nil}, + {int64(2), "b", nil}, + {int64(3), "c", nil}, + {int64(4), "d", "e4"}, + {int64(5), "e", "e5"}, + })) + }) + + t.Run("wide first", func(t *testing.T) { + fs, client := startFakeFiler(t) + counts := put(fs) + merged, count, err := mergeNamedFiles(t, client, + []string{"wide.parquet", "narrow.parquet"}, counts, nil, nil, false) + if err != nil { + t.Fatalf("mergeParquetFiles: %v", err) + } + if count != 5 { + t.Fatalf("merged %d rows, want 5", count) + } + checkFields(t, checkMergedRows(t, merged, [][]any{ + {int64(4), "d", "e4"}, + {int64(5), "e", "e5"}, + {int64(1), "a", nil}, + {int64(2), "b", nil}, + {int64(3), "c", nil}, + })) + }) +} + +// The same Iceberg schema written REQUIRED by one engine and OPTIONAL by +// another merges; the writer keeps the wider, optional, repetition. +func TestMergeParquetFilesRequiredVsOptional(t *testing.T) { + fs, client := startFakeFiler(t) + putDataFileRows(t, fs, "required.parquet", []narrowSchemaRow{{1, "a"}, {2, "b"}}) + putDataFileRows(t, fs, "optional.parquet", []optionalIDSchemaRow{{3, "c"}, {4, "d"}}) + + merged, count, err := mergeNamedFiles(t, client, + []string{"required.parquet", "optional.parquet"}, + map[string]int64{"required.parquet": 2, "optional.parquet": 2}, nil, nil, false) + if err != nil { + t.Fatalf("mergeParquetFiles: %v", err) + } + if count != 4 { + t.Fatalf("merged %d rows, want 4", count) + } + schema := checkMergedRows(t, merged, [][]any{ + {int64(1), "a"}, + {int64(2), "b"}, + {int64(3), "c"}, + {int64(4), "d"}, + }) + if !schema.Fields()[0].Optional() { + t.Error("merged id column is REQUIRED, want OPTIONAL") + } +} + +// A column dropped and re-added under the same name and type has a new +// field id; merging its file with an old one would file the new column's +// values under the dropped field id, so the bin is refused. +func TestMergeParquetFilesReaddedColumn(t *testing.T) { + for _, names := range [][]string{ + {"narrow.parquet", "re-added.parquet"}, + {"re-added.parquet", "narrow.parquet"}, + } { + fs, client := startFakeFiler(t) + putDataFileRows(t, fs, "narrow.parquet", []narrowSchemaRow{{1, "a"}}) + putDataFileRows(t, fs, "re-added.parquet", []readdedNameSchemaRow{{2, "new-2"}}) + + _, _, err := mergeNamedFiles(t, client, names, + map[string]int64{"narrow.parquet": 1, "re-added.parquet": 1}, nil, nil, false) + if err == nil || !strings.Contains(err.Error(), "cannot merge files with different schemas") { + t.Errorf("merge of %v: %v, want schema mismatch", names, err) + } + } +} + +type disjointColumnsRow struct { // id id=1, extra id=3: no name + ID int64 `parquet:"id,id(1)"` + Extra string `parquet:"extra,id(3),optional"` +} + +type requiredExtraSchemaRow struct { // extra REQUIRED id=3 + ID int64 `parquet:"id,id(1)"` + Name string `parquet:"name,id(2),optional"` + Extra string `parquet:"extra,id(3)"` +} + +// Differences field-id matching does not cover stay refused: a rename, a +// type change, files without field ids; and a control pair still compacts. +func TestMergeParquetFilesSchemaRefusals(t *testing.T) { + type renamedSchemaRow struct { + ID int64 `parquet:"id,id(1)"` + Other string `parquet:"other,id(2),optional"` + } + type widenedSchemaRow struct { + ID int32 `parquet:"id,id(1)"` + Name string `parquet:"name,id(2),optional"` + } + type noIDExtraRow struct { + ID int64 `parquet:"id"` + Name string `parquet:"name,optional"` + Extra string `parquet:"extra,optional"` + } + + cases := []struct { + name string + writeFile func(fs *fakeFilerServer, name string) + }{ + {"rename", func(fs *fakeFilerServer, name string) { + putDataFileRows(t, fs, name, []renamedSchemaRow{{5, "e"}}) + }}, + {"type change", func(fs *fakeFilerServer, name string) { + putDataFileRows(t, fs, name, []widenedSchemaRow{{5, "e"}}) + }}, + {"no field ids", func(fs *fakeFilerServer, name string) { + putDataFileRows(t, fs, name, []noIDExtraRow{{5, "e", "x"}}) + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + fs, client := startFakeFiler(t) + putDataFileRows(t, fs, "narrow.parquet", []narrowSchemaRow{{1, "a"}}) + tc.writeFile(fs, "other.parquet") + + _, _, err := mergeNamedFiles(t, client, + []string{"narrow.parquet", "other.parquet"}, + map[string]int64{"narrow.parquet": 1, "other.parquet": 1}, nil, nil, false) + if err == nil || !strings.Contains(err.Error(), "cannot merge files with different schemas") { + t.Errorf("merge: %v, want schema mismatch", err) + } + }) + } + + t.Run("control", func(t *testing.T) { + fs, client := startFakeFiler(t) + putDataFileRows(t, fs, "a.parquet", []narrowSchemaRow{{1, "a"}}) + putDataFileRows(t, fs, "b.parquet", []narrowSchemaRow{{2, "b"}}) + if _, _, err := mergeNamedFiles(t, client, []string{"a.parquet", "b.parquet"}, + map[string]int64{"a.parquet": 1, "b.parquet": 1}, nil, nil, false); err != nil { + t.Errorf("control merge: %v", err) + } + }) +} + +// An equality-delete column can sit at a different position in an older +// file: it is resolved on each file's own schema. +func TestMergeParquetFilesEvolvedSchemaEqualityDelete(t *testing.T) { + fs, client := startFakeFiler(t) + putDataFileRows(t, fs, "narrow.parquet", []narrowSchemaRow{{1, "a"}, {2, "b"}, {3, "c"}}) + putDataFileRows(t, fs, "reordered.parquet", []reorderedSchemaRow{ + {"e4", 4, "d"}, {"e5", 5, "e"}, + }) + + // An equality delete on name (field id 2) for "b" (index 1 in the + // narrow file) and "e" (index 2 in the reordered file). + type eqDeleteRow struct { + Name string `parquet:"name"` + } + var eqBuf bytes.Buffer + eqWriter := parquet.NewWriter(&eqBuf, parquet.SchemaOf(new(eqDeleteRow))) + for _, r := range []eqDeleteRow{{"b"}, {"e"}} { + if err := eqWriter.Write(&r); err != nil { + t.Fatalf("write eq delete: %v", err) + } + } + if err := eqWriter.Close(); err != nil { + t.Fatalf("close eq writer: %v", err) + } + fs.putEntry("/buckets/test-bucket/ns/tbl/data", "eq-del.parquet", &filer_pb.Entry{ + Name: "eq-del.parquet", Content: eqBuf.Bytes(), + }) + + eqDfb, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec, iceberg.EntryContentEqDeletes, + "data/eq-del.parquet", iceberg.ParquetFile, map[int]any{}, nil, nil, 2, int64(eqBuf.Len())) + if err != nil { + t.Fatalf("build eq delete: %v", err) + } + eqDfb.EqualityFieldIDs([]int{2}) + snapID := int64(1) + eqEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, eqDfb.Build()) + + eqGroups, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", + []iceberg.ManifestEntry{eqEntry}, newTestSchema()) + if err != nil { + t.Fatalf("collectEqualityDeletes: %v", err) + } + + merged, count, err := mergeNamedFiles(t, client, + []string{"narrow.parquet", "reordered.parquet"}, + map[string]int64{"narrow.parquet": 3, "reordered.parquet": 2}, eqGroups, newTestSchema(), false) + if err != nil { + t.Fatalf("mergeParquetFiles: %v", err) + } + if count != 3 { + t.Fatalf("merged %d rows, want 3", count) + } + // The union schema (id, name, extra) is the writer's. + checkMergedRows(t, merged, [][]any{ + {int64(1), "a", nil}, + {int64(3), "c", nil}, + {int64(4), "d", "e4"}, + }) +} + +// File pairs where neither schema covers the other merge under the union +// writer schema: a REQUIRED/OPTIONAL difference plus an added column, a +// column each side lacks the other's of, and an added REQUIRED column +// (demoted to OPTIONAL: the files without it read null). +func TestMergeParquetFilesUnionWriterSchema(t *testing.T) { + t.Run("added column and repetition differ", func(t *testing.T) { + for _, names := range [][]string{ + {"wide.parquet", "optional.parquet"}, + {"optional.parquet", "wide.parquet"}, + } { + fs, client := startFakeFiler(t) + putDataFileRows(t, fs, "wide.parquet", []wideSchemaRow{{4, "d", "e4"}, {5, "e", "e5"}}) + putDataFileRows(t, fs, "optional.parquet", []optionalIDSchemaRow{{1, "a"}, {2, "b"}}) + + merged, count, err := mergeNamedFiles(t, client, names, + map[string]int64{"wide.parquet": 2, "optional.parquet": 2}, nil, nil, false) + if err != nil { + t.Fatalf("merge of %v: %v", names, err) + } + if count != 4 { + t.Fatalf("merged %d rows, want 4", count) + } + var want [][]any + if names[0] == "wide.parquet" { + want = [][]any{{int64(4), "d", "e4"}, {int64(5), "e", "e5"}, {int64(1), "a", nil}, {int64(2), "b", nil}} + } else { + want = [][]any{{int64(1), "a", nil}, {int64(2), "b", nil}, {int64(4), "d", "e4"}, {int64(5), "e", "e5"}} + } + schema := checkMergedRows(t, merged, want) + if !schema.Fields()[0].Optional() { + t.Errorf("merged id column is REQUIRED, want OPTIONAL") + } + } + }) + + t.Run("disjoint columns", func(t *testing.T) { + for _, names := range [][]string{ + {"narrow.parquet", "disjoint.parquet"}, + {"disjoint.parquet", "narrow.parquet"}, + } { + fs, client := startFakeFiler(t) + putDataFileRows(t, fs, "narrow.parquet", []narrowSchemaRow{{1, "a"}}) + putDataFileRows(t, fs, "disjoint.parquet", []disjointColumnsRow{{5, "e"}}) + + merged, count, err := mergeNamedFiles(t, client, names, + map[string]int64{"narrow.parquet": 1, "disjoint.parquet": 1}, nil, nil, false) + if err != nil { + t.Fatalf("merge of %v: %v", names, err) + } + if count != 2 { + t.Fatalf("merged %d rows, want 2", count) + } + var want [][]any + if names[0] == "narrow.parquet" { // union (id, name, extra) + want = [][]any{{int64(1), "a", nil}, {int64(5), nil, "e"}} + } else { // union (id, extra, name) + want = [][]any{{int64(5), "e", nil}, {int64(1), nil, "a"}} + } + checkMergedRows(t, merged, want) + } + }) + + t.Run("added required column", func(t *testing.T) { + fs, client := startFakeFiler(t) + putDataFileRows(t, fs, "narrow.parquet", []narrowSchemaRow{{1, "a"}}) + putDataFileRows(t, fs, "required-extra.parquet", []requiredExtraSchemaRow{{5, "e", "x"}}) + + merged, count, err := mergeNamedFiles(t, client, + []string{"narrow.parquet", "required-extra.parquet"}, + map[string]int64{"narrow.parquet": 1, "required-extra.parquet": 1}, nil, nil, false) + if err != nil { + t.Fatalf("mergeParquetFiles: %v", err) + } + if count != 2 { + t.Fatalf("merged %d rows, want 2", count) + } + schema := checkMergedRows(t, merged, [][]any{ + {int64(1), "a", nil}, + {int64(5), "e", "x"}, + }) + if !schema.Fields()[2].Optional() { + t.Error("merged extra column is REQUIRED, want OPTIONAL") + } + }) +} + +// An equality delete naming a column added after some files were written: +// the files that lack it read the delete key's column as null, so a real +// key does not match their rows and a null key deletes them all. +func TestMergeParquetFilesEqualityDeleteAddedColumn(t *testing.T) { + strPtr := func(s string) *string { return &s } + put := func(fs *fakeFilerServer) map[string]int64 { + putDataFileRows(t, fs, "narrow.parquet", []narrowSchemaRow{{1, "a"}, {2, "b"}, {3, "c"}}) + putDataFileRows(t, fs, "wide.parquet", []nullableExtraSchemaRow{{4, "d", strPtr("e4")}, {5, "e", nil}}) + return map[string]int64{"narrow.parquet": 3, "wide.parquet": 2} + } + eqGroups := func(keys ...parquet.Value) []equalityDeleteGroup { + set := make(map[string]struct{}, len(keys)) + for _, k := range keys { + set[buildEqualityKey(parquet.Row{k}, []int{0})] = struct{}{} + } + return []equalityDeleteGroup{{FieldIDs: []int{3}, Keys: set}} + } + schema := newAddedColumnSchema() + + t.Run("key matches no null", func(t *testing.T) { + for _, names := range [][]string{ + {"narrow.parquet", "wide.parquet"}, + {"wide.parquet", "narrow.parquet"}, + } { + fs, client := startFakeFiler(t) + counts := put(fs) + merged, count, err := mergeNamedFiles(t, client, names, counts, + eqGroups(parquet.ValueOf("e4")), schema, false) + if err != nil { + t.Fatalf("merge of %v: %v", names, err) + } + if count != 4 { + t.Fatalf("merged %d rows, want 4", count) + } + var want [][]any + if names[0] == "narrow.parquet" { + want = [][]any{{int64(1), "a", nil}, {int64(2), "b", nil}, {int64(3), "c", nil}, {int64(5), "e", nil}} + } else { + want = [][]any{{int64(5), "e", nil}, {int64(1), "a", nil}, {int64(2), "b", nil}, {int64(3), "c", nil}} + } + checkMergedRows(t, merged, want) + } + }) + + t.Run("null key deletes nulls", func(t *testing.T) { + for _, names := range [][]string{ + {"narrow.parquet", "wide.parquet"}, + {"wide.parquet", "narrow.parquet"}, + } { + fs, client := startFakeFiler(t) + counts := put(fs) + merged, count, err := mergeNamedFiles(t, client, names, counts, + eqGroups(parquet.NullValue()), schema, false) + if err != nil { + t.Fatalf("merge of %v: %v", names, err) + } + if count != 1 { + t.Fatalf("merged %d rows, want 1", count) + } + checkMergedRows(t, merged, [][]any{{int64(4), "d", "e4"}}) + } + }) +} + +// A column renamed after its files were written still carries its field id, +// so an equality delete on it resolves the column by id and removes only the +// rows it names — the old name looking absent must not turn every key null. +func TestMergeParquetFilesEqualityDeleteRenamedColumn(t *testing.T) { + // The table renamed name (field id 2) to label; every file still spells + // the old name. + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Type: iceberg.PrimitiveTypes.Int64, Name: "id", Required: true}, + iceberg.NestedField{ID: 2, Type: iceberg.PrimitiveTypes.String, Name: "label", Required: false}, + ) + type eqDeleteRow struct { + Name string `parquet:"name,id(2)"` + } + + fs, client := startFakeFiler(t) + putDataFileRows(t, fs, "a.parquet", []narrowSchemaRow{{1, "a"}, {2, "b"}, {3, "c"}}) + putDataFileRows(t, fs, "b.parquet", []narrowSchemaRow{{4, "d"}}) + + var eqBuf bytes.Buffer + eqWriter := parquet.NewWriter(&eqBuf, parquet.SchemaOf(new(eqDeleteRow))) + if err := eqWriter.Write(&eqDeleteRow{"b"}); err != nil { + t.Fatalf("write eq delete: %v", err) + } + if err := eqWriter.Close(); err != nil { + t.Fatalf("close eq writer: %v", err) + } + fs.putEntry("/buckets/test-bucket/ns/tbl/data", "eq-del.parquet", &filer_pb.Entry{ + Name: "eq-del.parquet", Content: eqBuf.Bytes(), + }) + + eqDfb, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec, iceberg.EntryContentEqDeletes, + "data/eq-del.parquet", iceberg.ParquetFile, map[int]any{}, nil, nil, 1, int64(eqBuf.Len())) + if err != nil { + t.Fatalf("build eq delete: %v", err) + } + eqDfb.EqualityFieldIDs([]int{2}) + snapID := int64(1) + eqEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, eqDfb.Build()) + + eqGroups, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", + []iceberg.ManifestEntry{eqEntry}, schema) + if err != nil { + t.Fatalf("collectEqualityDeletes: %v", err) + } + + merged, count, err := mergeNamedFiles(t, client, + []string{"a.parquet", "b.parquet"}, + map[string]int64{"a.parquet": 3, "b.parquet": 1}, eqGroups, schema, false) + if err != nil { + t.Fatalf("mergeParquetFiles: %v", err) + } + if count != 3 { + t.Fatalf("merged %d rows, want 3", count) + } + checkMergedRows(t, merged, [][]any{ + {int64(1), "a"}, + {int64(3), "c"}, + {int64(4), "d"}, + }) +} + +// An equality-delete file that lacks its equality column is refused rather +// than read as null keys: the spec requires equality deletes to carry the +// columns they name, so the missing-column fallback is for data files only. +func TestEqualityDeleteFileMissingColumn(t *testing.T) { + fs, client := startFakeFiler(t) + type otherRow struct { + Other string `parquet:"other,id(9)"` + } + var eqBuf bytes.Buffer + eqWriter := parquet.NewWriter(&eqBuf, parquet.SchemaOf(new(otherRow))) + if err := eqWriter.Write(&otherRow{"x"}); err != nil { + t.Fatalf("write eq delete: %v", err) + } + if err := eqWriter.Close(); err != nil { + t.Fatalf("close eq writer: %v", err) + } + fs.putEntry("/buckets/test-bucket/ns/tbl/data", "eq-del.parquet", &filer_pb.Entry{ + Name: "eq-del.parquet", Content: eqBuf.Bytes(), + }) + + eqDfb, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec, iceberg.EntryContentEqDeletes, + "data/eq-del.parquet", iceberg.ParquetFile, map[int]any{}, nil, nil, 1, int64(eqBuf.Len())) + if err != nil { + t.Fatalf("build eq delete: %v", err) + } + eqDfb.EqualityFieldIDs([]int{2}) + snapID := int64(1) + eqEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, eqDfb.Build()) + + _, err = collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", + []iceberg.ManifestEntry{eqEntry}, newAddedColumnSchema()) + if err == nil { + t.Fatal("collectEqualityDeletes: want an error for a missing equality column") + } +} + +// The v3 merge matches files on the first file's columns; the re-added +// column is refused there too. +func TestMergeParquetFilesV3ReaddedColumn(t *testing.T) { + fs, client := startFakeFiler(t) + putDataFileRows(t, fs, "narrow.parquet", []narrowSchemaRow{{1, "a"}}) + putDataFileRows(t, fs, "re-added.parquet", []readdedNameSchemaRow{{2, "new-2"}}) + + _, _, err := mergeNamedFiles(t, client, + []string{"narrow.parquet", "re-added.parquet"}, + map[string]int64{"narrow.parquet": 1, "re-added.parquet": 1}, nil, nil, true) + if err == nil || !strings.Contains(err.Error(), "cannot merge files with different schemas") { + t.Fatalf("merge: %v, want schema mismatch", err) + } +} diff --git a/weed/worker/tasks/iceberg/filer_io.go b/weed/worker/tasks/iceberg/filer_io.go index a1c8e1c5be..482c8dc80a 100644 --- a/weed/worker/tasks/iceberg/filer_io.go +++ b/weed/worker/tasks/iceberg/filer_io.go @@ -1,6 +1,7 @@ package iceberg import ( + "bytes" "context" "crypto/rand" "encoding/hex" @@ -14,6 +15,7 @@ import ( "sync" "time" + "github.com/parquet-go/parquet-go" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/operation" @@ -160,9 +162,9 @@ func loadCurrentMetadata(ctx context.Context, client filer_pb.SeaweedFilerClient return state, nil } -// loadFileByIcebergPath loads a file from the filer given an Iceberg-style path. -// dataPath is the bucket-relative directory the table's files live in. -func loadFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, dataPath, icebergPath string) ([]byte, error) { +// lookupFileByIcebergPath returns the filer entry holding an Iceberg-style +// path. dataPath is the bucket-relative directory the table's files live in. +func lookupFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, dataPath, icebergPath string) (*filer_pb.Entry, error) { fullPath, err := icebergFilerPath(bucketName, dataPath, icebergPath) if err != nil { return nil, err @@ -179,23 +181,65 @@ func loadFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClie if resp == nil || resp.Entry == nil { return nil, fmt.Errorf("file not found: %s/%s", dir, fileName) } + return resp.Entry, nil +} - if len(resp.Entry.Content) > 0 || len(resp.Entry.Chunks) == 0 { - return resp.Entry.Content, nil +// loadFileByIcebergPath loads a file from the filer given an Iceberg-style path. +func loadFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, dataPath, icebergPath string) ([]byte, error) { + entry, err := lookupFileByIcebergPath(ctx, client, bucketName, dataPath, icebergPath) + if err != nil { + return nil, err + } + + if len(entry.Content) > 0 || len(entry.Chunks) == 0 { + return entry.Content, nil } initGlobalHTTPClientOnce.Do(util_http.InitGlobalHttpClient) - reader := filer.NewFileReader(singleFilerClient{client: client}, resp.Entry) + reader := filer.NewFileReader(singleFilerClient{client: client}, entry) if closer, ok := reader.(io.Closer); ok { defer closer.Close() } data, err := io.ReadAll(reader) if err != nil { - return nil, fmt.Errorf("read chunked file %s/%s: %w", dir, fileName, err) + return nil, fmt.Errorf("read chunked file %s: %w", icebergPath, err) } return data, nil } +// parquetSchemaByIcebergPath reads a file's parquet schema from its footer, +// so the file's body is never fetched: a chunked file's ReadAt pulls only the +// chunks the footer spans. +func parquetSchemaByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, dataPath, icebergPath string) (*parquet.Schema, error) { + entry, err := lookupFileByIcebergPath(ctx, client, bucketName, dataPath, icebergPath) + if err != nil { + return nil, err + } + + var readerAt io.ReaderAt + size := int64(entry.GetAttributes().GetFileSize()) + if len(entry.Content) > 0 || len(entry.Chunks) == 0 { + readerAt = bytes.NewReader(entry.Content) + size = int64(len(entry.Content)) + } else { + initGlobalHTTPClientOnce.Do(util_http.InitGlobalHttpClient) + reader := filer.NewChunkStreamReader(singleFilerClient{client: client}, entry.Chunks) + defer reader.Close() + readerAt = reader + if size <= 0 { + if size, err = reader.Seek(0, io.SeekEnd); err != nil { + return nil, fmt.Errorf("size of %s: %w", icebergPath, err) + } + } + } + + f, err := parquet.OpenFile(readerAt, size) + if err != nil { + return nil, fmt.Errorf("open parquet file %s: %w", icebergPath, err) + } + return f.Schema(), nil +} + // normalizeIcebergPath converts an Iceberg path (an S3 URL, an absolute filer // path, or a path relative to the table root) into a path relative to the // bucket root, e.g. "ns/table/metadata/snap-1.avro". Absolute references are diff --git a/weed/worker/tasks/iceberg/row_lineage.go b/weed/worker/tasks/iceberg/row_lineage.go index 48e7250577..af53329d6e 100644 --- a/weed/worker/tasks/iceberg/row_lineage.go +++ b/weed/worker/tasks/iceberg/row_lineage.go @@ -210,7 +210,8 @@ func (l *lineageLayout) mapperFor(input *parquet.Schema, src lineageSource) (*li next++ continue } - if base >= len(l.baseFields) || f.Name() != l.baseFields[base].Name() || !parquet.EqualNodes(f, l.baseFields[base]) { + if base >= len(l.baseFields) || f.Name() != l.baseFields[base].Name() || !parquet.EqualNodes(f, l.baseFields[base]) || + !fieldIDsAgree(f, l.baseFields[base]) { return nil, fmt.Errorf("cannot merge files with different schemas") } for i := 0; i < n; i++ { diff --git a/weed/worker/tasks/iceberg/schema_evolution.go b/weed/worker/tasks/iceberg/schema_evolution.go new file mode 100644 index 0000000000..85b5647b79 --- /dev/null +++ b/weed/worker/tasks/iceberg/schema_evolution.go @@ -0,0 +1,184 @@ +package iceberg + +import ( + "fmt" + "reflect" + + "github.com/parquet-go/parquet-go" +) + +// The bin-pack merge of a v1/v2 table whose data files spell its schema +// differently. A bin used to take the first file's Parquet schema as the +// writer's and refuse every file whose schema was not structurally identical +// ("cannot merge files with different schemas"). That refused: +// - schema evolution: a column added to the table after some files were +// written; +// - one Iceberg schema written REQUIRED by one engine and OPTIONAL by another. +// +// The writer's schema is now the union of the bin's files' schemas, matched +// by Iceberg field id and computed before any row is written (schemaUnion), +// so every file converts into it in a single pass: +// - the same field id carries the same name and type; +// - the writer's column may be optional where the file's is required, never +// the reverse; +// - a writer column the file lacks must be optional, and reads as null. +// +// Anything else (a rename, a type change, the same name under different field +// ids, nested columns, no field ids) is refused as before. + +// flatColumnsByFieldID is the schema's columns by Iceberg field id when every +// column is a top-level, non-repeated leaf with a field id of its own; ok is +// false otherwise. +func flatColumnsByFieldID(s *parquet.Schema) (map[int]parquet.Field, bool) { + if s == nil { + return nil, false + } + out := make(map[int]parquet.Field) + for _, f := range s.Fields() { + if !f.Leaf() || f.Repeated() || f.ID() <= 0 { + return nil, false + } + if _, dup := out[f.ID()]; dup { + return nil, false + } + out[f.ID()] = f + } + return out, true +} + +// schemaCovers reports whether every row of a file written in source converts +// to target without losing or inventing a value (the rules above). +func schemaCovers(target, source *parquet.Schema) bool { + targetCols, ok := flatColumnsByFieldID(target) + if !ok { + return false + } + sourceCols, ok := flatColumnsByFieldID(source) + if !ok { + return false + } + for id, s := range sourceCols { + t, ok := targetCols[id] + if !ok || t.Name() != s.Name() || !parquet.EqualTypes(t.Type(), s.Type()) { + return false + } + if t.Required() && !s.Required() { + return false + } + } + for id, t := range targetCols { + if _, ok := sourceCols[id]; !ok && !t.Optional() { + return false + } + } + return true +} + +// schemaUnion folds a bin's file schemas into the writer schema the merge +// needs: every field id any file carries, in the order first seen, under the +// name and type all of its files agree on. A column is REQUIRED only when +// every file has it and spells it so; a file that lacks it or spells it +// optional reads and writes nulls there. A shared field id with a different +// name or type, or the same name under different field ids (a column dropped +// and re-added), is a conflict the merge refuses. +type schemaUnion struct { + fields []unionField + byID map[int]int + byName map[string]int + folds int +} + +type unionField struct { + field parquet.Field // the field as first seen + required bool // required in every file that carries it + presentInAll bool // present in every file folded so far +} + +func (u *schemaUnion) fold(s *parquet.Schema, path string) error { + cols, ok := flatColumnsByFieldID(s) + if !ok { + return fmt.Errorf("schema of %s does not key flat columns by field id", path) + } + if u.byID == nil { + u.byID, u.byName = make(map[int]int), make(map[string]int) + } + for _, f := range s.Fields() { + id, name := f.ID(), f.Name() + if j, ok := u.byID[id]; ok { + e := &u.fields[j] + if e.field.Name() != name || !parquet.EqualTypes(e.field.Type(), f.Type()) { + return fmt.Errorf("field id %d is %q in one file but %q in %s", id, e.field.Name(), name, path) + } + e.required = e.required && f.Required() + continue + } + if _, ok := u.byName[name]; ok { + return fmt.Errorf("column %q carries different field ids in %s", name, path) + } + u.byID[id], u.byName[name] = len(u.fields), id + // A field folded in after the first schema is absent from every + // file seen before it. + u.fields = append(u.fields, unionField{field: f, required: f.Required(), presentInAll: u.folds == 0}) + } + for i := range u.fields { + if _, ok := cols[u.fields[i].field.ID()]; !ok { + u.fields[i].presentInAll = false + } + } + u.folds++ + return nil +} + +// schema is the union as a writer schema; name and group give the root node +// (their Go type is never read: the merge writes parquet.Row values). +func (u *schemaUnion) schema(name string, group parquet.Node) *parquet.Schema { + fields := make([]parquet.Field, len(u.fields)) + for i, e := range u.fields { + fields[i] = e.field + if !e.field.Optional() && !(e.required && e.presentInAll) { + fields[i] = unionOptionalField{Node: parquet.Optional(e.field), name: e.field.Name()} + } + } + return parquet.NewSchema(name, &orderedGroup{Node: group, fields: fields}) +} + +// unionOptionalField spells a field optional in the union schema when the +// file it came from wrote it required. +type unionOptionalField struct { + parquet.Node + name string +} + +func (f unionOptionalField) Name() string { return f.name } +func (f unionOptionalField) Value(base reflect.Value) reflect.Value { return reflect.Value{} } + +// widerSchemaError stops a merge that met a file whose schema covers the +// writer's: the bin is merged again with it. +type widerSchemaError struct { + schema *parquet.Schema + path string +} + +func (e *widerSchemaError) Error() string { + return fmt.Sprintf("schema of %s is wider than the bin's writer schema", e.path) +} + +// evolvedFileConversion is how a file of schema source joins a merge written +// in target: nil when they are equal, a conversion when target covers it, a +// *widerSchemaError when it covers target, else the merge's mismatch error. +func evolvedFileConversion(target, source *parquet.Schema, path string) (parquet.Conversion, error) { + if schemasEqual(target, source) { + return nil, nil + } + if schemaCovers(target, source) { + conv, err := parquet.Convert(target, source) + if err != nil { + return nil, fmt.Errorf("schema of %s: %w", path, err) + } + return conv, nil + } + if schemaCovers(source, target) { + return nil, &widerSchemaError{schema: source, path: path} + } + return nil, fmt.Errorf("schema mismatch in %s: cannot merge files with different schemas", path) +}