iceberg maintenance: apply equality deletes only to older data files in their partition (#12187)

* iceberg maintenance: apply equality deletes only to older data files in their partition

Compaction applied every equality-delete key to every data file of every
bin. The Iceberg spec ("Scan Planning") applies an equality delete file only
to data files with a strictly lower data sequence number, and only to data
files in the delete file's own partition unless its spec is unpartitioned.

An upsert writer (Flink, the Iceberg Kafka Connect sink, CDC tools) commits
an update as the row's new version plus an equality delete of its key in the
same commit, so the compaction deleted both the old and the new version of
every updated row.

- equalityDeleteGroup is keyed by equality field ids and scope (unpartitioned,
  or spec id plus partition) and keeps, per key, the highest data sequence
  number of the delete files carrying it. A delete file without a data
  sequence number is an error rather than a guess.
- Partition values are compared typed through the spec's partition type, so
  one partition decoded in different Go shapes (a uuid as uuid.UUID or
  [16]byte, a long as int32 or int64, a null present or missing) is one
  partition. A spec whose fields are all void is unpartitioned. A delete whose
  scope cannot be decided (unknown spec, no partition values under a
  partitioned spec, a value that does not convert) stops the compaction; a bin
  whose own partition cannot be typed while a delete of its spec exists is
  not compacted.
- Each bin is merged with only the groups that apply to its partition, and a
  row is deleted only by a key whose sequence number is higher than its data
  file's.
- A where filter no longer drops unpartitioned delete files (the clause's
  field is not in their spec), which let deleted rows return in a filtered
  compaction.

Tests: the table fixture now commits its delete files after the data (data
sequence number 1, deletes 2), as writers do; hand-built entries in the unit
tests get sequence numbers. New eqdelete_sequence_test.go covers rows written
with and after their delete, partition scope, unpartitioned and all-void
specs, typed partition matching, missing sequence numbers and partition
values, where filters, and group-key collisions.

* iceberg maintenance: split the undecidable-scope branch in deleteInPredicateScope

---------

Co-authored-by: user.email <chrislusf@users.noreply.github.com>

weed/worker/tasks/iceberg/compact.go

@@ -140,18 +140,12 @@ func (h *Handler) compactDataFiles(
 			if entry.DataFile().FileFormat() == iceberg.PuffinFile {
 				continue
 			}
-			if predicate != nil {
-				spec, ok := specsByID[int(entry.DataFile().SpecID())]
-				if !ok {
-					continue
-				}
-				match, err := predicate.Matches(spec, entry.DataFile().Partition())
-				if err != nil {
-					return "", nil, err
-				}
-				if !match {
-					continue
-				}
+			inScope, err := deleteInPredicateScope(entry, predicate, specsByID, schema)
+			if err != nil {
+				return "", nil, err
+			}
+			if !inScope {
+				continue
 			}
 			switch entry.DataFile().ContentType() {
 			case iceberg.EntryContentPosDeletes:
@@ -169,7 +163,7 @@ func (h *Handler) compactDataFiles(
 		}
 
 		if len(eqDeleteEntries) > 0 {
-			eqDeleteGroups, err = collectEqualityDeletes(ctx, filerClient, bucketName, dataPath, eqDeleteEntries, schema)
+			eqDeleteGroups, err = collectEqualityDeletes(ctx, filerClient, bucketName, dataPath, eqDeleteEntries, schema, specsByID)
 			if err != nil {
 				return "", nil, fmt.Errorf("collect equality deletes: %w", err)
 			}
@@ -291,10 +285,18 @@ func (h *Handler) compactDataFiles(
 		var mergedData []byte
 		var recordCount int64
 		rowGroupRows := rowsPerRowGroup(bin, config)
+		// Only the equality deletes of the bin's partition (or unpartitioned
+		// ones) apply to it. When that cannot be told, the bin is left as it
+		// is rather than merged without deletes that may apply.
+		binEqDeleteGroups, scopeErr := eqDeleteGroupsForBin(eqDeleteGroups, bin, specsByID, schema)
+		if scopeErr != nil {
+			glog.Warningf("iceberg compact: bin %d (%d files) not compacted: %v", binIdx, len(bin.Entries), scopeErr)
+			goto binDone
+		}
 		if rewritePlan != nil && rewritePlan.strategy == "sort" {
-			mergedData, recordCount, err = mergeParquetFilesSorted(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rewritePlan, rowGroupRows, lineage)
+			mergedData, recordCount, err = mergeParquetFilesSorted(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, binEqDeleteGroups, schema, rewritePlan, rowGroupRows, lineage)
 		} else {
-			mergedData, recordCount, err = mergeParquetFiles(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rowGroupRows, lineage)
+			mergedData, recordCount, err = mergeParquetFiles(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, binEqDeleteGroups, schema, rowGroupRows, lineage)
 		}
 		if err != nil {
 			glog.Warningf("iceberg compact: failed to merge bin %d (%d files): %v", binIdx, len(bin.Entries), err)
@@ -898,28 +900,52 @@ func readPositionDeleteFile(
 	return result, nil
 }
 
-// equalityDeleteGroup holds a set of delete keys for a specific set of equality field IDs.
-// Different equality delete files may use different field IDs, so deletes are grouped.
+// equalityDeleteGroup holds the delete keys of the equality delete files that
+// share one set of equality field IDs and one scope. An equality delete file
+// applies only to data files with a lower data sequence number, and only in
+// its own partition unless its spec is unpartitioned (Iceberg spec, "Scan
+// Planning"): each key keeps the highest data sequence number of the delete
+// files carrying it, and the group the partition it applies to.
 type equalityDeleteGroup struct {
 	FieldIDs []int
-	Keys     map[string]struct{}
+	Keys     map[string]int64 // key -> highest data sequence number of a delete file carrying it
+	Scope    deleteScope      // where the group applies
+}
+
+// deleteInPredicateScope reports whether a delete file may apply to a bin a
+// where filter keeps: an unpartitioned one always (it has no field a clause
+// could match, and applies to every partition), one whose scope cannot be
+// decided always (collectEqualityDeletes then refuses it rather than drop it),
+// a partitioned one when its partition matches.
+func deleteInPredicateScope(entry iceberg.ManifestEntry, predicate *partitionPredicate, specs map[int]iceberg.PartitionSpec, schema *iceberg.Schema) (bool, error) {
+	if predicate == nil {
+		return true, nil
+	}
+	scope, err := deleteFileScope(entry.DataFile(), specs, schema)
+	if err != nil {
+		return true, nil
+	}
+	if scope.global {
+		return true, nil
+	}
+	return predicate.Matches(specs[int(entry.DataFile().SpecID())], entry.DataFile().Partition())
 }
 
 // collectEqualityDeletes reads equality delete Parquet files and returns groups
-// of delete keys, one per distinct set of equality field IDs. This correctly
-// handles the case where different delete files use different equality columns.
+// of delete keys, one per distinct set of equality field IDs and scope. This
+// correctly handles the case where different delete files use different
+// equality columns. A delete file without a data sequence number cannot be
+// ordered against the rows it may delete, and is an error.
 func collectEqualityDeletes(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
 	bucketName, dataPath string,
 	deleteEntries []iceberg.ManifestEntry,
 	schema *iceberg.Schema,
+	specs map[int]iceberg.PartitionSpec,
 ) ([]equalityDeleteGroup, error) {
-	type groupState struct {
-		fieldIDs []int
-		keys     map[string]struct{}
-	}
-	groups := make(map[string]*groupState)
+	groups := make(map[string]*equalityDeleteGroup)
+	var order []string
 
 	for _, entry := range deleteEntries {
 		if entry.DataFile().ContentType() != iceberg.EntryContentEqDeletes {
@@ -930,11 +956,21 @@ func collectEqualityDeletes(
 			continue
 		}
 
-		groupKey := fmt.Sprint(eqFieldIDs)
+		seq := entry.SequenceNum()
+		if seq < 0 {
+			return nil, fmt.Errorf("equality delete file %s has no data sequence number", entry.DataFile().FilePath())
+		}
+		scope, err := deleteFileScope(entry.DataFile(), specs, schema)
+		if err != nil {
+			return nil, err
+		}
+
+		groupKey := fmt.Sprintf("%v\x00%s", eqFieldIDs, scope.key())
 		gs, ok := groups[groupKey]
 		if !ok {
-			gs = &groupState{fieldIDs: eqFieldIDs, keys: make(map[string]struct{})}
+			gs = &equalityDeleteGroup{FieldIDs: eqFieldIDs, Keys: make(map[string]int64), Scope: scope}
 			groups[groupKey] = gs
+			order = append(order, groupKey)
 		}
 
 		keys, err := readEqualityDeleteFile(ctx, filerClient, bucketName, dataPath, entry.DataFile().FilePath(), eqFieldIDs, schema)
@@ -942,13 +978,15 @@ func collectEqualityDeletes(
 			return nil, fmt.Errorf("read equality delete file %s: %w", entry.DataFile().FilePath(), err)
 		}
 		for k := range keys {
-			gs.keys[k] = struct{}{}
+			if cur, seen := gs.Keys[k]; !seen || seq > cur {
+				gs.Keys[k] = seq
+			}
 		}
 	}
 
 	result := make([]equalityDeleteGroup, 0, len(groups))
-	for _, gs := range groups {
-		result = append(result, equalityDeleteGroup{FieldIDs: gs.fieldIDs, Keys: gs.keys})
+	for _, k := range order {
+		result = append(result, *groups[k])
 	}
 	return result, nil
 }
@@ -1331,7 +1369,19 @@ func mergeParquetFilesInto(
 
 type resolvedEqDeleteGroup struct {
 	colIndices []int
-	keys       map[string]struct{}
+	keys       map[string]int64 // key -> highest data sequence number of a delete carrying it
+}
+
+// equalityDeletesFor returns the data sequence number of entry's data file: a
+// row is deleted only by a key whose delete files' sequence number is higher.
+// The groups are already the bin's (eqDeleteGroupsForBin). A data file without
+// one, when deletes may apply to it, is an error rather than a guess.
+func equalityDeletesFor(groups []resolvedEqDeleteGroup, entry iceberg.ManifestEntry) ([]resolvedEqDeleteGroup, int64, error) {
+	dataSeq := entry.SequenceNum()
+	if len(groups) > 0 && dataSeq < 0 {
+		return nil, 0, fmt.Errorf("data file %s has no data sequence number to order its equality deletes against", entry.DataFile().FilePath())
+	}
+	return groups, dataSeq, nil
 }
 
 func resolveEqualityDeleteGroupsForSchema(
@@ -1360,6 +1410,7 @@ func visitFilteredParquetRows(
 	source, bucketName, dataPath string,
 	positionDeletes map[string][]int64,
 	resolvedEqGroups []resolvedEqDeleteGroup,
+	dataSeq int64, // the file's data sequence number: only later deletes apply to it
 	onRows func(rows []parquet.Row, positions []int64) error,
 ) (int64, error) {
 	defer reader.Close()
@@ -1402,7 +1453,7 @@ func visitFilteredParquetRows(
 				deleted := false
 				for _, g := range resolvedEqGroups {
 					key := buildEqualityKey(rows[i], g.colIndices)
-					if _, ok := g.keys[key]; ok {
+					if deleteSeq, ok := g.keys[key]; ok && deleteSeq > dataSeq {
 						deleted = true
 						break
 					}

weed/worker/tasks/iceberg/compact_schema_test.go

@@ -69,7 +69,7 @@ func schemaFileEntry(t *testing.T, path string, recordCount int64) iceberg.Manif
 		t.Fatalf("build data file: %v", err)
 	}
 	snapID := int64(1)
-	return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build())
+	return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, &testDataSeq, &testDataSeq, dfb.Build())
 }
 
 func mergeNamedFiles(t *testing.T, client filer_pb.SeaweedFilerClient, names []string, counts map[string]int64,
@@ -356,10 +356,10 @@ func TestMergeParquetFilesEvolvedSchemaEqualityDelete(t *testing.T) {
 	}
 	eqDfb.EqualityFieldIDs([]int{2})
 	snapID := int64(1)
-	eqEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, eqDfb.Build())
+	eqEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, &testDeleteSeq, &testDeleteSeq, eqDfb.Build())
 
 	eqGroups, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl",
-		[]iceberg.ManifestEntry{eqEntry}, newTestSchema())
+		[]iceberg.ManifestEntry{eqEntry}, newTestSchema(), testUnpartitionedSpecs)
 	if err != nil {
 		t.Fatalf("collectEqualityDeletes: %v", err)
 	}
@@ -478,11 +478,11 @@ func TestMergeParquetFilesEqualityDeleteAddedColumn(t *testing.T) {
 		return map[string]int64{"narrow.parquet": 3, "wide.parquet": 2}
 	}
 	eqGroups := func(keys ...parquet.Value) []equalityDeleteGroup {
-		set := make(map[string]struct{}, len(keys))
+		set := make(map[string]int64, len(keys))
 		for _, k := range keys {
-			set[buildEqualityKey(parquet.Row{k}, []int{0})] = struct{}{}
+			set[buildEqualityKey(parquet.Row{k}, []int{0})] = 2 // committed after the data files (sequence number 1)
 		}
-		return []equalityDeleteGroup{{FieldIDs: []int{3}, Keys: set}}
+		return []equalityDeleteGroup{{FieldIDs: []int{3}, Keys: set, Scope: deleteScope{global: true}}}
 	}
 	schema := newAddedColumnSchema()
 
@@ -568,10 +568,10 @@ func TestMergeParquetFilesEqualityDeleteRenamedColumn(t *testing.T) {
 	}
 	eqDfb.EqualityFieldIDs([]int{2})
 	snapID := int64(1)
-	eqEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, eqDfb.Build())
+	eqEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, &testDeleteSeq, &testDeleteSeq, eqDfb.Build())
 
 	eqGroups, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl",
-		[]iceberg.ManifestEntry{eqEntry}, schema)
+		[]iceberg.ManifestEntry{eqEntry}, schema, testUnpartitionedSpecs)
 	if err != nil {
 		t.Fatalf("collectEqualityDeletes: %v", err)
 	}
@@ -619,10 +619,10 @@ func TestEqualityDeleteFileMissingColumn(t *testing.T) {
 	}
 	eqDfb.EqualityFieldIDs([]int{2})
 	snapID := int64(1)
-	eqEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, eqDfb.Build())
+	eqEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, &testDeleteSeq, &testDeleteSeq, eqDfb.Build())
 
 	_, err = collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl",
-		[]iceberg.ManifestEntry{eqEntry}, newAddedColumnSchema())
+		[]iceberg.ManifestEntry{eqEntry}, newAddedColumnSchema(), testUnpartitionedSpecs)
 	if err == nil {
 		t.Fatal("collectEqualityDeletes: want an error for a missing equality column")
 	}

