Bug(table): RowDelta.validate uses AlwaysTrue filter — false conflicts for concurrent appends to different partitions

[mzzz-zzm]

Apache Iceberg version

main (development)

Please describe the bug 🐞

RowDelta.validate() checks for conflicting concurrent data files by calling validateNoConflictingDataFiles(cc, iceberg.AlwaysTrue{}, level). The AlwaysTrue{} filter matches every data file in every partition. As a result, a concurrent FastAppend to a completely different partition causes a RowDelta equality-delete commit to be rejected with ErrConflictingDataFiles, even though the two operations cannot possibly interfere with each other.

In practice, any busy partitioned table with a streaming DELETE/UPDATE job (which uses RowDelta with equality-delete files) will almost never be able to commit under the default SERIALIZABLE isolation level, because some other writer is always appending to some partition.

Affected component

table/row_delta.go — RowDelta.validate()

Steps to reproduce

The test is self-contained. Save the file below as table/bug_repro_rowdelta_partition_test.go in an unmodified checkout of main and run:

go test ./table/ -run TestBugRepro_RowDeltaFalseConflictDifferentPartition -v

Expected output on unfixed upstream:

--- FAIL: TestBugRepro_RowDeltaFalseConflictDifferentPartition
    bug_repro_rowdelta_partition_test.go:232:
        Error:   Received unexpected error:
                 commit failed, refresh and try again: concurrent data files added:
                 snapshot ... added data file .../worker-b-cold.parquet matching filter AlwaysTrue()
        Messages: serializable isolation must allow an eq-delete when the only
                  concurrent data file is in a completely different partition (AlwaysTrue bug)

The test sets up a baseline commit first (so Writer A starts from a non-empty base), which isolates this bug from the separate empty-base
conflict-bypass bug. Worker B's data is in partition category="cold";
Writer A's equality-delete targets category="hot". The two partitions are completely disjoint.

Full source of table/bug_repro_rowdelta_partition_test.go:

package table_test

import (
	"context"
	"fmt"
	"path/filepath"
	"sync"
	"testing"

	"github.com/apache/iceberg-go"
	iceio "github.com/apache/iceberg-go/io"
	"github.com/apache/iceberg-go/table"
	"github.com/stretchr/testify/require"
)

type rowDeltaRepro3Catalog struct {
	mu       sync.Mutex
	current  table.Metadata
	location string
}

func newRowDeltaRepro3Catalog(meta table.Metadata, location string) *rowDeltaRepro3Catalog {
	return &rowDeltaRepro3Catalog{current: meta, location: location}
}

func (c *rowDeltaRepro3Catalog) LoadTable(_ context.Context, ident table.Identifier) (*table.Table, error) {
	c.mu.Lock()
	meta := c.current
	c.mu.Unlock()
	return table.New(
		ident, meta, c.location+"/metadata/v1.metadata.json",
		func(_ context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil },
		c,
	), nil
}

func (c *rowDeltaRepro3Catalog) CommitTable(
	_ context.Context,
	ident table.Identifier,
	reqs []table.Requirement,
	updates []table.Update,
) (table.Metadata, string, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	for _, req := range reqs {
		if err := req.Validate(c.current); err != nil {
			return nil, "", fmt.Errorf("%w: CAS check failed: %v", table.ErrCommitFailed, err)
		}
	}
	newMeta, err := table.UpdateTableMetadata(c.current, updates, "")
	if err != nil {
		return nil, "", err
	}
	c.current = newMeta
	return newMeta, c.location + "/metadata/committed.metadata.json", nil
}

func makeRepro3DataFile(t *testing.T, spec *iceberg.PartitionSpec, path, category string) iceberg.DataFile {
	t.Helper()
	var partition map[int]any
	if category != "" {
		partition = map[int]any{1000: category}
	}
	b, err := iceberg.NewDataFileBuilder(
		*spec, iceberg.EntryContentData, path,
		iceberg.ParquetFile, partition, nil, nil, 100, 1024,
	)
	require.NoError(t, err)
	return b.Build()
}

