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 912383979 perf(table): fast-path single int64 equality deletes (#2157)
912383979 is described below
commit 9123839791c3e28277fc87848c4555bfc253db68
Author: Minh Vu <[email protected]>
AuthorDate: Sun Oct 11 21:22:26 2026 +0200
perf(table): fast-path single int64 equality deletes (#2157)
---
table/equality_delete_reader.go | 179 ++++++++++++++++---
table/equality_delete_reader_bench_test.go | 108 ++++++++++++
table/equality_delete_reader_internal_test.go | 244 ++++++++++++++++++++++++++
3 files changed, 509 insertions(+), 22 deletions(-)
diff --git a/table/equality_delete_reader.go b/table/equality_delete_reader.go
index 400965701..e2f75f637 100644
--- a/table/equality_delete_reader.go
+++ b/table/equality_delete_reader.go
@@ -51,12 +51,18 @@ var ErrConflictingEqualityDeleteMetadata =
errors.New("conflicting equality dele
// equalityDeleteSet holds the set of delete keys and the column names
// used to look them up in data records. Each set corresponds to one
// group of equality field IDs — delete files with different field IDs
-// produce separate sets. The set is immutable after construction so it
-// can be shared by tasks with the same delete files.
+// produce separate sets. Key material and metadata are immutable after
+// construction; derived typed indexes are built once and shared safely.
+// Typed indexes retain additional map storage alongside keys for the lifetime
+// of the set. Merged groups can hold their own indexes on top of per-file
ones.
+// Do not copy the set after the first use of an index.
type equalityDeleteSet struct {
keys set[string]
fieldIDs []int
colNames []string
+
+ singleInt64Once sync.Once
+ singleInt64 *singleInt64EqualityDeleteSet
}
type arrowFieldRef struct {
@@ -1288,6 +1294,125 @@ func makeColEncoder(arr arrow.Array) colEncoder {
}
}
+type singleInt64EqualityDeleteSet struct {
+ keys set[uint64]
+ hasNull bool
+}
+
+func (e *equalityDeleteSet) singleInt64Set(fileSchema *iceberg.Schema)
*singleInt64EqualityDeleteSet {
+ if fileSchema == nil || len(e.fieldIDs) != 1 {
+ return nil
+ }
+
+ field, ok := fileSchema.FindFieldByID(e.fieldIDs[0])
+ if !ok {
+ return nil
+ }
+ if _, ok := field.Type.(iceberg.Int64Type); !ok {
+ return nil
+ }
+
+ e.singleInt64Once.Do(func() {
+ result := &singleInt64EqualityDeleteSet{
+ keys: make(set[uint64], len(e.keys)),
+ }
+ for key := range e.keys {
+ if len(key) == 1 && key[0] == 0 {
+ result.hasNull = true
+
+ continue
+ }
+ if len(key) != 9 || key[0] != 1 {
+ // Unexpected key formats must keep using the
generic path.
+ return
+ }
+
+ value := uint64(key[1])<<56 |
+ uint64(key[2])<<48 |
+ uint64(key[3])<<40 |
+ uint64(key[4])<<32 |
+ uint64(key[5])<<24 |
+ uint64(key[6])<<16 |
+ uint64(key[7])<<8 |
+ uint64(key[8])
+ result.keys[value] = struct{}{}
+ }
+
+ e.singleInt64 = result
+ })
+
+ return e.singleInt64
+}
+
+type equalityDeleteMask struct {
+ buffer *memory.Buffer
+ bits []byte
+}
+
+func (m *equalityDeleteMask) ensure(mem memory.Allocator, numRows int) {
+ if m.buffer != nil {
+ return
+ }
+
+ m.buffer = memory.NewResizableBuffer(mem)
+ m.buffer.Resize(int(bitutil.BytesForBits(int64(numRows))))
+ m.bits = m.buffer.Bytes()
+ for i := range m.bits {
+ m.bits[i] = 0xFF
+ }
+}
+
+func (m *equalityDeleteMask) release() {
+ if m.buffer != nil {
+ m.buffer.Release()
+ }
+}
+
+// applySingleInt64 returns false for unexpected Arrow column types, allowing
+// the caller to fall back to the generic field encoder.
+func (m *equalityDeleteMask) applySingleInt64(
+ column arrow.Array,
+ deleteSet *singleInt64EqualityDeleteSet,
+ mem memory.Allocator,
+) bool {
+ values, ok := column.(*array.Int64)
+ if !ok {
+ return false
+ }
+
+ rawValues := values.Int64Values()
+ numRows := values.Len()
+ if values.NullN() == 0 {
+ for row := range numRows {
+ if m.bits != nil && !bitutil.BitIsSet(m.bits, row) {
+ continue
+ }
+ if _, deleted :=
deleteSet.keys[uint64(rawValues[row])]; deleted {
+ m.ensure(mem, numRows)
+ bitutil.ClearBit(m.bits, row)
+ }
+ }
+ } else {
+ for row := range numRows {
+ if m.bits != nil && !bitutil.BitIsSet(m.bits, row) {
+ continue
+ }
+
+ if values.IsNull(row) {
+ if !deleteSet.hasNull {
+ continue
+ }
+ } else if _, deleted :=
deleteSet.keys[uint64(rawValues[row])]; !deleted {
+ continue
+ }
+ m.ensure(mem, numRows)
+ bitutil.ClearBit(m.bits, row)
+ }
+ }
+
+ return true
+}
+
// processEqualityDeletesColumnarForFile resolves field paths once per file and
// typed column encoders once per batch, then iterates rows without per-row
type
// switches. Each delete set is applied independently because sets may have
@@ -1323,18 +1448,38 @@ func processEqualityDeletesColumnarForFile(ctx
context.Context, eqDeleteSets []*
}
}
+ singleInt64Sets := make([]*singleInt64EqualityDeleteSet,
len(eqDeleteSets))
+ for i, eqDel := range eqDeleteSets {
+ // Nested fields need parent validity checks and use the
generic path.
+ // Do not duplicate their encoded keys in an index that cannot
be used.
+ if len(fieldRefs[i]) == 1 && len(fieldRefs[i][0].path) == 1 {
+ singleInt64Sets[i] = eqDel.singleInt64Set(fileSchema)
+ }
+ }
+
return func(r arrow.RecordBatch) (arrow.RecordBatch, error) {
defer r.Release()
mem := compute.GetAllocator(ctx)
numRows := int(r.NumRows())
- var maskBuf *memory.Buffer
- var maskBytes []byte
+ var mask equalityDeleteMask
+ defer mask.release()
var keyBuf bytes.Buffer
for setIdx, eqDel := range eqDeleteSets {
+ if int64Set := singleInt64Sets[setIdx]; int64Set != nil
{
+ column, _, err := arrowArraysAtField(
+ r, fieldRefs[setIdx][0],
eqDel.fieldIDs[0], eqDel.colNames[0], dataFilePath)
+ if err != nil {
+ return nil, err
+ }
+ if mask.applySingleInt64(column, int64Set, mem)
{
+ continue
+ }
+ }
+
encoders := make([]colEncoder, len(eqDel.colNames))
for i, name := range eqDel.colNames {
var err error
@@ -1345,7 +1490,7 @@ func processEqualityDeletesColumnarForFile(ctx
context.Context, eqDeleteSets []*
}
for row := range numRows {
- if maskBytes != nil &&
!bitutil.BitIsSet(maskBytes, row) {
+ if mask.bits != nil &&
!bitutil.BitIsSet(mask.bits, row) {
continue
}
@@ -1359,22 +1504,12 @@ func processEqualityDeletesColumnarForFile(ctx
context.Context, eqDeleteSets []*
continue
}
- if maskBuf == nil {
- maskBuf = memory.NewResizableBuffer(mem)
- defer maskBuf.Release()
-
maskBuf.Resize(int(bitutil.BytesForBits(int64(numRows))))
- maskBytes = maskBuf.Bytes()
-
- for i := range maskBytes {
- maskBytes[i] = 0xFF
- }
- }
-
- bitutil.ClearBit(maskBytes, row)
+ mask.ensure(mem, numRows)
+ bitutil.ClearBit(mask.bits, row)
}
}
- if maskBuf == nil {
+ if mask.buffer == nil {
r.Retain()
return r, nil
@@ -1382,14 +1517,14 @@ func processEqualityDeletesColumnarForFile(ctx
context.Context, eqDeleteSets []*
maskData := array.NewData(
arrow.FixedWidthTypes.Boolean, numRows,
- []*memory.Buffer{nil, maskBuf}, nil, 0, 0)
- mask := array.NewBooleanData(maskData)
+ []*memory.Buffer{nil, mask.buffer}, nil, 0, 0)
+ filterMask := array.NewBooleanData(maskData)
maskData.Release()
- defer mask.Release()
+ defer filterMask.Release()
filtered, err := compute.Filter(ctx,
compute.NewDatumWithoutOwning(r),
- compute.NewDatumWithoutOwning(mask),
+ compute.NewDatumWithoutOwning(filterMask),
*compute.DefaultFilterOptions())
if err != nil {
return nil, err
diff --git a/table/equality_delete_reader_bench_test.go
b/table/equality_delete_reader_bench_test.go
index bcdb536a7..4e55edc5d 100644
--- a/table/equality_delete_reader_bench_test.go
+++ b/table/equality_delete_reader_bench_test.go
@@ -528,6 +528,114 @@ func buildBenchDeleteSetStringNoMatch(numDeletes int)
*equalityDeleteSet {
}
}
+func buildBenchDeleteSetSingleInt64(numDeletes int) *equalityDeleteSet {
+ keys := make(set[string], numDeletes)
+ var buf bytes.Buffer
+
+ for i := range numDeletes {
+ buf.Reset()
+ buf.WriteByte(1)
+ bufPutUint64(&buf, uint64(int64(i)*3))
+ keys[buf.String()] = struct{}{}
+ }
+
+ return &equalityDeleteSet{
+ keys: keys,
+ fieldIDs: []int{1},
+ colNames: []string{"id"},
+ }
+}
+
+func buildBenchDeleteSetSingleInt64NoMatch(numDeletes int) *equalityDeleteSet {
+ keys := make(set[string], numDeletes)
+ var buf bytes.Buffer
+
+ for i := range numDeletes {
+ buf.Reset()
+ buf.WriteByte(1)
+ bufPutUint64(&buf, uint64(-int64(i)-1))
+ keys[buf.String()] = struct{}{}
+ }
+
+ return &equalityDeleteSet{
+ keys: keys,
+ fieldIDs: []int{1},
+ colNames: []string{"id"},
+ }
+}
+
+func buildBenchRecordNullableInt(mem memory.Allocator, numRows int)
arrow.RecordBatch {
+ schema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: true},
+ {Name: "category", Type: arrow.PrimitiveTypes.Int64},
+ }, nil)
+ builder := array.NewRecordBuilder(mem, schema)
+ defer builder.Release()
+ ids := builder.Field(0).(*array.Int64Builder)
+ categories := builder.Field(1).(*array.Int64Builder)
+
+ for i := range numRows {
+ // Keep 25% of rows null to exercise the null-aware lookup loop.
+ if i%4 == 0 {
+ ids.AppendNull()
+ } else {
+ ids.Append(int64(i))
+ }
+ categories.Append(int64(i % 100))
+ }
+
+ return builder.NewRecordBatch()
+}
+
+func benchSingleInt64EqualityDeletes(
+ b *testing.B,
+ buildRec func(memory.Allocator, int) arrow.RecordBatch,
+ buildDel func(int) *equalityDeleteSet,
+) {
+ b.Helper()
+
+ for _, benchmark := range []struct {
+ name string
+ fast bool
+ }{
+ {name: "generic"},
+ {name: "fast", fast: true},
+ } {
+ b.Run(benchmark.name, func(b *testing.B) {
+ benchEqDeletesForFile(b, buildRec, func(numDeletes int)
*equalityDeleteSet {
+ deleteSet := buildDel(numDeletes)
+ if !benchmark.fast {
+ // Keep the same schema and keys while
disabling the optional index.
+ deleteSet.singleInt64Once.Do(func() {})
+ }
+
+ return deleteSet
+ }, benchIntFileSchema())
+ })
+ }
+}
+
+func BenchmarkProcessEqualityDeletesSingleInt64(b *testing.B) {
+ benchSingleInt64EqualityDeletes(b, buildBenchRecordInt,
buildBenchDeleteSetSingleInt64)
+}
+
+func BenchmarkProcessEqualityDeletesSingleInt64NoMatch(b *testing.B) {
+ benchSingleInt64EqualityDeletes(b, buildBenchRecordInt,
buildBenchDeleteSetSingleInt64NoMatch)
+}
+
+func BenchmarkProcessEqualityDeletesSingleInt64NullsRetained(b *testing.B) {
+ benchSingleInt64EqualityDeletes(b, buildBenchRecordNullableInt,
buildBenchDeleteSetSingleInt64)
+}
+
+func BenchmarkProcessEqualityDeletesSingleInt64NullsDeleted(b *testing.B) {
+ benchSingleInt64EqualityDeletes(b, buildBenchRecordNullableInt,
func(numDeletes int) *equalityDeleteSet {
+ deleteSet := buildBenchDeleteSetSingleInt64(numDeletes)
+ deleteSet.keys[string([]byte{0})] = struct{}{}
+
+ return deleteSet
+ })
+}
+
func BenchmarkProcessEqualityDeletesInt(b *testing.B) {
benchEqDeletes(b, buildBenchRecordInt, buildBenchDeleteSetInt,
benchIntFileSchema())
}
diff --git a/table/equality_delete_reader_internal_test.go
b/table/equality_delete_reader_internal_test.go
index 0368e7fd8..3870bba85 100644
--- a/table/equality_delete_reader_internal_test.go
+++ b/table/equality_delete_reader_internal_test.go
@@ -22,6 +22,7 @@ import (
"context"
"errors"
"fmt"
+ "math"
"os"
"path/filepath"
"strings"
@@ -864,6 +865,249 @@ func
TestProcessEqualityDeletesReturnsOriginalBatchWhenNoRowsMatch(t *testing.T)
result.Release()
}
+func TestProcessEqualityDeletesSingleInt64MatchesGeneric(t *testing.T) {
+ fileSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64},
+ )
+ schema := arrow.NewSchema([]arrow.Field{{
+ Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: true,
+ }}, nil)
+ values := []int64{math.MinInt64, math.MaxInt64, -1, 0, 1, 7, 0, 42}
+ deleteValues := []int64{math.MinInt64, math.MaxInt64, -1, 0}
+ valid := []bool{true, true, true, false, true, true, false, true}
+
+ for _, tt := range []struct {
+ name string
+ valid []bool
+ deletes []int64
+ hasNull bool
+ wantRows int64
+ wantNulls int
+ }{
+ {name: "integer extremes", deletes: deleteValues, wantRows: 3},
+ {name: "nulls retained", valid: valid, deletes: deleteValues,
wantRows: 5, wantNulls: 2},
+ {name: "nulls deleted", valid: valid, deletes: deleteValues,
hasNull: true, wantRows: 3},
+ {name: "no matches", valid: valid, deletes: []int64{99},
wantRows: 8, wantNulls: 2},
+ {name: "empty set", valid: valid, wantRows: 8, wantNulls: 2},
+ } {
+ for _, offset := range []int{0, 3, 9} {
+ t.Run(fmt.Sprintf("%s/offset=%d", tt.name, offset),
func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(t.Context(), mem)
+
+ builder := array.NewInt64Builder(mem)
+ builder.AppendValues(make([]int64, offset), nil)
+ builder.AppendValues(values, tt.valid)
+ full := builder.NewArray()
+ builder.Release()
+ sliced := array.NewSlice(full, int64(offset),
int64(offset+len(values)))
+ full.Release()
+ record := array.NewRecordBatch(schema,
[]arrow.Array{sliced}, int64(len(values)))
+ sliced.Release()
+ defer record.Release()
+
+ deleteSet :=
int64EqualityDeleteSet(tt.deletes...)
+ if tt.hasNull {
+ deleteSet.keys[string([]byte{0})] =
struct{}{}
+ }
+ genericSet := &equalityDeleteSet{
+ keys: deleteSet.keys, fieldIDs:
deleteSet.fieldIDs, colNames: deleteSet.colNames,
+ }
+ // Disable only the optional index, keeping the
same schema and keys.
+ genericSet.singleInt64Once.Do(func() {})
+ generic, err :=
processEqualityDeletesColumnarForFile(ctx,
+ []*equalityDeleteSet{genericSet},
fileSchema, "data.parquet")
+ require.NoError(t, err)
+ require.Nil(t, genericSet.singleInt64)
+ record.Retain()
+ want, err := generic(record)
+ require.NoError(t, err)
+ defer want.Release()
+
+ fast, err :=
processEqualityDeletesColumnarForFile(ctx,
+ []*equalityDeleteSet{deleteSet},
fileSchema, "data.parquet")
+ require.NoError(t, err)
+ require.NotNil(t, deleteSet.singleInt64, "the
typed path must be exercised")
+ record.Retain()
+ got, err := fast(record)
+ require.NoError(t, err)
+ defer got.Release()
+
+ assert.True(t, array.RecordEqual(want, got),
"generic: %s; fast: %s", want, got)
+ assert.Equal(t, tt.wantRows, got.NumRows())
+ assert.Equal(t, tt.wantNulls,
got.Column(0).NullN())
+ if tt.wantRows == record.NumRows() {
+ assert.Same(t, record, want)
+ assert.Same(t, record, got)
+ }
+ })
+ }
+ }
+}
+
+func TestProcessEqualityDeletesCombinesFastAndGenericSets(t *testing.T) {
+ for _, nullable := range []bool{false, true} {
+ for _, genericFirst := range []bool{false, true} {
+ t.Run(fmt.Sprintf("nullable=%t/generic_first=%t",
nullable, genericFirst), func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(t.Context(), mem)
+
+ schema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type:
arrow.PrimitiveTypes.Int64, Nullable: true},
+ {Name: "category", Type:
arrow.BinaryTypes.String},
+ }, nil)
+ builder := array.NewRecordBuilder(mem, schema)
+ ids := builder.Field(0).(*array.Int64Builder)
+ ids.AppendValues([]int64{1, 2, 3, 4, 5, 6}, nil)
+ if nullable {
+ ids.AppendNull()
+ } else {
+ ids.Append(0)
+ }
+ ids.Append(7)
+
builder.Field(1).(*array.StringBuilder).AppendValues(
+ []string{"keep", "drop", "drop",
"keep", "keep", "drop", "keep", "keep"}, nil)
+ record := builder.NewRecordBatch()
+ builder.Release()
+ defer record.Release()
+
+ fastSet := int64EqualityDeleteSet(1, 2, 5, 0)
+ fastSet.keys[string([]byte{0})] = struct{}{}
+ var key bytes.Buffer
+ encodeArrowValue(&key, record.Column(1), 1)
+ genericSet := &equalityDeleteSet{
+ keys: set[string]{key.String(): {}},
fieldIDs: []int{2}, colNames: []string{"category"},
+ }
+ deleteSets := []*equalityDeleteSet{fastSet,
genericSet}
+ if genericFirst {
+ deleteSets[0], deleteSets[1] =
deleteSets[1], deleteSets[0]
+ }
+ fileSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id",
Type: iceberg.PrimitiveTypes.Int64},
+ iceberg.NestedField{ID: 2, Name:
"category", Type: iceberg.PrimitiveTypes.String},
+ )
+ process, err :=
processEqualityDeletesColumnarForFile(ctx, deleteSets, fileSchema,
"data.parquet")
+ require.NoError(t, err)
+ require.NotNil(t, fastSet.singleInt64)
+ require.Nil(t, genericSet.singleInt64)
+
+ record.Retain()
+ result, err := process(record)
+ require.NoError(t, err)
+ defer result.Release()
+ require.Equal(t, int64(2), result.NumRows())
+ assert.Equal(t, []int64{4, 7},
result.Column(0).(*array.Int64).Int64Values())
+ })
+ }
+ }
+}
+
+func TestProcessEqualityDeletesSingleInt64FallsBackForUnexpectedKeys(t
*testing.T) {
+ for _, key := range []string{"", "\x01\x00",
"\x02\x00\x00\x00\x00\x00\x00\x00\x00"} {
+ t.Run(fmt.Sprintf("key=%x", key), func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ builder := array.NewInt64Builder(mem)
+ builder.AppendValues([]int64{math.MinInt64, -1, 42},
nil)
+ values := builder.NewArray()
+ builder.Release()
+ schema := arrow.NewSchema([]arrow.Field{{Name: "id",
Type: arrow.PrimitiveTypes.Int64}}, nil)
+ record := array.NewRecordBatch(schema,
[]arrow.Array{values}, 3)
+ values.Release()
+ defer record.Release()
+
+ deleteSet := int64EqualityDeleteSet(math.MinInt64, -1)
+ deleteSet.keys[key] = struct{}{}
+ fileSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64},
+ )
+ process, err :=
processEqualityDeletesColumnarForFile(ctx,
+ []*equalityDeleteSet{deleteSet}, fileSchema,
"data.parquet")
+ require.NoError(t, err)
+ require.Nil(t, deleteSet.singleInt64, "unexpected key
formats must use the generic path")
+ record.Retain()
+ result, err := process(record)
+ require.NoError(t, err)
+ defer result.Release()
+ assert.Equal(t, []int64{42},
result.Column(0).(*array.Int64).Int64Values())
+ })
+ }
+}
+
+func TestProcessEqualityDeletesReleasesMaskAfterLaterSetError(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(t.Context(), mem)
+ builder := array.NewInt64Builder(mem)
+ builder.AppendValues([]int64{1, 2}, nil)
+ values := builder.NewArray()
+ builder.Release()
+ schema := arrow.NewSchema([]arrow.Field{{Name: "id", Type:
arrow.PrimitiveTypes.Int64}}, nil)
+ record := array.NewRecordBatch(schema, []arrow.Array{values}, 2)
+ values.Release()
+ defer record.Release()
+
+ fileSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64},
+ iceberg.NestedField{ID: 2, Name: "missing", Type:
iceberg.PrimitiveTypes.String},
+ )
+ process, err := processEqualityDeletesColumnarForFile(ctx,
[]*equalityDeleteSet{
+ int64EqualityDeleteSet(1),
+ {fieldIDs: []int{2}, colNames: []string{"missing"}},
+ }, fileSchema, "data.parquet")
+ require.NoError(t, err)
+
+ record.Retain()
+ result, err := process(record)
+ require.ErrorContains(t, err, "equality field ID 2 (missing) not found
in data.parquet")
+ assert.Nil(t, result)
+}
+
+func TestProcessEqualityDeletesNestedInt64DoesNotBuildUnusedIndex(t
*testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ ctx := compute.WithAllocator(t.Context(), mem)
+
+ builder := array.NewInt64Builder(mem)
+ builder.AppendValues([]int64{123, 456, 789}, nil)
+ child := builder.NewArray()
+ builder.Release()
+ defer child.Release()
+ validity := memory.NewBufferBytes([]byte{0x06})
+ defer validity.Release()
+ parent, err := array.NewStructArrayWithNulls(
+ []arrow.Array{child}, []string{"id"}, validity, 1, 0)
+ require.NoError(t, err)
+
+ arrowSchema := arrow.NewSchema([]arrow.Field{{
+ Name: "person", Type: parent.DataType(), Nullable: true,
+ }}, nil)
+ record := array.NewRecordBatch(arrowSchema, []arrow.Array{parent}, 3)
+ parent.Release()
+
+ fileSchema := iceberg.NewSchema(0, iceberg.NestedField{
+ ID: 2, Name: "person", Type: &iceberg.StructType{FieldList:
[]iceberg.NestedField{
+ {ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64,
Required: true},
+ }},
+ })
+ deleteSet := int64EqualityDeleteSet(456)
+ deleteSet.colNames[0] = "person.id"
+ deleteSet.keys[string([]byte{0})] = struct{}{}
+ process, err := processEqualityDeletesColumnarForFile(ctx,
+ []*equalityDeleteSet{deleteSet}, fileSchema, "nested.parquet")
+ require.NoError(t, err)
+ assert.Nil(t, deleteSet.singleInt64, "nested fields must not allocate
an unused typed index")
+
+ result, err := process(record)
+ require.NoError(t, err)
+ defer result.Release()
+ require.Equal(t, int64(1), result.NumRows())
+ assert.Equal(t, int64(789),
result.Column(0).(*array.Struct).Field(0).(*array.Int64).Value(0))
+}
+
func int64EqualityDeleteSet(values ...int64) *equalityDeleteSet {
keys := make(set[string], len(values))
for _, value := range values {