weed/worker/tasks/iceberg/eqdelete_scope.go

@@ -0,0 +1,239 @@
+package iceberg
+
+import (
+	"encoding/hex"
+	"fmt"
+	"math"
+	"reflect"
+	"strconv"
+	"strings"
+
+	"github.com/apache/iceberg-go"
+	"github.com/google/uuid"
+)
+
+// Where a delete file applies, decided on typed partition values. Partition maps come from manifests written by other
+// engines, and one value can arrive in different Go shapes (a uuid as uuid.UUID, [16]byte or []byte; an int as int32 or
+// int64; a null as nil or as a missing field), so comparing them as JSON strings misses equal partitions, and a delete
+// that misses its partition is skipped: its deleted rows come back in the merged file for good. Every value is converted
+// through the type the spec gives its field; a value that cannot be converted, a spec the table does not list, or a
+// partitioned spec with no partition values makes the scope undecidable, and an undecidable scope fails closed: a
+// delete file whose scope cannot be decided stops the compaction, and a bin whose own partition cannot be typed while a
+// delete of its spec exists is not compacted.
+
+// deleteScope is where a delete file may apply: every partition (global), or the data files of specID whose typed
+// partition key is partKey.
+type deleteScope struct {
+	global  bool
+	specID  int32
+	partKey string
+}
+
+// covers reports whether the scope reaches a bin of specID whose typed partition key is partKey.
+func (s deleteScope) covers(specID int32, partKey string) bool {
+	return s.global || (specID == s.specID && partKey == s.partKey)
+}
+
+// key is the scope as a map key, its parts separated by NUL so no two scopes collide.
+func (s deleteScope) key() string {
+	if s.global {
+		return "global"
+	}
+	return fmt.Sprintf("%d\x00%s", s.specID, s.partKey)
+}
+
+// deleteFileScope is the scope of a delete file. A spec with no fields or only void ones is global; any other spec
+// needs a partition value for each of its fields, typed.
+func deleteFileScope(df iceberg.DataFile, specs map[int]iceberg.PartitionSpec, schema *iceberg.Schema) (deleteScope, error) {
+	spec, ok := specs[int(df.SpecID())]
+	if !ok {
+		return deleteScope{}, fmt.Errorf("delete file %s names partition spec %d, which the table does not list", df.FilePath(), df.SpecID())
+	}
+	if spec.IsUnpartitioned() {
+		return deleteScope{global: true}, nil
+	}
+	if len(df.Partition()) == 0 {
+		return deleteScope{}, fmt.Errorf("delete file %s has no partition values under partitioned spec %d", df.FilePath(), df.SpecID())
+	}
+	key, err := typedPartitionKey(spec, schema, df.Partition())
+	if err != nil {
+		return deleteScope{}, fmt.Errorf("delete file %s: %w", df.FilePath(), err)
+	}
+	return deleteScope{specID: df.SpecID(), partKey: key}, nil
+}
+
+// binTypedPartitionKey is the typed partition key of a bin: every file of a bin shares its spec and partition.
+func binTypedPartitionKey(bin compactionBin, specs map[int]iceberg.PartitionSpec, schema *iceberg.Schema) (string, error) {
+	spec, ok := specs[int(bin.SpecID)]
+	if !ok {
+		return "", fmt.Errorf("partition spec %d is not listed by the table", bin.SpecID)
+	}
+	if spec.IsUnpartitioned() {
+		return "", nil
+	}
+	return typedPartitionKey(spec, schema, bin.Partition)
+}
+
+// eqDeleteGroupsForBin narrows the equality delete groups to those that apply to bin. When a group of the bin's spec
+// exists and the bin's own partition cannot be typed, it cannot be told whether the group applies: an error, and the
+// bin is not compacted.
+func eqDeleteGroupsForBin(groups []equalityDeleteGroup, bin compactionBin, specs map[int]iceberg.PartitionSpec, schema *iceberg.Schema) ([]equalityDeleteGroup, error) {
+	if len(groups) == 0 {
+		return nil, nil
+	}
+	binKey, keyErr := binTypedPartitionKey(bin, specs, schema)
+	var applicable []equalityDeleteGroup
+	for _, g := range groups {
+		if g.Scope.global {
+			applicable = append(applicable, g)
+			continue
+		}
+		if g.Scope.specID != bin.SpecID {
+			continue
+		}
+		if keyErr != nil {
+			return nil, fmt.Errorf("equality deletes of spec %d may apply, but the bin's partition cannot be compared: %w", bin.SpecID, keyErr)
+		}
+		if g.Scope.partKey == binKey {
+			applicable = append(applicable, g)
+		}
+	}
+	return applicable, nil
+}
+
+// typedPartitionKey encodes a partition map field by field in the spec's order, each value converted through the
+// type the spec gives its field. A field missing from the map is null, as a null value is.
+func typedPartitionKey(spec iceberg.PartitionSpec, schema *iceberg.Schema, partition map[int]any) (string, error) {
+	if schema == nil {
+		return "", fmt.Errorf("no table schema to type the partition values against")
+	}
+	fields := spec.PartitionType(schema).FieldList
+	parts := make([]string, 0, len(fields))
+	for _, f := range fields {
+		v, err := typedPartitionValue(f.Type, partition[f.ID])
+		if err != nil {
+			return "", fmt.Errorf("partition field %d (%s): %w", f.ID, f.Name, err)
+		}
+		parts = append(parts, strconv.Itoa(f.ID)+"="+v)
+	}
+	return strings.Join(parts, "\x00"), nil
+}
+
+// literalAny unwraps an iceberg literal to its Go value.
+type literalAny interface{ Any() any }
+
+func typedPartitionValue(typ iceberg.Type, v any) (string, error) {
+	switch v.(type) {
+	case iceberg.AboveMaxLiteral, iceberg.BelowMinLiteral:
+		return "", fmt.Errorf("an out-of-range literal %v", v)
+	}
+	if lit, ok := v.(literalAny); ok {
+		v = lit.Any()
+	}
+	if v == nil {
+		return "null", nil
+	}
+	if rv := reflect.ValueOf(v); rv.Kind() == reflect.Pointer {
+		if rv.IsNil() {
+			return "null", nil
+		}
+		v = rv.Elem().Interface()
+	}
+	switch t := typ.(type) {
+	case iceberg.UnknownType:
+		return "null", nil // a dropped source field: its values are null by definition
+	case iceberg.BooleanType:
+		if b, ok := v.(bool); ok {
+			return strconv.FormatBool(b), nil
+		}
+	case iceberg.Int32Type, iceberg.Int64Type, iceberg.DateType, iceberg.TimeType, iceberg.TimestampType,
+		iceberg.TimestampTzType, iceberg.TimestampNsType, iceberg.TimestampTzNsType:
+		if i, ok := integerValue(v); ok {
+			return "i" + strconv.FormatInt(i, 10), nil
+		}
+	case iceberg.Float32Type, iceberg.Float64Type:
+		var f float64
+		switch x := v.(type) {
+		case float32:
+			f = float64(x)
+		case float64:
+			f = x
+		default:
+			return "", fmt.Errorf("value %v (%T) does not convert to %s", v, v, typ)
+		}
+		switch {
+		case math.IsNaN(f):
+			return "fNaN", nil
+		case f == 0:
+			f = 0 // -0 and +0 are one partition value
+		}
+		return "f" + strconv.FormatUint(math.Float64bits(f), 16), nil
+	case iceberg.StringType:
+		switch s := v.(type) {
+		case string:
+			return strconv.Quote(s), nil
+		case []byte:
+			return strconv.Quote(string(s)), nil
+		}
+	case iceberg.UUIDType:
+		switch u := v.(type) {
+		case uuid.UUID:
+			return "u" + u.String(), nil
+		case string:
+			if p, err := uuid.Parse(u); err == nil {
+				return "u" + p.String(), nil
+			}
+		default:
+			if b, ok := byteValue(v); ok && len(b) == 16 {
+				return "u" + uuid.UUID(b).String(), nil
+			}
+		}
+	case iceberg.FixedType, iceberg.BinaryType:
+		if s, ok := v.(string); ok {
+			return "b" + hex.EncodeToString([]byte(s)), nil
+		}
+		if b, ok := byteValue(v); ok {
+			return "b" + hex.EncodeToString(b), nil
+		}
+	case iceberg.DecimalType:
+		if d, ok := v.(iceberg.Decimal); ok {
+			if d.Scale != t.Scale() {
+				return "", fmt.Errorf("decimal of scale %d under type %s", d.Scale, t)
+			}
+			return "d" + d.Val.ToString(int32(t.Scale())), nil
+		}
+	default:
+		return "", fmt.Errorf("partition values of type %s are not compared", typ)
+	}
+	return "", fmt.Errorf("value %v (%T) does not convert to %s", v, v, typ)
+}
+
+// integerValue reads any Go integer kind (iceberg.Date, Time and Timestamp included) as int64.
+func integerValue(v any) (int64, bool) {
+	rv := reflect.ValueOf(v)
+	switch rv.Kind() {
+	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
+		return rv.Int(), true
+	case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32:
+		return int64(rv.Uint()), true
+	case reflect.Uint64:
+		if u := rv.Uint(); u <= math.MaxInt64 {
+			return int64(u), true
+		}
+	}
+	return 0, false
+}
+
+// byteValue reads a []byte or a byte array of any length ([16]byte, uuid.UUID) as bytes.
+func byteValue(v any) ([]byte, bool) {
+	if b, ok := v.([]byte); ok {
+		return b, true
+	}
+	rv := reflect.ValueOf(v)
+	if rv.Kind() == reflect.Array && rv.Type().Elem().Kind() == reflect.Uint8 {
+		b := make([]byte, rv.Len())
+		reflect.Copy(reflect.ValueOf(b), rv)
+		return b, true
+	}
+	return nil, false
+}

