laskoviymishka commented on code in PR #1242:
URL: https://github.com/apache/iceberg-go/pull/1242#discussion_r3455871173
##########
table/equality_delete_reader.go:
##########
@@ -361,76 +371,120 @@ func makeColEncoder(arr arrow.Array) colEncoder {
vals := a.Int8Values()
return func(buf *bytes.Buffer, row int) {
+ if writeNullTagIfNull(buf, a, row) {
Review Comment:
This is the main thing I'd like to settle before the rest. This two-line
guard is now copy-pasted into all 11 numeric closures, and there's nothing
structural stopping the next fast-path someone adds from forgetting it — which
is exactly how this bug got in.
I'd lift it to a wrapper applied at the `makeColEncoder` return site:
```go
func withNullGuard(arr arrow.Array, enc colEncoder) colEncoder {
return func(buf *bytes.Buffer, row int) {
if writeNullTagIfNull(buf, arr, row) {
return
}
enc(buf, row)
}
}
```
Then each branch becomes `return withNullGuard(a, func(buf, row) {
buf.WriteByte(1); ... })` — one guard to review instead of eleven, and the
string/binary/boolean arms could fold into it too so we stop carrying three
null-guard idioms in one function. If the extra indirection has measurable
hot-path cost I'd keep the inline form but leave a comment marking it
load-bearing. wdyt?
##########
table/equality_delete_reader_test.go:
##########
@@ -280,3 +280,226 @@ func TestEqualityDeleteMultiColumnKey(t *testing.T) {
{id: 1, name: "charlie"},
}, rows)
}
+
+func TestEqualityDeleteNullableFastPathKeys(t *testing.T) {
+ tests := []struct {
+ name string
+ keyType iceberg.Type
+ dataJSON string
+ deleteJSON string
+ }{
+ {
+ name: "int",
+ keyType: iceberg.PrimitiveTypes.Int32,
+ dataJSON: `[
+ {"row_id": 1, "key": null},
+ {"row_id": 2, "key": 0},
+ {"row_id": 3, "key": 7}
+ ]`,
+ deleteJSON: `[{"key": null}]`,
Review Comment:
Every case here uses `deleteJSON: [{"key": null}]`, so we only ever prove a
null delete key matches null rows — the easy half. The half with the data-loss
teeth is the converse: a non-null delete key, especially `0`, must *not* delete
a null-key row.
Pre-fix, a null row encoded as `0x01` + a zero payload, byte-identical to a
real key of `0`. So `deleteJSON: [{"key": 0}]` against this exact data (row 1
null, row 2 = 0) would have deleted *both* rows, and should now delete only row
2. That case catches the bug from the dangerous direction and guards against a
future regression where the null tag collides with a zero value. Could we add a
subcase (per type, or a representative few) with `deleteJSON: [{"key": 0}]` and
the null row expected to survive?
##########
table/equality_delete_reader_internal_test.go:
##########
@@ -0,0 +1,190 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "bytes"
+ "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/stretchr/testify/assert"
+)
+
+func TestMakeColEncoderMatchesGenericForNullFastPathTypes(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ tests := []struct {
+ name string
+ arr arrow.Array
+ }{
+ {name: "int8", arr: buildNullableInt8Array(mem)},
+ {name: "int16", arr: buildNullableInt16Array(mem)},
+ {name: "int32", arr: buildNullableInt32Array(mem)},
+ {name: "int64", arr: buildNullableInt64Array(mem)},
+ {name: "float32", arr: buildNullableFloat32Array(mem)},
+ {name: "float64", arr: buildNullableFloat64Array(mem)},
+ {name: "date32", arr: buildNullableDate32Array(mem)},
+ {name: "date64", arr: buildNullableDate64Array(mem)},
+ {name: "time32", arr: buildNullableTime32Array(mem)},
+ {name: "time64", arr: buildNullableTime64Array(mem)},
+ {name: "timestamp", arr: buildNullableTimestampArray(mem)},
+ }
+
+ for _, tt := range tests {
+ defer tt.arr.Release()
Review Comment:
`defer tt.arr.Release()` inside the range loop stacks all 11 releases on the
parent function's defer list rather than running per-iteration — they all fire
when the test returns, after `mem.AssertSize`. It's correct today because the
subtests are synchronous, but it reads as if it's scoped per-iteration when it
isn't, and the day someone adds `t.Parallel()` in the `t.Run` closure the
parent can return while a subtest still holds a live ref. I'd move it inside
the `t.Run` closure so the lifetime matches the scope.
##########
table/equality_delete_reader_internal_test.go:
##########
@@ -0,0 +1,190 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "bytes"
+ "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/stretchr/testify/assert"
+)
+
+func TestMakeColEncoderMatchesGenericForNullFastPathTypes(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ tests := []struct {
+ name string
+ arr arrow.Array
+ }{
+ {name: "int8", arr: buildNullableInt8Array(mem)},
+ {name: "int16", arr: buildNullableInt16Array(mem)},
+ {name: "int32", arr: buildNullableInt32Array(mem)},
+ {name: "int64", arr: buildNullableInt64Array(mem)},
+ {name: "float32", arr: buildNullableFloat32Array(mem)},
+ {name: "float64", arr: buildNullableFloat64Array(mem)},
+ {name: "date32", arr: buildNullableDate32Array(mem)},
+ {name: "date64", arr: buildNullableDate64Array(mem)},
+ {name: "time32", arr: buildNullableTime32Array(mem)},
+ {name: "time64", arr: buildNullableTime64Array(mem)},
+ {name: "timestamp", arr: buildNullableTimestampArray(mem)},
+ }
+
+ for _, tt := range tests {
+ defer tt.arr.Release()
+
+ t.Run(tt.name, func(t *testing.T) {
+ encoder := makeColEncoder(tt.arr)
+
+ var fast, generic bytes.Buffer
+ for row := 0; row < tt.arr.Len(); row++ {
+ fast.Reset()
+ generic.Reset()
+
+ encoder(&fast, row)
+ encodeArrowValue(&generic, tt.arr, row)
+
+ assert.Equal(t, generic.Bytes(), fast.Bytes(),
"row %d", row)
+ }
+ })
+ }
+}
+
+func buildNullableInt8Array(mem memory.Allocator) arrow.Array {
+ builder := array.NewInt8Builder(mem)
+ defer builder.Release()
+
+ builder.Append(7)
Review Comment:
Every builder produces the same shape — `[non-null, null, non-null]` with
the null fixed at index 1. That leaves a few edges unexercised: null at index
0, null at `Len-1`, an all-null array, and (the one I'd actually want) an
all-valid array with `NullN == 0`. The all-valid case is the only thing that
proves `writeNullTagIfNull` is a true no-op when there's no null bitmap. A
4-element variant with nulls at the ends plus one all-valid subcase would round
this out.
##########
table/equality_delete_reader_test.go:
##########
@@ -280,3 +280,226 @@ func TestEqualityDeleteMultiColumnKey(t *testing.T) {
{id: 1, name: "charlie"},
}, rows)
}
+
+func TestEqualityDeleteNullableFastPathKeys(t *testing.T) {
+ tests := []struct {
+ name string
+ keyType iceberg.Type
+ dataJSON string
+ deleteJSON string
+ }{
+ {
+ name: "int",
+ keyType: iceberg.PrimitiveTypes.Int32,
+ dataJSON: `[
+ {"row_id": 1, "key": null},
+ {"row_id": 2, "key": 0},
+ {"row_id": 3, "key": 7}
+ ]`,
+ deleteJSON: `[{"key": null}]`,
+ },
+ {
+ name: "long",
+ keyType: iceberg.PrimitiveTypes.Int64,
+ dataJSON: `[
+ {"row_id": 1, "key": null},
+ {"row_id": 2, "key": 0},
+ {"row_id": 3, "key": 7}
+ ]`,
+ deleteJSON: `[{"key": null}]`,
+ },
+ {
+ name: "float",
+ keyType: iceberg.PrimitiveTypes.Float32,
+ dataJSON: `[
+ {"row_id": 1, "key": null},
+ {"row_id": 2, "key": 0.0},
+ {"row_id": 3, "key": 7.25}
+ ]`,
+ deleteJSON: `[{"key": null}]`,
+ },
+ {
+ name: "double",
+ keyType: iceberg.PrimitiveTypes.Float64,
+ dataJSON: `[
+ {"row_id": 1, "key": null},
+ {"row_id": 2, "key": 0.0},
+ {"row_id": 3, "key": 7.25}
+ ]`,
+ deleteJSON: `[{"key": null}]`,
+ },
+ {
+ name: "date",
+ keyType: iceberg.PrimitiveTypes.Date,
+ dataJSON: `[
+ {"row_id": 1, "key": null},
+ {"row_id": 2, "key": "1970-01-01"},
+ {"row_id": 3, "key": "2024-01-02"}
+ ]`,
+ deleteJSON: `[{"key": null}]`,
+ },
+ {
+ name: "timestamp",
+ keyType: iceberg.PrimitiveTypes.Timestamp,
+ dataJSON: `[
+ {"row_id": 1, "key": null},
+ {"row_id": 2, "key":
"1970-01-01T00:00:00.000000Z"},
+ {"row_id": 3, "key":
"2024-01-02T03:04:05.000000Z"}
+ ]`,
+ deleteJSON: `[{"key": null}]`,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ tbl := newNullableEqDeleteReadTestTable(t, tt.keyType)
+ arrowSc, err :=
table.SchemaToArrowSchema(tbl.Metadata().CurrentSchema(), nil, false, false)
+ require.NoError(t, err)
+
+ dataPath := tbl.Location() + "/data/data-001.parquet"
+ writeParquetFile(t, dataPath, arrowSc, tt.dataJSON)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(),
[]string{dataPath}, nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ delArrowSc, err := table.SchemaToArrowSchema(
+ iceberg.NewSchema(0, iceberg.NestedField{
+ ID: 2, Name: "key", Type: tt.keyType,
Required: false,
+ }), nil, true, false)
+ require.NoError(t, err)
+
+ eqDelPath := tbl.Location() +
"/data/eq-del-null.parquet"
+ writeParquetFile(t, eqDelPath, delArrowSc,
tt.deleteJSON)
+
+ eqDelBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec,
iceberg.EntryContentEqDeletes,
+ eqDelPath, iceberg.ParquetFile, nil, nil, nil,
1, 128)
+ require.NoError(t, err)
+ eqDelBuilder.EqualityFieldIDs([]int{2})
+
+ tx2 := tbl.NewTransaction()
+ rd := tx2.NewRowDelta(nil)
+ rd.AddDeletes(eqDelBuilder.Build())
+ require.NoError(t, rd.Commit(t.Context()))
+ tbl, err = tx2.Commit(t.Context())
+ require.NoError(t, err)
+
+ assert.Equal(t, []int64{2, 3}, collectRowIDs(t, tbl))
+ })
+ }
+}
+
+func TestEqualityDeleteNullableCompositeFastPathKey(t *testing.T) {
Review Comment:
This composite case only covers `{id: 7, deleted_at: null}` — one non-null
field, one null. The concat case I'd want is the all-null key `{id: null,
deleted_at: null}` producing `[0x00][0x00]`, which is what proves key
concatenation handles a fully-null composite. Worth a subcase where both fields
are null and only the matching all-null row is deleted.
##########
table/equality_delete_reader_test.go:
##########
@@ -280,3 +280,226 @@ func TestEqualityDeleteMultiColumnKey(t *testing.T) {
{id: 1, name: "charlie"},
}, rows)
}
+
+func TestEqualityDeleteNullableFastPathKeys(t *testing.T) {
+ tests := []struct {
+ name string
+ keyType iceberg.Type
+ dataJSON string
+ deleteJSON string
+ }{
+ {
+ name: "int",
+ keyType: iceberg.PrimitiveTypes.Int32,
+ dataJSON: `[
+ {"row_id": 1, "key": null},
+ {"row_id": 2, "key": 0},
+ {"row_id": 3, "key": 7}
+ ]`,
+ deleteJSON: `[{"key": null}]`,
+ },
+ {
+ name: "long",
+ keyType: iceberg.PrimitiveTypes.Int64,
+ dataJSON: `[
+ {"row_id": 1, "key": null},
+ {"row_id": 2, "key": 0},
+ {"row_id": 3, "key": 7}
+ ]`,
+ deleteJSON: `[{"key": null}]`,
+ },
+ {
+ name: "float",
+ keyType: iceberg.PrimitiveTypes.Float32,
+ dataJSON: `[
+ {"row_id": 1, "key": null},
+ {"row_id": 2, "key": 0.0},
+ {"row_id": 3, "key": 7.25}
+ ]`,
+ deleteJSON: `[{"key": null}]`,
+ },
+ {
+ name: "double",
+ keyType: iceberg.PrimitiveTypes.Float64,
+ dataJSON: `[
+ {"row_id": 1, "key": null},
+ {"row_id": 2, "key": 0.0},
+ {"row_id": 3, "key": 7.25}
+ ]`,
+ deleteJSON: `[{"key": null}]`,
+ },
+ {
+ name: "date",
+ keyType: iceberg.PrimitiveTypes.Date,
+ dataJSON: `[
+ {"row_id": 1, "key": null},
+ {"row_id": 2, "key": "1970-01-01"},
+ {"row_id": 3, "key": "2024-01-02"}
+ ]`,
+ deleteJSON: `[{"key": null}]`,
+ },
+ {
+ name: "timestamp",
+ keyType: iceberg.PrimitiveTypes.Timestamp,
+ dataJSON: `[
+ {"row_id": 1, "key": null},
+ {"row_id": 2, "key":
"1970-01-01T00:00:00.000000Z"},
+ {"row_id": 3, "key":
"2024-01-02T03:04:05.000000Z"}
+ ]`,
+ deleteJSON: `[{"key": null}]`,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ tbl := newNullableEqDeleteReadTestTable(t, tt.keyType)
+ arrowSc, err :=
table.SchemaToArrowSchema(tbl.Metadata().CurrentSchema(), nil, false, false)
+ require.NoError(t, err)
+
+ dataPath := tbl.Location() + "/data/data-001.parquet"
+ writeParquetFile(t, dataPath, arrowSc, tt.dataJSON)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(),
[]string{dataPath}, nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ delArrowSc, err := table.SchemaToArrowSchema(
+ iceberg.NewSchema(0, iceberg.NestedField{
+ ID: 2, Name: "key", Type: tt.keyType,
Required: false,
+ }), nil, true, false)
+ require.NoError(t, err)
+
+ eqDelPath := tbl.Location() +
"/data/eq-del-null.parquet"
+ writeParquetFile(t, eqDelPath, delArrowSc,
tt.deleteJSON)
+
+ eqDelBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec,
iceberg.EntryContentEqDeletes,
+ eqDelPath, iceberg.ParquetFile, nil, nil, nil,
1, 128)
+ require.NoError(t, err)
+ eqDelBuilder.EqualityFieldIDs([]int{2})
+
+ tx2 := tbl.NewTransaction()
+ rd := tx2.NewRowDelta(nil)
+ rd.AddDeletes(eqDelBuilder.Build())
+ require.NoError(t, rd.Commit(t.Context()))
+ tbl, err = tx2.Commit(t.Context())
+ require.NoError(t, err)
+
+ assert.Equal(t, []int64{2, 3}, collectRowIDs(t, tbl))
+ })
+ }
+}
+
+func TestEqualityDeleteNullableCompositeFastPathKey(t *testing.T) {
+ location := filepath.ToSlash(t.TempDir())
+
+ iceSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "row_id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "id", Type:
iceberg.PrimitiveTypes.Int32, Required: false},
+ iceberg.NestedField{ID: 3, Name: "deleted_at", Type:
iceberg.PrimitiveTypes.Timestamp, Required: false},
+ )
+
+ meta, err := table.NewMetadata(iceSchema, iceberg.UnpartitionedSpec,
+ table.UnsortedSortOrder, location,
+ iceberg.Properties{table.PropertyFormatVersion: "2"})
+ require.NoError(t, err)
+
+ tbl := table.New(
+ table.Identifier{"db", "eq_del_nullable_composite"},
+ meta, location+"/metadata/v1.metadata.json",
+ func(ctx context.Context) (iceio.IO, error) {
+ return iceio.LocalFS{}, nil
+ },
+ &rowDeltaCatalog{metadata: meta},
+ )
+
+ arrowSc, err := table.SchemaToArrowSchema(iceSchema, nil, false, false)
+ require.NoError(t, err)
+
+ dataPath := location + "/data/data-001.parquet"
+ writeParquetFile(t, dataPath, arrowSc, `[
+ {"row_id": 1, "id": 7, "deleted_at": null},
+ {"row_id": 2, "id": 7, "deleted_at":
"1970-01-01T00:00:00.000000Z"},
+ {"row_id": 3, "id": 8, "deleted_at": null}
+ ]`)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(), []string{dataPath}, nil,
false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ delArrowSc, err := table.SchemaToArrowSchema(
+ iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 2, Name: "id", Type:
iceberg.PrimitiveTypes.Int32, Required: false},
+ iceberg.NestedField{ID: 3, Name: "deleted_at", Type:
iceberg.PrimitiveTypes.Timestamp, Required: false},
+ ), nil, true, false)
+ require.NoError(t, err)
+
+ eqDelPath := location + "/data/eq-del-composite.parquet"
+ writeParquetFile(t, eqDelPath, delArrowSc, `[{"id": 7, "deleted_at":
null}]`)
+
+ eqDelBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentEqDeletes,
+ eqDelPath, iceberg.ParquetFile, nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+ eqDelBuilder.EqualityFieldIDs([]int{2, 3})
+
+ tx2 := tbl.NewTransaction()
+ rd := tx2.NewRowDelta(nil)
+ rd.AddDeletes(eqDelBuilder.Build())
+ require.NoError(t, rd.Commit(t.Context()))
+ tbl, err = tx2.Commit(t.Context())
+ require.NoError(t, err)
+
+ assert.Equal(t, []int64{2, 3}, collectRowIDs(t, tbl))
+}
+
+func newNullableEqDeleteReadTestTable(t *testing.T, keyType iceberg.Type)
*table.Table {
+ t.Helper()
+
+ location := filepath.ToSlash(t.TempDir())
+
+ iceSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "row_id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "key", Type: keyType,
Required: false},
+ )
+
+ meta, err := table.NewMetadata(iceSchema, iceberg.UnpartitionedSpec,
+ table.UnsortedSortOrder, location,
+ iceberg.Properties{table.PropertyFormatVersion: "2"})
+ require.NoError(t, err)
+
+ return table.New(
+ table.Identifier{"db", "eq_del_nullable_" + keyType.String()},
+ meta, location+"/metadata/v1.metadata.json",
+ func(ctx context.Context) (iceio.IO, error) {
+ return iceio.LocalFS{}, nil
+ },
+ &rowDeltaCatalog{metadata: meta},
+ )
+}
+
+func collectRowIDs(t *testing.T, tbl *table.Table) []int64 {
+ t.Helper()
+
+ _, itr, err := tbl.Scan().ToArrowRecords(t.Context())
+ require.NoError(t, err)
+
+ var ids []int64
+ for rec, err := range itr {
+ require.NoError(t, err)
+
+ indices := rec.Schema().FieldIndices("row_id")
+ require.NotEmpty(t, indices)
+
+ col := rec.Column(indices[0]).(*array.Int64)
Review Comment:
Bare type assertion will panic with an unhelpful message if the column ever
comes back as something other than `Int64`. Safe given the hardcoded schema,
but `require.IsType` or the two-value form keeps the failure readable. Minor
companion just below: the loop uses `for i := 0; i < col.Len(); i++` — `for i
:= range col.Len()` is available on Go 1.25. Both very minor.
--
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]
