This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 019b76781 fix(table): reject ambiguous row-id columns in position
deltas (#1616)
019b76781 is described below
commit 019b767816aeff557c71b49b1b08287f20b1c669
Author: Minh Vu <[email protected]>
AuthorDate: Tue Aug 4 12:17:47 2026 +0200
fix(table): reject ambiguous row-id columns in position deltas (#1616)
## What changed
Require position-delta batches to contain at most one `_row_id` column.
The shared lookup is used by `Reinsert`, `Insert`, and the null-column
append path.
## Why
Arrow schemas can contain duplicate field names. Selecting the first
`_row_id` silently made reserved row-lineage metadata ambiguous and
could ignore a conflicting or malformed second column.
## Testing
- Added duplicate `_row_id` coverage for both `Reinsert` and `Insert`
- `go test ./table`
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/position_delta_writer.go | 36 ++++++++++++++++++++++++-----
table/position_delta_writer_test.go | 46 +++++++++++++++++++++++++++++++++++++
2 files changed, 76 insertions(+), 6 deletions(-)
diff --git a/table/position_delta_writer.go b/table/position_delta_writer.go
index e7c97541f..14dc00a2e 100644
--- a/table/position_delta_writer.go
+++ b/table/position_delta_writer.go
@@ -114,13 +114,16 @@ func (w *PositionDeltaWriter) Reinsert(batch
arrow.RecordBatch) error {
return fmt.Errorf("%w: writer is already closed",
ErrInvalidOperation)
}
- indices := batch.Schema().FieldIndices(iceberg.RowIDColumnName)
- if len(indices) == 0 {
+ rowIDIndex, found, err := rowIDColumnIndex(batch.Schema())
+ if err != nil {
+ return err
+ }
+ if !found {
return fmt.Errorf("%w: Reinsert batch must contain %s column",
iceberg.ErrInvalidArgument, iceberg.RowIDColumnName)
}
- col := batch.Column(indices[0])
+ col := batch.Column(rowIDIndex)
if col.NullN() > 0 {
return fmt.Errorf("%w: Reinsert batch %s column must not
contain null values",
iceberg.ErrInvalidArgument, iceberg.RowIDColumnName)
@@ -140,8 +143,12 @@ func (w *PositionDeltaWriter) Insert(batch
arrow.RecordBatch) error {
return fmt.Errorf("%w: writer is already closed",
ErrInvalidOperation)
}
- if indices := batch.Schema().FieldIndices(iceberg.RowIDColumnName);
len(indices) > 0 {
- if batch.Column(indices[0]).NullN() != int(batch.NumRows()) {
+ rowIDIndex, found, err := rowIDColumnIndex(batch.Schema())
+ if err != nil {
+ return err
+ }
+ if found {
+ if batch.Column(rowIDIndex).NullN() != int(batch.NumRows()) {
return fmt.Errorf("%w: Insert batch %s column must be
all null (use Reinsert for preserved IDs)",
iceberg.ErrInvalidArgument,
iceberg.RowIDColumnName)
}
@@ -245,7 +252,11 @@ func (w *PositionDeltaWriter) buildUnifiedIterator()
iter.Seq2[arrow.RecordBatch
// already has _row_id (validated to be all-null by Insert), it is returned
// retained as-is.
func appendNullRowIDColumn(alloc memory.Allocator, batch arrow.RecordBatch)
(arrow.RecordBatch, error) {
- if indices := batch.Schema().FieldIndices(iceberg.RowIDColumnName);
len(indices) > 0 {
+ _, found, err := rowIDColumnIndex(batch.Schema())
+ if err != nil {
+ return nil, err
+ }
+ if found {
batch.Retain()
return batch, nil
@@ -278,3 +289,16 @@ func appendNullRowIDColumn(alloc memory.Allocator, batch
arrow.RecordBatch) (arr
return array.NewRecordBatch(newSchema, cols, nrows), nil
}
+
+func rowIDColumnIndex(schema *arrow.Schema) (int, bool, error) {
+ indices := schema.FieldIndices(iceberg.RowIDColumnName)
+ switch len(indices) {
+ case 0:
+ return 0, false, nil
+ case 1:
+ return indices[0], true, nil
+ default:
+ return 0, false, fmt.Errorf("%w: batch contains multiple %s
columns",
+ iceberg.ErrInvalidArgument, iceberg.RowIDColumnName)
+ }
+}
diff --git a/table/position_delta_writer_test.go
b/table/position_delta_writer_test.go
index 55eb524a0..629d3931a 100644
--- a/table/position_delta_writer_test.go
+++ b/table/position_delta_writer_test.go
@@ -280,3 +280,49 @@ func
TestPositionDeltaWriter_ReinsertRejectsMissingColumn(t *testing.T) {
require.Error(t, err)
assert.Contains(t, err.Error(), "must contain _row_id")
}
+
+func TestPositionDeltaWriter_RejectsDuplicateRowIDColumns(t *testing.T) {
+ for _, layout := range []struct {
+ name string
+ appendRows func(first, second *array.Int64Builder)
+ }{
+ {"both null", func(first, second *array.Int64Builder) {
+ first.AppendNull()
+ second.AppendNull()
+ }},
+ {"both non-null", func(first, second *array.Int64Builder) {
+ first.Append(10)
+ second.Append(11)
+ }},
+ } {
+ t.Run(layout.name, func(t *testing.T) {
+ schema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64},
+ {Name: iceberg.RowIDColumnName, Type:
arrow.PrimitiveTypes.Int64, Nullable: true},
+ {Name: iceberg.RowIDColumnName, Type:
arrow.PrimitiveTypes.Int64, Nullable: true},
+ }, nil)
+ builder :=
array.NewRecordBuilder(memory.DefaultAllocator, schema)
+ defer builder.Release()
+ builder.Field(0).(*array.Int64Builder).Append(1)
+
layout.appendRows(builder.Field(1).(*array.Int64Builder),
builder.Field(2).(*array.Int64Builder))
+ batch := builder.NewRecordBatch()
+ defer batch.Release()
+
+ for _, operation := range []struct {
+ name string
+ call func(*table.PositionDeltaWriter) error
+ }{
+ {"Reinsert", func(w *table.PositionDeltaWriter)
error { return w.Reinsert(batch) }},
+ {"Insert", func(w *table.PositionDeltaWriter)
error { return w.Insert(batch) }},
+ } {
+ t.Run(operation.name, func(t *testing.T) {
+ writer, err :=
table.NewPositionDeltaWriter(newV3RowLineageTestTable(t))
+ require.NoError(t, err)
+ err = operation.call(writer)
+ require.ErrorIs(t, err,
iceberg.ErrInvalidArgument)
+ require.ErrorContains(t, err, "multiple
_row_id columns")
+ })
+ }
+ })
+ }
+}