weed/worker/tasks/iceberg/eqdelete_sequence_test.go

@@ -0,0 +1,445 @@
+package iceberg
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"path"
+	"sort"
+	"strings"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	"github.com/google/uuid"
+	"github.com/parquet-go/parquet-go"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
+)
+
+// testDataSeq and testDeleteSeq are the data sequence numbers of the unit
+// tests' hand-built entries: the delete files committed after the data.
+var testDataSeq, testDeleteSeq = int64(1), int64(2)
+
+// testUnpartitionedSpecs is the spec list of the unit tests' unpartitioned tables.
+var testUnpartitionedSpecs = map[int]iceberg.PartitionSpec{0: *iceberg.UnpartitionedSpec}
+
+type idNameRow struct {
+	ID   int64  `parquet:"id"`
+	Name string `parquet:"name"`
+}
+
+func readIDNameRows(t *testing.T, content []byte) []idNameRow {
+	t.Helper()
+	reader := parquet.NewReader(bytes.NewReader(content))
+	defer reader.Close()
+	var rows []idNameRow
+	for {
+		var r idNameRow
+		if err := reader.Read(&r); err != nil {
+			if err == io.EOF {
+				break
+			}
+			t.Fatalf("read: %v", err)
+		}
+		rows = append(rows, r)
+	}
+	sort.Slice(rows, func(i, j int) bool { return rows[i].ID < rows[j].ID })
+	return rows
+}
+
+// An upsert writer (Flink, the Iceberg Kafka Connect sink, CDC tools) commits
+// an update as the row's new version plus an equality delete of its key, in
+// one commit. The delete applies only to data files with a lower data sequence
+// number: the version written with it, and any written later, stay. Applying
+// every key to every file drops both versions of every updated row.
+func TestCompactionAppliesEqualityDeletesOnlyToOlderDataFiles(t *testing.T) {
+	skipIfEqualityDeleteAvroUnsupported(t)
+	fs, client := startFakeFiler(t)
+
+	setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl",
+		// d1 was committed before the deletes (1), d2 with them (2), d3 after them (3)
+		DataSequenceNumbers: map[string]int64{"d2.parquet": 2, "d3.parquet": 3}}
+	type rows = []struct {
+		ID   int64
+		Name string
+	}
+	populateTableWithDeleteFiles(t, fs, setup,
+		[]struct {
+			Name string
+			Rows rows
+		}{
+			{"d1.parquet", rows{{1, "alice"}, {2, "bob"}, {3, "carol"}}},
+			{"d2.parquet", rows{{1, "alice v2"}}},
+			{"d3.parquet", rows{{2, "bob v3"}}},
+		},
+		nil,
+		[]struct {
+			Name     string
+			FieldIDs []int
+			Rows     rows
+		}{
+			{"ed1.parquet", []int{1}, rows{{1, ""}, {2, ""}}}, // field 1 = id
+		},
+	)
+
+	handler := NewHandler(nil)
+	config := Config{TargetFileSizeBytes: 256 * 1024 * 1024, MinInputFiles: 2, MaxCommitRetries: 3, ApplyDeletes: true}
+	result, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil)
+	if err != nil {
+		t.Fatalf("compactDataFiles: %v", err)
+	}
+	if !strings.Contains(result, "compacted") {
+		t.Fatalf("expected compaction result, got %q", result)
+	}
+
+	var merged []byte
+	for _, e := range fs.listDir(path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data")) {
+		if strings.HasPrefix(e.Name, "compact-") {
+			merged = e.Content
+		}
+	}
+	if merged == nil {
+		t.Fatal("no merged file found")
+	}
+	got := readIDNameRows(t, merged)
+	want := []idNameRow{{1, "alice v2"}, {2, "bob v3"}, {3, "carol"}}
+	if len(got) != len(want) {
+		t.Fatalf("merged rows = %v, want %v", got, want)
+	}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Fatalf("merged rows = %v, want %v", got, want)
+		}
+	}
+}
+
+func putEqDeleteFile(t *testing.T, fs *fakeFilerServer, name string, rows []idNameRow) int64 {
+	t.Helper()
+	var buf bytes.Buffer
+	w := parquet.NewWriter(&buf, parquet.SchemaOf(new(idNameRow)))
+	for i := range rows {
+		if err := w.Write(&rows[i]); err != nil {
+			t.Fatalf("write eq delete: %v", err)
+		}
+	}
+	if err := w.Close(); err != nil {
+		t.Fatalf("close eq delete: %v", err)
+	}
+	fs.putEntry("/buckets/test-bucket/ns/tbl/data", name, &filer_pb.Entry{Name: name, Content: buf.Bytes()})
+	return int64(buf.Len())
+}
+
+func putIDNameDataFile(t *testing.T, fs *fakeFilerServer, name string, rows []idNameRow) int64 {
+	t.Helper()
+	in := make([]struct {
+		ID   int64
+		Name string
+	}, len(rows))
+	for i, r := range rows {
+		in[i].ID, in[i].Name = r.ID, r.Name
+	}
+	return int64(len(writeTestParquetFile(t, fs, "/buckets/test-bucket/ns/tbl/data", name, in)))
+}
+
+func entryWithSeq(t *testing.T, spec iceberg.PartitionSpec, content iceberg.ManifestEntryContent, filePath string,
+	partition map[int]any, records, size int64, seq *int64, eqFieldIDs []int) iceberg.ManifestEntry {
+	t.Helper()
+	dfb, err := iceberg.NewDataFileBuilder(spec, content, filePath, iceberg.ParquetFile, partition, nil, nil, records, size)
+	if err != nil {
+		t.Fatalf("build %s: %v", filePath, err)
+	}
+	if eqFieldIDs != nil {
+		dfb.EqualityFieldIDs(eqFieldIDs)
+	}
+	snapID := int64(1)
+	return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, seq, seq, dfb.Build())
+}
+
+// An equality delete file applies only within its own partition, unless its
+// partition spec is unpartitioned (Iceberg spec, "Scan Planning").
+func TestEqualityDeletesApplyOnlyWithinTheirPartition(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	schema := newTestSchema()
+	spec := iceberg.NewPartitionSpecID(1, iceberg.PartitionField{SourceIDs: []int{2}, FieldID: 1000, Name: "name", Transform: iceberg.IdentityTransform{}})
+	specs := map[int]iceberg.PartitionSpec{0: *iceberg.UnpartitionedSpec, 1: spec}
+	one, two := int64(1), int64(2)
+
+	sizeA := putIDNameDataFile(t, fs, "pa.parquet", []idNameRow{{1, "a"}, {2, "a"}})
+	sizeB := putIDNameDataFile(t, fs, "pb.parquet", []idNameRow{{1, "b"}})
+	dataA := entryWithSeq(t, spec, iceberg.EntryContentData, "data/pa.parquet", map[int]any{1000: "a"}, 2, sizeA, &one, nil)
+	dataB := entryWithSeq(t, spec, iceberg.EntryContentData, "data/pb.parquet", map[int]any{1000: "b"}, 1, sizeB, &one, nil)
+
+	// a delete of id 1 in partition a
+	sizeDel := putEqDeleteFile(t, fs, "eq-a.parquet", []idNameRow{{1, ""}})
+	delA := entryWithSeq(t, spec, iceberg.EntryContentEqDeletes, "data/eq-a.parquet", map[int]any{1000: "a"}, 1, sizeDel, &two, []int{1})
+	groups, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{delA}, schema, specs)
+	if err != nil {
+		t.Fatalf("collectEqualityDeletes: %v", err)
+	}
+	// as the compaction does: the groups narrowed to the bin's partition, then merged
+	merge := func(entry iceberg.ManifestEntry, groups []equalityDeleteGroup) []idNameRow {
+		bin := compactionBin{SpecID: entry.DataFile().SpecID(), Partition: entry.DataFile().Partition(), Entries: []iceberg.ManifestEntry{entry}}
+		groups, err := eqDeleteGroupsForBin(groups, bin, specs, schema)
+		if err != nil {
+			t.Fatalf("eqDeleteGroupsForBin %s: %v", entry.DataFile().FilePath(), err)
+		}
+		out, _, err := mergeParquetFiles(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{entry}, nil, groups, schema, 0, false)
+		if err != nil {
+			t.Fatalf("mergeParquetFiles %s: %v", entry.DataFile().FilePath(), err)
+		}
+		return readIDNameRows(t, out)
+	}
+	if got := merge(dataA, groups); len(got) != 1 || got[0] != (idNameRow{2, "a"}) {
+		t.Errorf("partition a = %v, want only {2 a}", got)
+	}
+	if got := merge(dataB, groups); len(got) != 1 || got[0] != (idNameRow{1, "b"}) {
+		t.Errorf("partition b = %v, want {1 b} kept: the delete is partition a's", got)
+	}
+
+	// the same delete from an unpartitioned delete file applies to every partition
+	sizeGlobal := putEqDeleteFile(t, fs, "eq-all.parquet", []idNameRow{{1, ""}})
+	delAll := entryWithSeq(t, *iceberg.UnpartitionedSpec, iceberg.EntryContentEqDeletes, "data/eq-all.parquet", map[int]any{}, 1, sizeGlobal, &two, []int{1})
+	groups, err = collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{delAll}, schema, specs)
+	if err != nil {
+		t.Fatalf("collectEqualityDeletes: %v", err)
+	}
+	if got := merge(dataB, groups); len(got) != 0 {
+		t.Errorf("partition b = %v, want nothing: an unpartitioned delete applies everywhere", got)
+	}
+}
+
+// Without data sequence numbers the order of a delete and a row cannot be
+// known, so the compaction refuses rather than guesses.
+func TestEqualityDeletesWithoutSequenceNumbersAreRefused(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	schema := newTestSchema()
+	spec := *iceberg.UnpartitionedSpec
+	one, two := int64(1), int64(2)
+
+	sizeDel := putEqDeleteFile(t, fs, "eq.parquet", []idNameRow{{1, ""}})
+	unsequencedDelete := entryWithSeq(t, spec, iceberg.EntryContentEqDeletes, "data/eq.parquet", map[int]any{}, 1, sizeDel, nil, []int{1})
+	if _, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{unsequencedDelete}, schema, testUnpartitionedSpecs); err == nil ||
+		!strings.Contains(err.Error(), "sequence number") {
+		t.Errorf("collectEqualityDeletes without a sequence number: err = %v, want a sequence number error", err)
+	}
+
+	sequencedDelete := entryWithSeq(t, spec, iceberg.EntryContentEqDeletes, "data/eq.parquet", map[int]any{}, 1, sizeDel, &two, []int{1})
+	groups, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{sequencedDelete}, schema, testUnpartitionedSpecs)
+	if err != nil {
+		t.Fatalf("collectEqualityDeletes: %v", err)
+	}
+	size := putIDNameDataFile(t, fs, "d.parquet", []idNameRow{{1, "x"}, {2, "y"}})
+	unsequencedData := entryWithSeq(t, spec, iceberg.EntryContentData, "data/d.parquet", map[int]any{}, 2, size, nil, nil)
+	if _, _, err := mergeParquetFiles(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{unsequencedData}, nil, groups, schema, 0, false); err == nil ||
+		!strings.Contains(err.Error(), "sequence number") {
+		t.Errorf("merge of a data file without a sequence number: err = %v, want a sequence number error", err)
+	}
+	// control: the same file with its sequence number merges, the delete applied
+	sequencedData := entryWithSeq(t, spec, iceberg.EntryContentData, "data/d.parquet", map[int]any{}, 2, size, &one, nil)
+	out, _, err := mergeParquetFiles(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{sequencedData}, nil, groups, schema, 0, false)
+	if err != nil {
+		t.Fatalf("mergeParquetFiles: %v", err)
+	}
+	if got := readIDNameRows(t, out); len(got) != 1 || got[0] != (idNameRow{2, "y"}) {
+		t.Errorf("merged = %v, want only {2 y}", got)
+	}
+}
+
+// A spec whose every field is void (a v1 table's dropped partition field) is
+// unpartitioned: a delete under it applies to every data file, whatever the
+// data file's spec. Its partition map is not empty
+// ({1001: nil}), so the map alone cannot tell.
+func TestEqualityDeleteUnderAnAllVoidSpecAppliesEverywhere(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	schema := newTestSchema()
+	byName := iceberg.NewPartitionSpecID(1, iceberg.PartitionField{SourceIDs: []int{2}, FieldID: 1000, Name: "name", Transform: iceberg.IdentityTransform{}})
+	allVoid := iceberg.NewPartitionSpecID(2, iceberg.PartitionField{SourceIDs: []int{2}, FieldID: 1001, Name: "name_dropped", Transform: iceberg.VoidTransform{}})
+	specs := map[int]iceberg.PartitionSpec{1: byName, 2: allVoid}
+	one, two := int64(1), int64(2)
+
+	size := putIDNameDataFile(t, fs, "pb.parquet", []idNameRow{{1, "b"}, {2, "b"}})
+	data := entryWithSeq(t, byName, iceberg.EntryContentData, "data/pb.parquet", map[int]any{1000: "b"}, 2, size, &one, nil)
+	sizeDel := putEqDeleteFile(t, fs, "eq-void.parquet", []idNameRow{{1, ""}})
+	del := entryWithSeq(t, allVoid, iceberg.EntryContentEqDeletes, "data/eq-void.parquet", map[int]any{1001: nil}, 1, sizeDel, &two, []int{1})
+
+	groups, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{del}, schema, specs)
+	if err != nil {
+		t.Fatalf("collectEqualityDeletes: %v", err)
+	}
+	out, _, err := mergeParquetFiles(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{data}, nil, groups, schema, 0, false)
+	if err != nil {
+		t.Fatalf("mergeParquetFiles: %v", err)
+	}
+	if got := readIDNameRows(t, out); len(got) != 1 || got[0] != (idNameRow{2, "b"}) {
+		t.Errorf("merged = %v, want only {2 b}: an all-void spec's delete applies to every partition", got)
+	}
+}
+
+// A where filter narrows the bins, not the deletes that may apply to them: an
+// unpartitioned delete file (no field the clause could match) still applies
+// to the bins in the filter's partitions.
+func TestAWhereFilterKeepsUnpartitionedDeletes(t *testing.T) {
+	byName := iceberg.NewPartitionSpecID(1, iceberg.PartitionField{SourceIDs: []int{2}, FieldID: 1000, Name: "name", Transform: iceberg.IdentityTransform{}})
+	specs := map[int]iceberg.PartitionSpec{0: *iceberg.UnpartitionedSpec, 1: byName}
+	predicate := &partitionPredicate{Clauses: []whereClause{{Field: "name", Literals: []string{"'b'"}}}}
+	two := int64(2)
+	cases := []struct {
+		name  string
+		entry iceberg.ManifestEntry
+		want  bool
+	}{
+		{"unpartitioned delete", entryWithSeq(t, *iceberg.UnpartitionedSpec, iceberg.EntryContentEqDeletes, "data/u.parquet", map[int]any{}, 1, 1, &two, []int{1}), true},
+		{"delete in the filter's partition", entryWithSeq(t, byName, iceberg.EntryContentEqDeletes, "data/b.parquet", map[int]any{1000: "b"}, 1, 1, &two, []int{1}), true},
+		{"delete in another partition", entryWithSeq(t, byName, iceberg.EntryContentEqDeletes, "data/a.parquet", map[int]any{1000: "a"}, 1, 1, &two, []int{1}), false},
+	}
+	for _, c := range cases {
+		got, err := deleteInPredicateScope(c.entry, predicate, specs, newTestSchema())
+		if err != nil {
+			t.Fatalf("%s: %v", c.name, err)
+		}
+		if got != c.want {
+			t.Errorf("%s: in scope = %v, want %v", c.name, got, c.want)
+		}
+	}
+}
+
+// Partition values are compared as the spec types them, not as their Go shape:
+// two writers' manifests decode one uuid as uuid.UUID and as [16]byte (or a
+// long as int32 and int64), and a delete compared as JSON text misses its own
+// partition, so the merged file keeps the rows it deleted. A value that does
+// not convert fails closed: the bin is refused.
+func TestEqualityDeletesMatchTheirPartitionByTypedValue(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	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: "name"},
+		iceberg.NestedField{ID: 3, Type: iceberg.PrimitiveTypes.UUID, Name: "tenant"},
+		iceberg.NestedField{ID: 4, Type: iceberg.PrimitiveTypes.Int64, Name: "region"},
+	)
+	spec := iceberg.NewPartitionSpecID(1,
+		iceberg.PartitionField{SourceIDs: []int{3}, FieldID: 1000, Name: "tenant", Transform: iceberg.IdentityTransform{}},
+		iceberg.PartitionField{SourceIDs: []int{4}, FieldID: 1001, Name: "region", Transform: iceberg.IdentityTransform{}})
+	specs := map[int]iceberg.PartitionSpec{0: *iceberg.UnpartitionedSpec, 1: spec}
+	tenant := uuid.MustParse("8f14e45f-ceea-467f-a0e6-6f4d9c2b1a01")
+	one, two := int64(1), int64(2)
+
+	// the data file as one writer's manifest decodes it, the delete as another's
+	dataPartition := map[int]any{1000: tenant, 1001: int64(7)}
+	deletePartition := map[int]any{1000: [16]byte(tenant), 1001: int32(7)}
+	if partitionKey(dataPartition) == partitionKey(deletePartition) {
+		t.Fatal("precondition: the two shapes must differ as JSON text")
+	}
+	bin := compactionBin{SpecID: 1, Partition: dataPartition}
+	size := putIDNameDataFile(t, fs, "d.parquet", []idNameRow{{1, "x"}, {2, "y"}})
+	data := entryWithSeq(t, spec, iceberg.EntryContentData, "data/d.parquet", dataPartition, 2, size, &one, nil)
+	bin.Entries = []iceberg.ManifestEntry{data}
+	sizeDel := putEqDeleteFile(t, fs, "eq.parquet", []idNameRow{{1, ""}})
+	del := entryWithSeq(t, spec, iceberg.EntryContentEqDeletes, "data/eq.parquet", deletePartition, 1, sizeDel, &two, []int{1})
+
+	groups, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{del}, schema, specs)
+	if err != nil {
+		t.Fatalf("collectEqualityDeletes: %v", err)
+	}
+	binGroups, err := eqDeleteGroupsForBin(groups, bin, specs, schema)
+	if err != nil {
+		t.Fatalf("eqDeleteGroupsForBin: %v", err)
+	}
+	out, _, err := mergeParquetFiles(context.Background(), client, "test-bucket", "ns/tbl", bin.Entries, nil, binGroups, schema, 0, false)
+	if err != nil {
+		t.Fatalf("mergeParquetFiles: %v", err)
+	}
+	if got := readIDNameRows(t, out); len(got) != 1 || got[0] != (idNameRow{2, "y"}) {
+		t.Errorf("merged = %v, want only {2 y}: the delete is in the data file's partition", got)
+	}
+
+	// control: a delete of another tenant does not apply
+	other := map[int]any{1000: uuid.MustParse("00000000-0000-4000-8000-000000000001"), 1001: int64(7)}
+	delOther := entryWithSeq(t, spec, iceberg.EntryContentEqDeletes, "data/eq.parquet", other, 1, sizeDel, &two, []int{1})
+	groups, err = collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{delOther}, schema, specs)
+	if err != nil {
+		t.Fatalf("collectEqualityDeletes: %v", err)
+	}
+	if binGroups, err = eqDeleteGroupsForBin(groups, bin, specs, schema); err != nil || len(binGroups) != 0 {
+		t.Errorf("another tenant's delete: %d groups, err %v; want none", len(binGroups), err)
+	}
+
+	// a bin whose partition value does not convert is refused while a delete of its spec exists
+	undecidable := compactionBin{SpecID: 1, Partition: map[int]any{1000: "not-a-uuid", 1001: int64(7)}, Entries: bin.Entries}
+	if _, err := eqDeleteGroupsForBin(groups, undecidable, specs, schema); err == nil || !strings.Contains(err.Error(), "cannot be compared") {
+		t.Errorf("an undecidable bin: err = %v, want the bin refused", err)
+	}
+	// and a delete whose value does not convert is refused when it is read
+	delBad := entryWithSeq(t, spec, iceberg.EntryContentEqDeletes, "data/eq.parquet", map[int]any{1000: "not-a-uuid", 1001: int64(7)}, 1, sizeDel, &two, []int{1})
+	if _, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{delBad}, schema, specs); err == nil {
+		t.Error("a delete whose partition does not convert: want an error")
+	}
+	// a null written as nil and a null left out are one partition
+	if a, err := typedPartitionKey(spec, schema, map[int]any{1000: nil, 1001: int64(7)}); err != nil {
+		t.Fatal(err)
+	} else if b, err := typedPartitionKey(spec, schema, map[int]any{1001: int32(7)}); err != nil || a != b {
+		t.Errorf("null present %q vs missing %q (err %v): want one key", a, b, err)
+	}
+}
+
+// A delete file under a partitioned spec with no partition values cannot be
+// placed: read as unpartitioned by its map it would match no bin, and its
+// deletes would be skipped. It is refused, and the compaction with it.
+func TestADeleteWithoutPartitionValuesUnderAPartitionedSpecIsRefused(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	schema := newTestSchema()
+	byName := iceberg.NewPartitionSpecID(1, iceberg.PartitionField{SourceIDs: []int{2}, FieldID: 1000, Name: "name", Transform: iceberg.IdentityTransform{}})
+	specs := map[int]iceberg.PartitionSpec{0: *iceberg.UnpartitionedSpec, 1: byName}
+	two := int64(2)
+	sizeDel := putEqDeleteFile(t, fs, "eq-empty.parquet", []idNameRow{{1, ""}})
+	del := entryWithSeq(t, byName, iceberg.EntryContentEqDeletes, "data/eq-empty.parquet", map[int]any{}, 1, sizeDel, &two, []int{1})
+
+	if _, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{del}, schema, specs); err == nil ||
+		!strings.Contains(err.Error(), "no partition values") {
+		t.Fatalf("collectEqualityDeletes: err = %v, want the delete refused for its missing partition values", err)
+	}
+	// where filters keep it in, so it is refused rather than filtered away
+	predicate := &partitionPredicate{Clauses: []whereClause{{Field: "name", Literals: []string{"'a'"}}}}
+	if in, err := deleteInPredicateScope(del, predicate, specs, schema); err != nil || !in {
+		t.Errorf("in predicate scope = %v (err %v), want true", in, err)
+	}
+}
+
+// Delete groups are keyed by fields and scope with separators: spec 1 with
+// partition field 11 and spec 11 with field 1 printed the same key without
+// them, so one group would take both files' keys under the first one's scope
+// and the other spec's bin would get no deletes.
+func TestEqualityDeleteGroupsOfDifferentScopesDoNotCollide(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	schema := newTestSchema()
+	spec1 := iceberg.NewPartitionSpecID(1, iceberg.PartitionField{SourceIDs: []int{2}, FieldID: 11, Name: "n11", Transform: iceberg.IdentityTransform{}})
+	spec11 := iceberg.NewPartitionSpecID(11, iceberg.PartitionField{SourceIDs: []int{2}, FieldID: 1, Name: "n1", Transform: iceberg.IdentityTransform{}})
+	specs := map[int]iceberg.PartitionSpec{0: *iceberg.UnpartitionedSpec, 1: spec1, 11: spec11}
+	one, two := int64(1), int64(2)
+
+	size11 := putEqDeleteFile(t, fs, "eq-11.parquet", []idNameRow{{1, ""}})
+	del11 := entryWithSeq(t, spec11, iceberg.EntryContentEqDeletes, "data/eq-11.parquet", map[int]any{1: "x"}, 1, size11, &two, []int{1})
+	size1 := putEqDeleteFile(t, fs, "eq-1.parquet", []idNameRow{{2, ""}})
+	del1 := entryWithSeq(t, spec1, iceberg.EntryContentEqDeletes, "data/eq-1.parquet", map[int]any{11: "x"}, 1, size1, &two, []int{1})
+	groups, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{del11, del1}, schema, specs)
+	if err != nil {
+		t.Fatalf("collectEqualityDeletes: %v", err)
+	}
+	if len(groups) != 2 {
+		t.Fatalf("%d groups, want 2: the two scopes are different", len(groups))
+	}
+
+	size := putIDNameDataFile(t, fs, "d1.parquet", []idNameRow{{1, "x"}, {2, "x"}})
+	data := entryWithSeq(t, spec1, iceberg.EntryContentData, "data/d1.parquet", map[int]any{11: "x"}, 2, size, &one, nil)
+	bin := compactionBin{SpecID: 1, Partition: map[int]any{11: "x"}, Entries: []iceberg.ManifestEntry{data}}
+	binGroups, err := eqDeleteGroupsForBin(groups, bin, specs, schema)
+	if err != nil {
+		t.Fatalf("eqDeleteGroupsForBin: %v", err)
+	}
+	out, _, err := mergeParquetFiles(context.Background(), client, "test-bucket", "ns/tbl", bin.Entries, nil, binGroups, schema, 0, false)
+	if err != nil {
+		t.Fatalf("mergeParquetFiles: %v", err)
+	}
+	if got := readIDNameRows(t, out); len(got) != 1 || got[0] != (idNameRow{1, "x"}) {
+		t.Errorf("spec 1's bin = %v, want only {1 x}: spec 1's delete of id 2 applies, spec 11's of id 1 does not", got)
+	}
+}

