This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-go.git
The following commit(s) were added to refs/heads/main by this push:
new 11dd9545 perf(parquet): copy DELTA_LENGTH Arrow payloads by valid runs
(#1392)
11dd9545 is described below
commit 11dd954594c6c176595d8649ed6c3ccfeaf072a2
Author: Minh Vu <[email protected]>
AuthorDate: Wed Oct 21 22:05:39 2026 +0200
perf(parquet): copy DELTA_LENGTH Arrow payloads by valid runs (#1392)
### What
- Copy DELTA_LENGTH_BYTE_ARRAY payload bytes once per contiguous valid
run.
### Why
- Writing each short value separately adds unnecessary work.
### Implementation
- Use one bulk copy for null-free arrays and one copy per valid bitmap
run.
- Keep the length encoding unchanged.
### Benchmark
`BenchmarkDeltaLengthArrowContiguousRuns`, 65,536 null-free values; Go
1.26.3, Apple M1 Pro, median of 5 runs.
| Metric | Before | After | Change |
|---|---:|---:|---:|
| 4-byte values (µs/op) | 819 | 487 | 1.68x faster |
| 16-byte values (µs/op) | 880 | 561 | 1.57x faster |
| Allocations (allocs/op) | 2,054 | 2,054 | No change |
This measures the DELTA_LENGTH encoder path, not full Parquet file
encoding.
---
.../internal/encoding/byte_array_arrow_encoder.go | 10 +-
.../encoding/byte_array_arrow_encoder_test.go | 117 +++++++++++++++++++++
2 files changed, 121 insertions(+), 6 deletions(-)
diff --git a/parquet/internal/encoding/byte_array_arrow_encoder.go
b/parquet/internal/encoding/byte_array_arrow_encoder.go
index c9ec4b8f..fa0ec57c 100644
--- a/parquet/internal/encoding/byte_array_arrow_encoder.go
+++ b/parquet/internal/encoding/byte_array_arrow_encoder.go
@@ -102,9 +102,8 @@ func putArrowDeltaLength[T arrowByteArrayOffset](enc
*DeltaLengthByteArrayEncode
}
enc.sink.Reserve(totalLen)
- for i := 0; i < len(offsets)-1; i++ {
-
enc.sink.UnsafeWrite(values[int(offsets[i]):int(offsets[i+1])])
- }
+ // Arrow offsets describe one contiguous byte range when there
are no nulls.
+
enc.sink.UnsafeWrite(values[int(offsets[0]):int(offsets[len(offsets)-1])])
return
}
@@ -130,9 +129,8 @@ func putArrowDeltaLength[T arrowByteArrayOffset](enc
*DeltaLengthByteArrayEncode
enc.sink.Reserve(totalLen)
bitutils.VisitSetBitRunsNoErr(validBits, validBitsOffset,
int64(len(offsets)-1), func(pos, length int64) {
- for i := pos; i < pos+length; i++ {
-
enc.sink.UnsafeWrite(values[int(offsets[i]):int(offsets[i+1])])
- }
+ // A valid run is contiguous; the gaps between runs can hold
null payloads.
+
enc.sink.UnsafeWrite(values[int(offsets[pos]):int(offsets[pos+length])])
})
}
diff --git a/parquet/internal/encoding/byte_array_arrow_encoder_test.go
b/parquet/internal/encoding/byte_array_arrow_encoder_test.go
index 30a7e528..16f76ed1 100644
--- a/parquet/internal/encoding/byte_array_arrow_encoder_test.go
+++ b/parquet/internal/encoding/byte_array_arrow_encoder_test.go
@@ -270,6 +270,123 @@ func benchmarkPlainByteArrayPutArrow[T
arrowByteArrayOffset](b *testing.B) {
}
}
+// Exercise direct Arrow buffer writes with non-zero offsets, batches that
+// straddle a DELTA miniblock, and runs separated by non-empty null payloads.
+func TestDeltaLengthArrowContiguousRuns(t *testing.T) {
+ t.Run("int32", testDeltaLengthArrowContiguousRuns[int32])
+ t.Run("int64", testDeltaLengthArrowContiguousRuns[int64])
+}
+
+func testDeltaLengthArrowContiguousRuns[T arrowByteArrayOffset](t *testing.T) {
+ const n = 2*deltaByteArrayBatchSize + 17
+ data := []byte("discard before first value")
+ offsets := make([]T, n+1)
+ values := make([]parquet.ByteArray, n)
+ for i := range values {
+ offsets[i] = T(len(data))
+ values[i] = make([]byte, (i*17+i/13)%19)
+ for j := range values[i] {
+ values[i][j] = byte(i*7 + j)
+ }
+ data = append(data, values[i]...)
+ }
+ offsets[n] = T(len(data))
+
+ for _, span := range []struct {
+ name string
+ begin, end int
+ }{
+ {"whole", 0, n},
+ {"sliced", 11, n - 9},
+ {"single", 47, 48},
+ {"empty", 47, 47},
+ } {
+ for _, pattern := range []string{"all", "all-valid-bitmap",
"none", "alternating", "long-runs"} {
+ t.Run(span.name+"/"+pattern, func(t *testing.T) {
+ var bits []byte
+ if pattern != "all" {
+ bits = make([]byte,
bitutil.BytesForBits(int64(span.end-span.begin+3)))
+ }
+ want := []parquet.ByteArray{[]byte("before")}
+ for i := span.begin; i < span.end; i++ {
+ valid := pattern == "all" || pattern ==
"all-valid-bitmap" ||
+ (pattern == "alternating" &&
i%2 == 0) ||
+ (pattern == "long-runs" &&
(i/31)%2 == 0)
+ if valid {
+ want = append(want, values[i])
+ if bits != nil {
+ bitutil.SetBit(bits,
3+i-span.begin)
+ }
+ }
+ }
+ want = append(want, []byte("after"))
+
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+ enc := NewEncoder(parquet.Types.ByteArray,
parquet.Encodings.DeltaLengthByteArray,
+ false, nil,
mem).(*DeltaLengthByteArrayEncoder)
+ defer enc.Release()
+ enc.Put([]parquet.ByteArray{[]byte("before")})
+ middle := span.begin + (span.end-span.begin)/2
+ for _, part := range [][2]int{{span.begin,
middle}, {middle, span.end}} {
+ putArrowDeltaLength(enc, data,
offsets[part[0]:part[1]+1], bits,
+ int64(3+part[0]-span.begin))
+ }
+ enc.Put([]parquet.ByteArray{[]byte("after")})
+ got, err := enc.FlushValues()
+ require.NoError(t, err)
+ defer got.Release()
+ require.Equal(t, encodedByteArrays(t,
parquet.Encodings.DeltaLengthByteArray,
+ want, false, nil, 0), got.Bytes())
+ })
+ }
+ }
+}
+
+func BenchmarkDeltaLengthArrowContiguousRuns(b *testing.B) {
+ const n = 64 * 1024
+ for _, width := range []int{0, 4, 16, 64} {
+ for _, pattern := range []string{"all", "long-runs",
"alternating"} {
+ b.Run(fmt.Sprintf("bytes=%d/%s", width, pattern),
func(b *testing.B) {
+ data := make([]byte, 7+n*width)
+ offsets := make([]int32, n+1)
+ for i := range offsets {
+ offsets[i] = int32(7 + i*width)
+ }
+ var bits []byte
+ if pattern != "all" {
+ bits = make([]byte,
bitutil.BytesForBits(n+3))
+ for i := 0; i < n; i++ {
+ valid := pattern ==
"alternating" && i%2 != 0 ||
+ pattern == "long-runs"
&& (i/128)%2 == 0
+ if valid {
+ bitutil.SetBit(bits,
i+3)
+ }
+ }
+ }
+
+ enc := NewEncoder(parquet.Types.ByteArray,
parquet.Encodings.DeltaLengthByteArray,
+ false, nil,
memory.DefaultAllocator).(*DeltaLengthByteArrayEncoder)
+ defer enc.Release()
+ b.SetBytes(int64(n * width))
+ b.ReportAllocs()
+ for b.Loop() {
+ if bits == nil {
+ enc.PutArrow(data, offsets)
+ } else {
+ enc.PutArrowSpaced(data,
offsets, bits, 3)
+ }
+ buf, err := enc.FlushValues()
+ if err != nil {
+ b.Fatal(err)
+ }
+ buf.Release()
+ }
+ })
+ }
+ }
+}
+
func TestDeltaByteArrayArrowEncoderPreservesStateAcrossBatches(t *testing.T) {
values, data, offsets32, offsets64 := arrowByteArrayInput()
