fix(table): reject ambiguous equality-delete columns

table/equality_delete_reader.go

@@ -217,11 +217,15 @@ func readEqualityDeleteFile(ctx context.Context, fs iceio.IO, tableSchema *icebe
 
 		colNames[i] = name
 		indices := tbl.Schema().FieldIndices(name)
-		if len(indices) == 0 {
+		switch len(indices) {
+		case 0:
 			return nil, nil, fmt.Errorf("equality delete column %q not found in delete file %s", name, dataFile.FilePath())
+		case 1:
+			colIndices[i] = indices[0]
+		default:
+			return nil, nil, fmt.Errorf("equality delete column %q is ambiguous in delete file %s: found %d columns",
+				name, dataFile.FilePath(), len(indices))
 		}
-
-		colIndices[i] = indices[0]
 	}
 
 	// Build the set of encoded delete keys by iterating aligned batches.

table/equality_delete_reader_test.go

@@ -22,7 +22,11 @@ import (
 	"path/filepath"
 	"testing"
 
+	"github.com/apache/arrow-go/v18/arrow"
 	"github.com/apache/arrow-go/v18/arrow/array"
+	"github.com/apache/arrow-go/v18/arrow/memory"
+	"github.com/apache/arrow-go/v18/parquet"
+	"github.com/apache/arrow-go/v18/parquet/pqarrow"
 	"github.com/apache/iceberg-go"
 	iceio "github.com/apache/iceberg-go/io"
 	"github.com/apache/iceberg-go/table"
@@ -119,6 +123,55 @@ func TestEqualityDeleteReadRoundTrip(t *testing.T) {
 	assert.Equal(t, []int64{1, 3, 5}, ids, "expected rows with id=2 and id=4 deleted")
 }
 
+func TestEqualityDeleteReadRejectsAmbiguousColumns(t *testing.T) {
+	tbl := newEqDeleteReadTestTable(t)
+	arrowSc, err := table.SchemaToArrowSchema(tbl.Metadata().CurrentSchema(), nil, false, false)
+	require.NoError(t, err)
+	dataPath := tbl.Location() + "/data/data.parquet"
+	writeParquetFile(t, dataPath, arrowSc, `[{"id": 1, "data": "one"}]`)
+	tx := tbl.NewTransaction()
+	require.NoError(t, tx.AddFiles(t.Context(), []string{dataPath}, nil, false))
+	tbl, err = tx.Commit(t.Context())
+	require.NoError(t, err)
+
+	deleteSchema, err := table.SchemaToArrowSchema(
+		iceberg.NewSchema(0, iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}),
+		nil, true, false)
+	require.NoError(t, err)
+	duplicateSchema := arrow.NewSchema([]arrow.Field{deleteSchema.Field(0), deleteSchema.Field(0)}, nil)
+	deletePath := tbl.Location() + "/data/delete.parquet"
+	builder := array.NewInt64Builder(memory.DefaultAllocator)
+	builder.Append(1)
+	first := builder.NewArray()
+	builder.Append(2)
+	second := builder.NewArray()
+	builder.Release()
+	batch := array.NewRecordBatch(duplicateSchema, []arrow.Array{first, second}, 1)
+	first.Release()
+	second.Release()
+	deleteTable := array.NewTableFromRecords(duplicateSchema, []arrow.RecordBatch{batch})
+	batch.Release()
+	file, err := iceio.LocalFS{}.Create(deletePath)
+	require.NoError(t, err)
+	require.NoError(t, pqarrow.WriteTable(deleteTable, file, 1,
+		parquet.NewWriterProperties(parquet.WithStats(true)), pqarrow.DefaultWriterProps()))
+	deleteTable.Release()
+	deleteBuilder, err := iceberg.NewDataFileBuilder(
+		*iceberg.UnpartitionedSpec, iceberg.EntryContentEqDeletes,
+		deletePath, iceberg.ParquetFile, nil, nil, nil, 1, 128)
+	require.NoError(t, err)
+	deleteBuilder.EqualityFieldIDs([]int{1})
+	tx = tbl.NewTransaction()
+	rd := tx.NewRowDelta(nil)
+	rd.AddDeletes(deleteBuilder.Build())
+	require.NoError(t, rd.Commit(t.Context()))
+	tbl, err = tx.Commit(t.Context())
+	require.NoError(t, err)
+
+	_, _, err = tbl.Scan().ToArrowRecords(t.Context())
+	require.ErrorContains(t, err, `equality delete column "id" is ambiguous`)
+}
+
 func TestEqualityDeleteDoesNotAffectSameSnapshot(t *testing.T) {
 	tbl := newEqDeleteReadTestTable(t)