weed/worker/tasks/iceberg/exec_test.go

@@ -306,6 +306,11 @@ type tableSetup struct {
 	// table does not fit.
 	Schema *iceberg.Schema
 	Spec   *iceberg.PartitionSpec
+	// DataSequenceNumbers gives populateTableWithDeleteFiles' data files, by
+	// name, a data sequence number other than 1: a file written in the same
+	// commit as the delete files (sequence number 2) is one they must not
+	// delete from.
+	DataSequenceNumbers map[string]int64
 }
 
 func (ts tableSetup) tablePath() string {
@@ -2701,9 +2706,18 @@ func populateTableWithDeleteFilesAndSortOrder(
 			t.Fatalf("build data file %s: %v", df.Name, err)
 		}
 		snapID := int64(1)
-		dataManifestEntries = append(dataManifestEntries, iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build()))
+		dataSeq := int64(1)
+		if seq, ok := setup.DataSequenceNumbers[df.Name]; ok {
+			dataSeq = seq
+		}
+		dataManifestEntries = append(dataManifestEntries, iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, &dataSeq, &dataSeq, dfb.Build()))
 	}
 
+	// The delete files are committed after the data, as a writer commits
+	// them: a delete applies only to data files with a lower data sequence
+	// number (equality) or one no higher (position).
+	deleteSeq := int64(2)
+
 	// Write data manifest
 	var dataManifestBuf bytes.Buffer
 	dataManifestName := "data-manifest-1.avro"
