From 160490b55d0572b98d3561aa2f3d2fbbacaaa344 Mon Sep 17 00:00:00 2001 From: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 20 Oct 2026 17:54:11 -0700 Subject: [PATCH] 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> --- weed/worker/tasks/iceberg/compact.go | 20 ++++- .../tasks/iceberg/compact_schema_test.go | 78 +++++++++++++++++++ 2 files changed, 95 insertions(+), 3 deletions(-) diff --git a/weed/worker/tasks/iceberg/compact.go b/weed/worker/tasks/iceberg/compact.go index f627e99e83..c5c2721ae1 100644 --- a/weed/worker/tasks/iceberg/compact.go +++ b/weed/worker/tasks/iceberg/compact.go @@ -993,22 +993,30 @@ 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) { if len(fieldIDs) == 0 { @@ -1028,6 +1036,12 @@ func resolveEqualityColIndices(pqSchema *parquet.Schema, fieldIDs []int, iceberg } 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 !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 diff --git a/weed/worker/tasks/iceberg/compact_schema_test.go b/weed/worker/tasks/iceberg/compact_schema_test.go index 05901d8679..8171a50d99 100644 --- a/weed/worker/tasks/iceberg/compact_schema_test.go +++ b/weed/worker/tasks/iceberg/compact_schema_test.go @@ -47,6 +47,20 @@ type readdedNameSchemaRow struct { // name re-added under field id 3 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, @@ -373,6 +387,70 @@ func TestMergeParquetFilesEvolvedSchemaEqualityDelete(t *testing.T) { }) } +// 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"}}) + } + }) +} + // The v3 merge matches files on the first file's columns; the re-added // column is refused there too. func TestMergeParquetFilesV3ReaddedColumn(t *testing.T) {