func makeRepro3EqDeleteFile(t *testing.T, spec *iceberg.PartitionSpec, path, category string) iceberg.DataFile {
	t.Helper()
	var partition map[int]any
	if category != "" {
		partition = map[int]any{1000: category}
	}
	b, err := iceberg.NewDataFileBuilder(
		*spec, iceberg.EntryContentEqDeletes, path,
		iceberg.ParquetFile, partition, nil, nil, 5, 512,
	)
	require.NoError(t, err)
	return b.EqualityFieldIDs([]int{3}).Build()
}

func TestBugRepro_RowDeltaFalseConflictDifferentPartition(t *testing.T) {
	location := filepath.ToSlash(t.TempDir())
	ctx := context.Background()

	schema := iceberg.NewSchema(0,
		iceberg.NestedField{ID: 1, Name: "category", Type: iceberg.PrimitiveTypes.String, Required: false},
		iceberg.NestedField{ID: 2, Name: "value", Type: iceberg.PrimitiveTypes.Int64, Required: false},
		iceberg.NestedField{ID: 3, Name: "event_id", Type: iceberg.PrimitiveTypes.Int64, Required: false},
	)
	spec := iceberg.NewPartitionSpec(
		iceberg.PartitionField{
			SourceIDs: []int{1},
			FieldID:   1000,
			Name:      "category",
			Transform: iceberg.IdentityTransform{},
		},
	)
	props := iceberg.Properties{
		table.PropertyFormatVersion:        "2",
		table.CommitNumRetriesKey:          "1",
		table.CommitMinRetryWaitMsKey:      "0",
		table.CommitMaxRetryWaitMsKey:      "0",
		table.CommitTotalRetryTimeoutMsKey: "60000",
	}

	metaEmpty, err := table.NewMetadata(schema, &spec, table.UnsortedSortOrder, location, props)
	require.NoError(t, err)

	cat := newRowDeltaRepro3Catalog(metaEmpty, location)
	fsF := func(_ context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil }

	newTbl := func(base table.Metadata) *table.Table {
		return table.New([]string{"default", "repro3"}, base,
			location+"/metadata/v1.metadata.json", fsF, cat)
	}

	// Step 1: Establish baseline snapshot S0 so Writer A starts from a
	// non-empty base (isolates AlwaysTrue bug from the empty-base bug).
	baseline := makeRepro3DataFile(t, &spec, location+"/data/baseline.parquet", "warm")
	txSetup := newTbl(metaEmpty).NewTransaction()
	require.NoError(t, txSetup.AddDataFiles(ctx, []iceberg.DataFile{baseline}, nil))
	_, err = txSetup.Commit(ctx)
	require.NoError(t, err, "baseline commit must succeed")

	cat.mu.Lock()
	metaS0 := cat.current
	cat.mu.Unlock()

	// Step 2: Worker B appends a data file in partition category="cold".
	workerBTbl, err := cat.LoadTable(ctx, []string{"default", "repro3"})
	require.NoError(t, err)
	dfCold := makeRepro3DataFile(t, &spec, location+"/data/worker-b-cold.parquet", "cold")
	txB := workerBTbl.NewTransaction()
	require.NoError(t, txB.AddDataFiles(ctx, []iceberg.DataFile{dfCold}, nil))
	_, err = txB.Commit(ctx)
	require.NoError(t, err, "Worker B must commit successfully")

	// Step 3: Writer A starts from S0, adds equality-delete in category="hot".
	writerATbl := newTbl(metaS0)
	eqDelHot := makeRepro3EqDeleteFile(t, &spec, location+"/data/writer-a-eq-del-hot.parquet", "hot")
	txA := writerATbl.NewTransaction()
	rd := txA.NewRowDelta(nil)
	rd.AddDeletes(eqDelHot)
	require.NoError(t, rd.Commit(ctx))

	// Step 4: Writer A commits.
	// Attempt 1: CAS fails (S0 vs Worker B's HEAD) → ErrCommitFailed.
	// Retry: newConflictContext returns [Worker_B_snapshot] as concurrent.
	//        RowDelta.validate calls validateNoConflictingDataFiles(AlwaysTrue{}).
	//        AlwaysTrue matches Worker B's "cold" file → ErrConflictingDataFiles.
	// Expected: no error ("cold" ≠ "hot", partitions are disjoint).
	_, err = txA.Commit(ctx)
	require.NoError(t, err,
		"serializable isolation must allow an eq-delete when the only concurrent "+
			"data file is in a completely different partition (AlwaysTrue bug)")
}