@@ -2745,7 +2759,7 @@ func populateTableWithDeleteFilesAndSortOrder(
 				t.Fatalf("build pos delete file: %v", err)
 			}
 			snapID := int64(1)
-			posDeleteEntries = append(posDeleteEntries, iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build()))
+			posDeleteEntries = append(posDeleteEntries, iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, &deleteSeq, &deleteSeq, dfb.Build()))
 		}
 
 		posManifestName := "pos-delete-manifest-1.avro"
@@ -2786,7 +2800,7 @@ func populateTableWithDeleteFilesAndSortOrder(
 			}
 			dfb.EqualityFieldIDs(edf.FieldIDs)
 			snapID := int64(1)
-			eqDeleteEntries = append(eqDeleteEntries, iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build()))
+			eqDeleteEntries = append(eqDeleteEntries, iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, &deleteSeq, &deleteSeq, dfb.Build()))
 		}
 
 		eqManifestName := "eq-delete-manifest-1.avro"
@@ -2801,7 +2815,12 @@ func populateTableWithDeleteFilesAndSortOrder(
 
 	// Write manifest list
 	var mlBuf bytes.Buffer
-	seqNum := int64(1)
+	seqNum := deleteSeq // the table's head: the commit that wrote the deletes, or a later data file's
+	for _, seq := range setup.DataSequenceNumbers {
+		if seq > seqNum {
+			seqNum = seq
+		}
+	}
 	if err := iceberg.WriteManifestList(version, &mlBuf, 1, nil, &seqNum, 0, allManifests); err != nil {
 		t.Fatalf("write manifest list: %v", err)
 	}
@@ -2812,7 +2831,7 @@ func populateTableWithDeleteFilesAndSortOrder(
 
 	// Build final metadata with snapshot
 	now := time.Now().UnixMilli()
-	snap := table.Snapshot{SnapshotID: 1, TimestampMs: now, ManifestList: setup.fileRef("metadata", "snap-1.avro")}
+	snap := table.Snapshot{SnapshotID: 1, SequenceNumber: seqNum, TimestampMs: now, ManifestList: setup.fileRef("metadata", "snap-1.avro")}
 	if setup.SnapshotSummary != nil {
 		snap.Summary = &table.Summary{Operation: table.OpAppend, Properties: setup.SnapshotSummary}
 	}

weed/worker/tasks/iceberg/handler_test.go

@@ -1229,9 +1229,9 @@ func TestCollectEqualityDeletes(t *testing.T) {
 	}
 	dfBuilder.EqualityFieldIDs([]int{1, 2}) // id + name
 	snapID := int64(1)
-	entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfBuilder.Build())
+	entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, &testDeleteSeq, &testDeleteSeq, dfBuilder.Build())
 
