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

@@ -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

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) {