Root cause

The relevant section of table/row_delta.go on main:

// Comment in code:
// "does not yet surface the bound predicate. AlwaysTrue is the 
//  conservative fallback [...] Follow-up:"
if err := validateNoConflictingDataFiles(cc, iceberg.AlwaysTrue{}, level); err != nil {
    return err
}

validateNoConflictingDataFiles iterates every ADDED entry in every concurrent snapshot and accepts any data file whose partition satisfies the supplied filter.
Because the filter is iceberg.AlwaysTrue{}, every data file in every partition satisfies it — turning any concurrent append anywhere in the table into a reported conflict for the RowDelta committer.

The Java reference implementation (MergingSnapshotProducer.validateNoNewDataFiles) derives the conflict filter from the equality-delete predicate itself, so only rows that could match the delete are checked. The Go implementation had not yet threaded that predicate through and fell back to AlwaysTrue as a temporary conservative measure. The result is that SERIALIZABLE RowDelta commits are effectively impossible on any active partitioned table — the default isolation level is unusable.

Expected behaviour

The conflict filter should be scoped to the partition(s) covered by the equality-delete files in the RowDelta. A concurrent append to a completely different partition cannot be affected by the equality deletes and must not block the commit. Only concurrent data files whose partition intersects with the equality-delete partition(s) should be flagged as a conflict.

[mzzz-zzm]

Suggested fix

RowDelta.validate() is changed to collect the partition tuple of each equality-delete file and pass those tuples to a new helper validateNoConflictingDataFilesInPartitions. That helper only flags concurrent data files whose partition tuple matches one of the equality-delete partitions, rather than matching everything via AlwaysTrue{}. For unpartitioned tables (or equality-delete files with empty partition tuples) the helper falls back to the existing AlwaysTrue behaviour to remain safe.

This change touches two functions in two files and is independent of the manifest list rebuild fix and the empty-table conflict bypass fix.

Files changed

File	Role
table/row_delta.go	Collects eq-delete partition tuples; calls new helper instead of AlwaysTrue
table/conflict_validation.go	Adds validateNoConflictingDataFilesInPartitions and partitionTupleKey helpers

Diff

table/row_delta.go

-var hasEqDeletes bool
+var eqDeletePartitions []map[int]any
 for _, f := range rd.delFiles {
 	switch f.ContentType() {
 	case iceberg.EntryContentPosDeletes:
 		if ref := f.ReferencedDataFile(); ref != nil && *ref != "" {
 			referenced = append(referenced, *ref)
 		}
 	case iceberg.EntryContentEqDeletes:
-		hasEqDeletes = true
+		eqDeletePartitions = append(eqDeletePartitions, f.Partition())
 	}
 }

 for _, path := range referenced {
 	if _, found := cc.AddedDataFiles()[path]; found {
 		return ErrConflictedWithUncommittedDeleteFiles
 	}
 }