-	groups, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{entry}, schema)
+	groups, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{entry}, schema, testUnpartitionedSpecs)
 	if err != nil {
 		t.Fatalf("collectEqualityDeletes: %v", err)
 	}
@@ -1377,7 +1377,7 @@ func TestMergeParquetFilesWithEqualityDeletes(t *testing.T) {
 	}
 	snapID := int64(1)
 	entries := []iceberg.ManifestEntry{
-		iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build()),
+		iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, &testDataSeq, &testDataSeq, dfb.Build()),
 	}
 
 	// Build equality deletes via the collect pipeline so keys match the
@@ -1409,10 +1409,10 @@ func TestMergeParquetFilesWithEqualityDeletes(t *testing.T) {
 		t.Fatalf("build eq delete: %v", err)
 	}
 	eqDfb.EqualityFieldIDs([]int{2}) // field 2 = name
-	eqEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, eqDfb.Build())
+	eqEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, &testDeleteSeq, &testDeleteSeq, eqDfb.Build())
 
 	eqGroups, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl",
-		[]iceberg.ManifestEntry{eqEntry}, schema)
+		[]iceberg.ManifestEntry{eqEntry}, schema, testUnpartitionedSpecs)
 	if err != nil {
 		t.Fatalf("collectEqualityDeletes: %v", err)
 	}

weed/worker/tasks/iceberg/row_lineage.go

@@ -360,7 +360,12 @@ func copyFileRows(ctx context.Context, reader *parquet.Reader, entry iceberg.Man
 		reader.Close()
 		return 0, err
 	}
-	return visitFilteredParquetRows(ctx, reader, entry.DataFile().FilePath(), bucketName, dataPath, positionDeletes, eqGroups, func(filtered []parquet.Row, positions []int64) error {
+	eqGroups, dataSeq, err := equalityDeletesFor(eqGroups, entry)
+	if err != nil {
+		reader.Close()
+		return 0, err
+	}
+	return visitFilteredParquetRows(ctx, reader, entry.DataFile().FilePath(), bucketName, dataPath, positionDeletes, eqGroups, dataSeq, func(filtered []parquet.Row, positions []int64) error {
 		if mapper != nil {
 			filtered = mapper.mapRows(filtered, positions)
 		}