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) +}