diff --git a/weed/worker/tasks/iceberg/compact.go b/weed/worker/tasks/iceberg/compact.go
index 7936df7d26..bef7990d53 100644
--- a/weed/worker/tasks/iceberg/compact.go
+++ b/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
 					}
diff --git a/weed/worker/tasks/iceberg/compact_schema_test.go b/weed/worker/tasks/iceberg/compact_schema_test.go
index 3d119037da..00cce99a03 100644
--- a/weed/worker/tasks/iceberg/compact_schema_test.go
+++ b/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")
 	}
diff --git a/weed/worker/tasks/iceberg/eqdelete_scope.go b/weed/worker/tasks/iceberg/eqdelete_scope.go
new file mode 100644
index 0000000000..8266fe8b42
--- /dev/null
+++ b/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
+}
diff --git a/weed/worker/tasks/iceberg/eqdelete_sequence_test.go b/weed/worker/tasks/iceberg/eqdelete_sequence_test.go
new file mode 100644
index 0000000000..e1b2c001ee
--- /dev/null
+++ b/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)
+	}
+}
diff --git a/weed/worker/tasks/iceberg/exec_test.go b/weed/worker/tasks/iceberg/exec_test.go
index 172e896a1e..cacbe1ef49 100644
--- a/weed/worker/tasks/iceberg/exec_test.go
+++ b/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}
 	}
diff --git a/weed/worker/tasks/iceberg/handler_test.go b/weed/worker/tasks/iceberg/handler_test.go
index 3d88044ab2..ef04681e68 100644
--- a/weed/worker/tasks/iceberg/handler_test.go
+++ b/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)
 	}
diff --git a/weed/worker/tasks/iceberg/row_lineage.go b/weed/worker/tasks/iceberg/row_lineage.go
index f6aa85c2aa..cc24b49237 100644
--- a/weed/worker/tasks/iceberg/row_lineage.go
+++ b/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)
 		}