laskoviymishka commented on code in PR #1617:
URL: https://github.com/apache/iceberg-go/pull/1617#discussion_r3719190256
##########
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:
Review Comment:
The fail-safe guard here is the right instinct, but we only apply it on the
delete-file side. The mirror lookup in `processEqualityDeletesColumnar` (around
line 550) does the same `r.Schema().FieldIndices(name)` and then takes
`indices[0]` unconditionally against the data record, so a data file with a
duplicate physical name for an equality column silently compares the wrong
column and drops the wrong rows. That's the same ambiguity we're rejecting
here, just on the read side, and it fails silently instead of loudly.
I'd apply the same guard at 550 so both paths fail the same way, and add a
test that puts the duplicate column in the data file rather than the delete
file, since the current test only exercises this branch. Once both sides are
covered, this is good to land.
##########
table/equality_delete_reader_test.go:
##########
@@ -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`)
Review Comment:
This asserts the literal error text, so a harmless rewording of the message
would break the test with no behavior change. We already have
`ErrEmptyEqualityFieldIDs` as a sentinel in this package, so I'd make the
ambiguity error a sentinel too (`ErrAmbiguousEqualityColumn`, wrapped with
`%w`) and assert with `require.ErrorIs`. Keeps the message free to change and
reads the intent more clearly.
##########
table/equality_delete_reader_test.go:
##########
@@ -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)
Review Comment:
`file` here is a WriteCloser but we never close it, and `pqarrow.WriteTable`
writes without closing. That leaks the fd and, depending on buffering, can
leave the parquet footer unflushed, so the later `tbl.Scan()` could read a
truncated file and flake.
`defer file.Close()` right after the NoError check fixes it, or we could
reuse the `writeParquetFile` helper with a custom schema. wdyt?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