-if hasEqDeletes {
+if len(eqDeletePartitions) > 0 {
 	level := readIsolationLevel(rd.txn.meta.props,
 		WriteDeleteIsolationLevelKey, WriteDeleteIsolationLevelDefault)
-	// Conservative: eq-deletes apply by predicate, and RowDelta
-	// does not yet surface the bound predicate. AlwaysTrue is the
-	// safest over-approximation and matches PR 2.3's contract on
-	// validateNoConflictingDataFiles under SERIALIZABLE. Follow-up:
-	// narrow with the actual eq-delete filter once it is carried
-	// on the RowDelta.
-	if err := validateNoConflictingDataFiles(cc, iceberg.AlwaysTrue{}, level); err != nil {
+	// Use partition-aware conflict detection: only flag concurrent data
+	// files whose partition tuple matches one of the equality delete
+	// files' partition tuples.
+	if err := validateNoConflictingDataFilesInPartitions(cc, eqDeletePartitions, level); err != nil {
 		return err
 	}
 }

table/conflict_validation.go

Two new functions added after validateNoConflictingDataFiles:

// validateNoConflictingDataFilesInPartitions is like
// validateNoConflictingDataFiles but scoped to specific partition
// tuples derived from equality-delete files. It only flags concurrent
// data files whose partition tuple matches one of the provided tuples,
// avoiding false conflicts when a concurrent append lands in a
// completely different partition.
//
// When any provided partition tuple is empty (the table is
// unpartitioned or the delete file covers all partitions), the check
// falls back to AlwaysTrue — the equality delete could affect any row.
//
// Under IsolationSnapshot this validator is a no-op.
func validateNoConflictingDataFilesInPartitions(ctx *conflictContext, eqDeletePartitions []map[int]any, level IsolationLevel) error {
	if level != IsolationSerializable {
		return nil
	}
	if len(ctx.concurrent) == 0 || len(eqDeletePartitions) == 0 {
		return nil
	}

	// If any eq-delete file is unpartitioned (empty tuple), the delete
	// could affect any row — fall back to the conservative AlwaysTrue check.
	for _, p := range eqDeletePartitions {
		if len(p) == 0 {
			return validateAddedDataFilesMatchingFilter(ctx, iceberg.AlwaysTrue{})
		}
	}

	// Build a set of partition tuple keys for O(1) lookup.
	partSet := make(map[string]struct{}, len(eqDeletePartitions))
	for _, p := range eqDeletePartitions {
		partSet[partitionTupleKey(p)] = struct{}{}
	}

	for _, snap := range ctx.concurrent {
		manifests, err := snap.Manifests(ctx.fs)
		if err != nil {
			return fmt.Errorf("loading manifests for concurrent snapshot %d: %w", snap.SnapshotID, err)
		}
		for _, mf := range manifests {
			if mf.ManifestContent() != iceberg.ManifestContentData {
				continue
			}
			entries, err := mf.FetchEntries(ctx.fs, false)
			if err != nil {
				return fmt.Errorf("reading entries from manifest %s: %w", mf.FilePath(), err)
			}
			for _, e := range entries {
				if e.Status() != iceberg.EntryStatusADDED || e.SnapshotID() != snap.SnapshotID {
					continue
				}
				if _, ok := partSet[partitionTupleKey(e.DataFile().Partition())]; ok {
					return fmt.Errorf("%w: snapshot %d added data file %s in eq-delete partition",
						ErrConflictingDataFiles, snap.SnapshotID, e.DataFile().FilePath())
				}
			}
		}
	}
	return nil
}

// partitionTupleKey returns a deterministic string key for a partition
// tuple map. Keys are sorted by field id so maps with the same content
// always produce the same string.
func partitionTupleKey(p map[int]any) string {
	if len(p) == 0 {
		return ""
	}
	keys := make([]int, 0, len(p))
	for k := range p {
		keys = append(keys, k)
	}
	sort.Ints(keys)
	var buf []byte
	for _, k := range keys {
		buf = fmt.Appendf(buf, "%d=%v;", k, p[k])
	}
	return string(buf)
}

Behaviour matrix

eq-delete partition	concurrent data partition	isolation	result
hot	hot	SERIALIZABLE	ErrConflictingDataFiles (same partition — real conflict)
hot	cold	SERIALIZABLE	no error (different partition — no conflict)
hot	hot	SNAPSHOT	no error (snapshot isolation skips the check)
unpartitioned	any	SERIALIZABLE	ErrConflictingDataFiles (AlwaysTrue fallback)

Test

Run the reproducer from the bug report:

go test ./table/ -run TestBugRepro_RowDeltaFalseConflictDifferentPartition -v

Expected after this fix:

--- PASS: TestBugRepro_RowDeltaFalseConflictDifferentPartition (0.11s)

[laskoviymishka]

@mzzz-zzm would you like to file a PR for that?

[tanmayrauth]

@laskoviymishka @mzzz-zzm Can I work on this PR?

[laskoviymishka]

let's wait a bit from @mzzz-zzm , i think he almost have a full solution in mind already

[mzzz-zzm]

Thanks for the opportunity. Sure. I can work on this