laskoviymishka commented on code in PR #1900:
URL: https://github.com/apache/iceberg-go/pull/1900#discussion_r3874730720
##########
table/equality_delete_reader_bench_test.go:
##########
@@ -242,10 +242,60 @@ func buildBenchDeleteSetString(numDeletes int)
*equalityDeleteSet {
}
}
+func buildBenchDeleteSetIntNoMatch(numDeletes int) *equalityDeleteSet {
+ keys := make(set[string])
+ var buf bytes.Buffer
+
+ for i := range numDeletes {
+ buf.Reset()
+ buf.WriteByte(1)
+ _ = binary.Write(&buf, binary.BigEndian, int64(i*3))
+ buf.WriteByte(1)
+ _ = binary.Write(&buf, binary.BigEndian, int64((i*3+1)%100))
+ keys[buf.String()] = struct{}{}
+ }
+
+ return &equalityDeleteSet{
+ keys: keys,
+ fieldIDs: []int{1, 2},
+ colNames: []string{"id", "category"},
+ }
+}
+
+func buildBenchDeleteSetStringNoMatch(numDeletes int) *equalityDeleteSet {
+ keys := make(set[string])
+ var buf bytes.Buffer
+
+ for i := range numDeletes {
+ buf.Reset()
+ buf.WriteByte(1)
+ _ = binary.Write(&buf, binary.BigEndian, int64(i*3))
+ buf.WriteByte(1)
+ name := fmt.Sprintf("user-%08d", i*3+1)
+ _ = binary.Write(&buf, binary.BigEndian, int32(len(name)))
+ buf.WriteString(name)
+ keys[buf.String()] = struct{}{}
+ }
+
+ return &equalityDeleteSet{
+ keys: keys,
+ fieldIDs: []int{1, 2},
+ colNames: []string{"id", "name"},
+ }
+}
+
func BenchmarkProcessEqualityDeletesInt(b *testing.B) {
benchEqDeletes(b, buildBenchRecordInt, buildBenchDeleteSetInt)
}
func BenchmarkProcessEqualityDeletesString(b *testing.B) {
benchEqDeletes(b, buildBenchRecordString, buildBenchDeleteSetString)
}
+
+func BenchmarkProcessEqualityDeletesNoMatchInt(b *testing.B) {
+ benchEqDeletes(b, buildBenchRecordInt, buildBenchDeleteSetIntNoMatch)
Review Comment:
I think these two benchmarks are measuring the wrong function.
`benchEqDeletes` routes through `processEqualityDeletes` →
`processEqualityDeletesColumnar`, but the lazy-alloc optimization this PR adds
lives only in `processEqualityDeletesColumnarForFile`, which this path never
calls. So both no-match benchmarks are timing the unchanged eager-allocation
code, and the numbers in the PR body don't actually demonstrate the improvement.
I'd add a `benchEqDeletesForFile` helper that calls
`processEqualityDeletesColumnarForFile(ctx, delSets, fileSchema,
"bench.parquet")` and point the two no-match benchmarks at it. Then the
before/after should actually show the skip. wdyt?
##########
table/equality_delete_reader_internal_test.go:
##########
@@ -481,6 +481,43 @@ func TestProcessEqualityDeletesUsesStructuralFieldPaths(t
*testing.T) {
result.Release()
}
+func TestProcessEqualityDeletesReturnsOriginalBatchWhenNoRowsMatch(t
*testing.T) {
+ builder := array.NewInt64Builder(memory.DefaultAllocator)
+ builder.AppendValues([]int64{1, 2}, nil)
+ values := builder.NewArray()
+ builder.Release()
+
+ unmatchedBuilder := array.NewInt64Builder(memory.DefaultAllocator)
+ unmatchedBuilder.Append(3)
+ unmatched := unmatchedBuilder.NewArray()
+ unmatchedBuilder.Release()
+ var unmatchedKey bytes.Buffer
+ encodeArrowValue(&unmatchedKey, unmatched, 0)
+ unmatched.Release()
+
+ schema := arrow.NewSchema([]arrow.Field{{
+ Name: "id", Type: arrow.PrimitiveTypes.Int64,
+ }}, nil)
+ record := array.NewRecordBatch(schema, []arrow.Array{values}, 2)
+ values.Release()
+
+ fileSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64},
+ )
+ process, err :=
processEqualityDeletesColumnarForFile(context.Background(),
[]*equalityDeleteSet{{
+ keys: set[string]{unmatchedKey.String(): {}},
Review Comment:
This only exercises a single delete set. The interesting new behavior is
across sets: the `maskBytes != nil` guard in the row loop and the mid-loop lazy
allocation only fire once there are two or more sets. Two cases worth covering:
set 0 matches (allocates the mask) and set 1 then skips an already-cleared row,
and the inverse where set 0 matches nothing (mask stays nil) and set 1 matches
and allocates mid-loop. I'd add a two-set test asserting both rows drop out.
wdyt?
##########
table/equality_delete_reader_internal_test.go:
##########
@@ -481,6 +481,43 @@ func TestProcessEqualityDeletesUsesStructuralFieldPaths(t
*testing.T) {
result.Release()
}
+func TestProcessEqualityDeletesReturnsOriginalBatchWhenNoRowsMatch(t
*testing.T) {
+ builder := array.NewInt64Builder(memory.DefaultAllocator)
+ builder.AppendValues([]int64{1, 2}, nil)
+ values := builder.NewArray()
+ builder.Release()
+
+ unmatchedBuilder := array.NewInt64Builder(memory.DefaultAllocator)
+ unmatchedBuilder.Append(3)
+ unmatched := unmatchedBuilder.NewArray()
+ unmatchedBuilder.Release()
+ var unmatchedKey bytes.Buffer
+ encodeArrowValue(&unmatchedKey, unmatched, 0)
+ unmatched.Release()
+
+ schema := arrow.NewSchema([]arrow.Field{{
+ Name: "id", Type: arrow.PrimitiveTypes.Int64,
+ }}, nil)
+ record := array.NewRecordBatch(schema, []arrow.Array{values}, 2)
+ values.Release()
+
+ fileSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64},
+ )
+ process, err :=
processEqualityDeletesColumnarForFile(context.Background(),
[]*equalityDeleteSet{{
Review Comment:
Nit: the rest of the tests in this file pass `t.Context()`. I'd use it here
too for consistency; it also gets cancelled when the test finishes, unlike
`context.Background()`.
##########
table/equality_delete_reader_internal_test.go:
##########
@@ -481,6 +481,43 @@ func TestProcessEqualityDeletesUsesStructuralFieldPaths(t
*testing.T) {
result.Release()
}
+func TestProcessEqualityDeletesReturnsOriginalBatchWhenNoRowsMatch(t
*testing.T) {
+ builder := array.NewInt64Builder(memory.DefaultAllocator)
Review Comment:
This test validates the no-match return path, which does `r.Retain()`
balanced by `defer r.Release()`, but with `memory.DefaultAllocator` a refcount
imbalance there goes completely undetected. The other tests in this file use
`memory.NewCheckedAllocator(memory.DefaultAllocator)` with `defer
mem.AssertSize(t, 0)` for exactly this reason. Since leak-safety on the
retain/release dance is the whole point of this test, I'd switch it to the
checked allocator.
##########
table/equality_delete_reader_bench_test.go:
##########
@@ -242,10 +242,60 @@ func buildBenchDeleteSetString(numDeletes int)
*equalityDeleteSet {
}
}
+func buildBenchDeleteSetIntNoMatch(numDeletes int) *equalityDeleteSet {
+ keys := make(set[string])
+ var buf bytes.Buffer
+
+ for i := range numDeletes {
+ buf.Reset()
+ buf.WriteByte(1)
+ _ = binary.Write(&buf, binary.BigEndian, int64(i*3))
+ buf.WriteByte(1)
+ _ = binary.Write(&buf, binary.BigEndian, int64((i*3+1)%100))
+ keys[buf.String()] = struct{}{}
+ }
+
+ return &equalityDeleteSet{
+ keys: keys,
+ fieldIDs: []int{1, 2},
+ colNames: []string{"id", "category"},
Review Comment:
These `colNames` (`"id"`/`"category"`) don't line up with the field names
`benchEqDeletes` puts in `fileSchema` (`"first"`/`"second"`). Harmless since
resolution is by field ID, and the existing benchmarks already do this, but a
mismatched name would show up in any field-resolution error message, so while
we're here I'd align them.
##########
table/equality_delete_reader_internal_test.go:
##########
@@ -481,6 +481,43 @@ func TestProcessEqualityDeletesUsesStructuralFieldPaths(t
*testing.T) {
result.Release()
}
+func TestProcessEqualityDeletesReturnsOriginalBatchWhenNoRowsMatch(t
*testing.T) {
+ builder := array.NewInt64Builder(memory.DefaultAllocator)
+ builder.AppendValues([]int64{1, 2}, nil)
+ values := builder.NewArray()
+ builder.Release()
+
+ unmatchedBuilder := array.NewInt64Builder(memory.DefaultAllocator)
Review Comment:
Minor: this builds a whole Int64 array just to encode the key for
`int64(3)`, then releases it. `buildBenchDeleteSetInt` writes the key bytes
directly. That said, going through `encodeArrowValue` keeps the key
guaranteed-consistent with what the reader produces, so I'm honestly fine
either way, just flagging the extra ceremony.
##########
table/equality_delete_reader_bench_test.go:
##########
@@ -242,10 +242,60 @@ func buildBenchDeleteSetString(numDeletes int)
*equalityDeleteSet {
}
}
+func buildBenchDeleteSetIntNoMatch(numDeletes int) *equalityDeleteSet {
+ keys := make(set[string])
+ var buf bytes.Buffer
+
+ for i := range numDeletes {
+ buf.Reset()
+ buf.WriteByte(1)
+ _ = binary.Write(&buf, binary.BigEndian, int64(i*3))
Review Comment:
Tiny style thing: the existing `buildBenchDeleteSet*` helpers call
`binary.Write` bare, and these new ones assign to `_`. Writing to a
`bytes.Buffer` never errors so both are safe, it's just now split within one
file. I'd match one way or the other (errcheck prefers the `_ =`, so probably
add it to the existing ones).
##########
table/equality_delete_reader.go:
##########
@@ -813,21 +811,40 @@ func processEqualityDeletesColumnarForFile(ctx
context.Context, eqDeleteSets []*
enc(&keyBuf, row)
}
- if _, deleted :=
eqDel.keys[bufString(&keyBuf)]; deleted {
- bitutil.ClearBit(maskBytes, row)
+ if _, deleted :=
eqDel.keys[bufString(&keyBuf)]; !deleted {
+ continue
}
+
+ if maskBuf == nil {
+ maskBuf = memory.NewResizableBuffer(mem)
+
maskBuf.Resize(int(bitutil.BytesForBits(int64(numRows))))
+ maskBytes = maskBuf.Bytes()
+
+ for i := range maskBytes {
+ maskBytes[i] = 0xFF
+ }
+ }
+
+ bitutil.ClearBit(maskBytes, row)
}
}
+ if maskBuf == nil {
+ r.Retain()
+
+ return r, nil
+ }
+
mask := array.NewBooleanData(array.NewData(
arrow.FixedWidthTypes.Boolean, numRows,
[]*memory.Buffer{nil, maskBuf}, nil, 0, 0))
- defer mask.Release()
filtered, err := compute.Filter(ctx,
compute.NewDatumWithoutOwning(r),
compute.NewDatumWithoutOwning(mask),
*compute.DefaultFilterOptions())
+ mask.Release()
Review Comment:
Small thing, and not a live bug: the old code had `defer mask.Release()` /
`defer maskBuf.Release()`, so both got freed even if `compute.Filter` panicked.
Now they're explicit calls right before the err check.
Arrow compute is synchronous and won't panic in practice, so this is fine
today, but the defers gave that safety for free, and they'd also protect
against a future failable call sneaking in between the release and the return.
I'd lean toward keeping them as defers (register right after `maskBuf` is
allocated and after `mask` is built). 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]
