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 42246263 perf(arrow/array): validate dictionary indices by valid runs
(#1181)
42246263 is described below
commit 4224626350af091d9e65b2ec59b01e6151488ce5
Author: Minh Vu <[email protected]>
AuthorDate: Thu Aug 27 23:25:22 2026 +0200
perf(arrow/array): validate dictionary indices by valid runs (#1181)
## Summary
- Validate nullable dictionary indices using contiguous valid runs.
- Keep the existing whole-slice min/max path for arrays without nulls.
- Fall back to the whole-slice path for fragmented validity when all
physical values are already in range.
- Ignore out-of-range physical values at null positions.
## Why
The current bounds check scans every physical index, including null
positions. This does unnecessary work for clustered nullable arrays and
can reject garbage payloads that are hidden by the validity bitmap.
The existing TODO points to the set-bit run reader for this case. This
change uses it for up to eight valid runs. More fragmented bitmaps keep
the vectorized whole-slice fast path. If that scan finds an invalid
physical value, valid runs are checked again so null payloads are still
ignored.
## Benchmark
1M int32 indices on an Apple M1 Pro, 300 ms per sample, 6 samples:
| Case | Before | After | Change |
| --- | ---: | ---: | ---: |
| all valid | 88.96 us/op | 92.68 us/op | no significant change |
| 10% valid, clustered | 88.08 us/op | 45.53 us/op | 48.3% faster |
| 50% valid, clustered | 90.51 us/op | 81.28 us/op | 10.2% faster |
| alternating | 86.83 us/op | 88.58 us/op | no significant change |
All cases remain at 0 allocations/op.
## Checks
- `go test ./arrow/...`
- `go vet ./arrow/array`
- `git diff --check`
No public API changes.
Co-authored-by: Matt Topol <[email protected]>
---
arrow/array/dictionary.go | 213 ++++++++++++++++++++++++++++++--
arrow/array/dictionary_internal_test.go | 153 +++++++++++++++++++++++
2 files changed, 356 insertions(+), 10 deletions(-)
diff --git a/arrow/array/dictionary.go b/arrow/array/dictionary.go
index a03fa457..db5c9422 100644
--- a/arrow/array/dictionary.go
+++ b/arrow/array/dictionary.go
@@ -30,6 +30,7 @@ import (
"github.com/apache/arrow-go/v18/arrow/float16"
"github.com/apache/arrow-go/v18/arrow/internal/debug"
"github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/internal/bitutils"
"github.com/apache/arrow-go/v18/internal/hashing"
"github.com/apache/arrow-go/v18/internal/json"
"github.com/apache/arrow-go/v18/internal/utils"
@@ -59,6 +60,72 @@ type Dictionary struct {
dict arrow.Array
}
+const maxDictionaryIndexValidRuns = 8
+
+type dictionaryIndexValidRuns struct {
+ runs [maxDictionaryIndexValidRuns]bitutils.SetBitRun
+ count int
+ fragmented bool
+}
+
+func findDictionaryIndexValidRuns(validBits []byte, offset, length int)
dictionaryIndexValidRuns {
+ var result dictionaryIndexValidRuns
+ reader := bitutils.NewSetBitRunReader(validBits, int64(offset),
int64(length))
+ for i := range result.runs {
+ run := reader.NextRun()
+ if run.AtEnd() {
+ result.count = i
+ return result
+ }
+ result.runs[i] = run
+ }
+
+ result.count = len(result.runs)
+ result.fragmented = !reader.NextRun().AtEnd()
+ return result
+}
+
+func getMinMaxRuns[T arrow.IntType | arrow.UintType](
+ values []T, offset int, runs []bitutils.SetBitRun, getMinMax func([]T)
(T, T),
+) (min, max T, hasValues bool) {
+ for _, run := range runs {
+ start := offset + int(run.Pos)
+ runMin, runMax := getMinMax(values[start :
start+int(run.Length)])
+ if !hasValues {
+ min, max, hasValues = runMin, runMax, true
+ continue
+ }
+ if runMin < min {
+ min = runMin
+ }
+ if runMax > max {
+ max = runMax
+ }
+ }
+ return
+}
+
+func getMinMaxValid[T arrow.IntType | arrow.UintType](
+ values []T, validBits []byte, offset, length int, getMinMax func([]T)
(T, T),
+) (min, max T, hasValues bool) {
+ visitRun := func(pos, runLength int64) {
+ start := offset + int(pos)
+ runMin, runMax := getMinMax(values[start :
start+int(runLength)])
+ if !hasValues {
+ min, max, hasValues = runMin, runMax, true
+ return
+ }
+ if runMin < min {
+ min = runMin
+ }
+ if runMax > max {
+ max = runMax
+ }
+ }
+ bitutils.VisitSetBitRunsNoErr(validBits, int64(offset), int64(length),
visitRun)
+ return
+}
+
// NewDictionaryArray constructs a dictionary array with the provided indices
// and dictionary using the given type.
func NewDictionaryArray(typ arrow.DataType, indices, dict arrow.Array)
*Dictionary {
@@ -103,55 +170,181 @@ func checkIndexBounds(indices *Data, upperlimit uint64)
error {
start := indices.offset
end := indices.offset + indices.length
- // TODO(ARROW-15950): lift BitSetRunReader from parquet to utils
- // and use it here for performance improvement.
+ var validBits []byte
+ var validRuns dictionaryIndexValidRuns
+ if indices.buffers[0] != nil && indices.nulls != 0 {
+ validBits = indices.buffers[0].Bytes()
+ // Run-based min/max is useful for clustered validity.
Fragmented
+ // bitmaps retain the whole-slice fast path below.
+ validRuns = findDictionaryIndexValidRuns(validBits, start,
indices.length)
+ }
switch indices.dtype.ID() {
case arrow.INT8:
data :=
arrow.Int8Traits.CastFromBytes(indices.buffers[1].Bytes())
- min, max := utils.GetMinMaxInt8(data[start:end])
+ min, max := int8(0), int8(0)
+ hasValues := true
+ if validBits == nil {
+ min, max = utils.GetMinMaxInt8(data[start:end])
+ } else if !validRuns.fragmented {
+ min, max, hasValues = getMinMaxRuns(data, start,
validRuns.runs[:validRuns.count], utils.GetMinMaxInt8)
+ } else {
+ min, max = utils.GetMinMaxInt8(data[start:end])
+ if min >= 0 && uint64(max) < upperlimit {
+ return nil
+ }
+ min, max, hasValues = getMinMaxValid(data, validBits,
start, indices.length, utils.GetMinMaxInt8)
+ }
+ if !hasValues {
+ return nil
+ }
if min < 0 || uint64(max) >= upperlimit {
return fmt.Errorf("contains out of bounds index: min:
%d, max: %d", min, max)
}
case arrow.UINT8:
data :=
arrow.Uint8Traits.CastFromBytes(indices.buffers[1].Bytes())
- _, max := utils.GetMinMaxUint8(data[start:end])
+ max := uint8(0)
+ hasValues := true
+ if validBits == nil {
+ _, max = utils.GetMinMaxUint8(data[start:end])
+ } else if !validRuns.fragmented {
+ _, max, hasValues = getMinMaxRuns(data, start,
validRuns.runs[:validRuns.count], utils.GetMinMaxUint8)
+ } else {
+ _, max = utils.GetMinMaxUint8(data[start:end])
+ if uint64(max) < upperlimit {
+ return nil
+ }
+ _, max, hasValues = getMinMaxValid(data, validBits,
start, indices.length, utils.GetMinMaxUint8)
+ }
+ if !hasValues {
+ return nil
+ }
if max >= uint8(upperlimit) {
return fmt.Errorf("contains out of bounds index: max:
%d", max)
}
case arrow.INT16:
data :=
arrow.Int16Traits.CastFromBytes(indices.buffers[1].Bytes())
- min, max := utils.GetMinMaxInt16(data[start:end])
+ min, max := int16(0), int16(0)
+ hasValues := true
+ if validBits == nil {
+ min, max = utils.GetMinMaxInt16(data[start:end])
+ } else if !validRuns.fragmented {
+ min, max, hasValues = getMinMaxRuns(data, start,
validRuns.runs[:validRuns.count], utils.GetMinMaxInt16)
+ } else {
+ min, max = utils.GetMinMaxInt16(data[start:end])
+ if min >= 0 && uint64(max) < upperlimit {
+ return nil
+ }
+ min, max, hasValues = getMinMaxValid(data, validBits,
start, indices.length, utils.GetMinMaxInt16)
+ }
+ if !hasValues {
+ return nil
+ }
if min < 0 || uint64(max) >= upperlimit {
return fmt.Errorf("contains out of bounds index: min:
%d, max: %d", min, max)
}
case arrow.UINT16:
data :=
arrow.Uint16Traits.CastFromBytes(indices.buffers[1].Bytes())
- _, max := utils.GetMinMaxUint16(data[start:end])
+ max := uint16(0)
+ hasValues := true
+ if validBits == nil {
+ _, max = utils.GetMinMaxUint16(data[start:end])
+ } else if !validRuns.fragmented {
+ _, max, hasValues = getMinMaxRuns(data, start,
validRuns.runs[:validRuns.count], utils.GetMinMaxUint16)
+ } else {
+ _, max = utils.GetMinMaxUint16(data[start:end])
+ if uint64(max) < upperlimit {
+ return nil
+ }
+ _, max, hasValues = getMinMaxValid(data, validBits,
start, indices.length, utils.GetMinMaxUint16)
+ }
+ if !hasValues {
+ return nil
+ }
if max >= uint16(upperlimit) {
return fmt.Errorf("contains out of bounds index: max:
%d", max)
}
case arrow.INT32:
data :=
arrow.Int32Traits.CastFromBytes(indices.buffers[1].Bytes())
- min, max := utils.GetMinMaxInt32(data[start:end])
+ min, max := int32(0), int32(0)
+ hasValues := true
+ if validBits == nil {
+ min, max = utils.GetMinMaxInt32(data[start:end])
+ } else if !validRuns.fragmented {
+ min, max, hasValues = getMinMaxRuns(data, start,
validRuns.runs[:validRuns.count], utils.GetMinMaxInt32)
+ } else {
+ min, max = utils.GetMinMaxInt32(data[start:end])
+ if min >= 0 && uint64(max) < upperlimit {
+ return nil
+ }
+ min, max, hasValues = getMinMaxValid(data, validBits,
start, indices.length, utils.GetMinMaxInt32)
+ }
+ if !hasValues {
+ return nil
+ }
if min < 0 || uint64(max) >= upperlimit {
return fmt.Errorf("contains out of bounds index: min:
%d, max: %d", min, max)
}
case arrow.UINT32:
data :=
arrow.Uint32Traits.CastFromBytes(indices.buffers[1].Bytes())
- _, max := utils.GetMinMaxUint32(data[start:end])
+ max := uint32(0)
+ hasValues := true
+ if validBits == nil {
+ _, max = utils.GetMinMaxUint32(data[start:end])
+ } else if !validRuns.fragmented {
+ _, max, hasValues = getMinMaxRuns(data, start,
validRuns.runs[:validRuns.count], utils.GetMinMaxUint32)
+ } else {
+ _, max = utils.GetMinMaxUint32(data[start:end])
+ if uint64(max) < upperlimit {
+ return nil
+ }
+ _, max, hasValues = getMinMaxValid(data, validBits,
start, indices.length, utils.GetMinMaxUint32)
+ }
+ if !hasValues {
+ return nil
+ }
if max >= uint32(upperlimit) {
return fmt.Errorf("contains out of bounds index: max:
%d", max)
}
case arrow.INT64:
data :=
arrow.Int64Traits.CastFromBytes(indices.buffers[1].Bytes())
- min, max := utils.GetMinMaxInt64(data[start:end])
+ min, max := int64(0), int64(0)
+ hasValues := true
+ if validBits == nil {
+ min, max = utils.GetMinMaxInt64(data[start:end])
+ } else if !validRuns.fragmented {
+ min, max, hasValues = getMinMaxRuns(data, start,
validRuns.runs[:validRuns.count], utils.GetMinMaxInt64)
+ } else {
+ min, max = utils.GetMinMaxInt64(data[start:end])
+ if min >= 0 && uint64(max) < upperlimit {
+ return nil
+ }
+ min, max, hasValues = getMinMaxValid(data, validBits,
start, indices.length, utils.GetMinMaxInt64)
+ }
+ if !hasValues {
+ return nil
+ }
if min < 0 || uint64(max) >= upperlimit {
return fmt.Errorf("contains out of bounds index: min:
%d, max: %d", min, max)
}
case arrow.UINT64:
data :=
arrow.Uint64Traits.CastFromBytes(indices.buffers[1].Bytes())
- _, max := utils.GetMinMaxUint64(data[indices.offset :
indices.offset+indices.length])
+ max := uint64(0)
+ hasValues := true
+ if validBits == nil {
+ _, max = utils.GetMinMaxUint64(data[start:end])
+ } else if !validRuns.fragmented {
+ _, max, hasValues = getMinMaxRuns(data, start,
validRuns.runs[:validRuns.count], utils.GetMinMaxUint64)
+ } else {
+ _, max = utils.GetMinMaxUint64(data[start:end])
+ if max < upperlimit {
+ return nil
+ }
+ _, max, hasValues = getMinMaxValid(data, validBits,
start, indices.length, utils.GetMinMaxUint64)
+ }
+ if !hasValues {
+ return nil
+ }
if max >= upperlimit {
return fmt.Errorf("contains out of bounds value: max:
%d", max)
}
diff --git a/arrow/array/dictionary_internal_test.go
b/arrow/array/dictionary_internal_test.go
index 5c51a18e..91f622fd 100644
--- a/arrow/array/dictionary_internal_test.go
+++ b/arrow/array/dictionary_internal_test.go
@@ -21,6 +21,7 @@ import (
"testing"
"github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/bitutil"
"github.com/apache/arrow-go/v18/arrow/memory"
"github.com/stretchr/testify/require"
)
@@ -53,3 +54,155 @@ func TestCheckIndexBoundsAllowsSignedIndexAtTypeLimit(t
*testing.T) {
})
}
}
+
+func TestCheckIndexBoundsIgnoresNullValues(t *testing.T) {
+ tests := []struct {
+ name string
+ indexType arrow.DataType
+ indexBytes []byte
+ }{
+ {"int8", arrow.PrimitiveTypes.Int8,
arrow.Int8Traits.CastToBytes([]int8{99, 99, 0, -1, 2, 99})},
+ {"uint8", arrow.PrimitiveTypes.Uint8,
arrow.Uint8Traits.CastToBytes([]uint8{99, 99, 0, math.MaxUint8, 2, 99})},
+ {"int16", arrow.PrimitiveTypes.Int16,
arrow.Int16Traits.CastToBytes([]int16{99, 99, 0, -1, 2, 99})},
+ {"uint16", arrow.PrimitiveTypes.Uint16,
arrow.Uint16Traits.CastToBytes([]uint16{99, 99, 0, math.MaxUint16, 2, 99})},
+ {"int32", arrow.PrimitiveTypes.Int32,
arrow.Int32Traits.CastToBytes([]int32{99, 99, 0, -1, 2, 99})},
+ {"uint32", arrow.PrimitiveTypes.Uint32,
arrow.Uint32Traits.CastToBytes([]uint32{99, 99, 0, math.MaxUint32, 2, 99})},
+ {"int64", arrow.PrimitiveTypes.Int64,
arrow.Int64Traits.CastToBytes([]int64{99, 99, 0, -1, 2, 99})},
+ {"uint64", arrow.PrimitiveTypes.Uint64,
arrow.Uint64Traits.CastToBytes([]uint64{99, 99, 0, math.MaxUint64, 2, 99})},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ validity := make([]byte, bitutil.BytesForBits(6))
+ bitutil.SetBit(validity, 2)
+ bitutil.SetBit(validity, 4)
+ validityBuffer := memory.NewBufferBytes(validity)
+ valuesBuffer := memory.NewBufferBytes(tt.indexBytes)
+ indices := NewData(
+ tt.indexType,
+ 3,
+ []*memory.Buffer{validityBuffer, valuesBuffer},
+ nil,
+ 1,
+ 2,
+ )
+ validityBuffer.Release()
+ valuesBuffer.Release()
+ defer indices.Release()
+
+ require.NoError(t, checkIndexBounds(indices, 3))
+ })
+ }
+}
+
+func TestCheckIndexBoundsAllowsAllNullIndices(t *testing.T) {
+ validityBuffer := memory.NewBufferBytes(make([]byte,
bitutil.BytesForBits(3)))
+ valuesBuffer :=
memory.NewBufferBytes(arrow.Int32Traits.CastToBytes([]int32{-1, -1, -1}))
+ indices := NewData(
+ arrow.PrimitiveTypes.Int32,
+ 3,
+ []*memory.Buffer{validityBuffer, valuesBuffer},
+ nil,
+ 3,
+ 0,
+ )
+ validityBuffer.Release()
+ valuesBuffer.Release()
+ defer indices.Release()
+
+ require.NoError(t, checkIndexBounds(indices, 0))
+}
+
+func TestCheckIndexBoundsIgnoresNullValuesWithFragmentedValidity(t *testing.T)
{
+ const length = 256
+ values := make([]int32, length)
+ validity := make([]byte, bitutil.BytesForBits(length))
+ for i := range values {
+ if i%2 == 0 {
+ values[i] = 1
+ bitutil.SetBit(validity, i)
+ } else {
+ values[i] = -1
+ }
+ }
+
+ validityBuffer := memory.NewBufferBytes(validity)
+ valuesBuffer :=
memory.NewBufferBytes(arrow.Int32Traits.CastToBytes(values))
+ indices := NewData(
+ arrow.PrimitiveTypes.Int32,
+ length,
+ []*memory.Buffer{validityBuffer, valuesBuffer},
+ nil,
+ length/2,
+ 0,
+ )
+ validityBuffer.Release()
+ valuesBuffer.Release()
+ defer indices.Release()
+
+ require.NoError(t, checkIndexBounds(indices, 2))
+
+ values[0] = -1
+ require.Error(t, checkIndexBounds(indices, 2))
+}
+
+func BenchmarkCheckIndexBounds(b *testing.B) {
+ const length = 1 << 20
+
+ values := make([]int32, length)
+ for i := range values {
+ values[i] = int32(i % 1024)
+ }
+ valuesBuffer :=
memory.NewBufferBytes(arrow.Int32Traits.CastToBytes(values))
+ defer valuesBuffer.Release()
+
+ benchmarks := []struct {
+ name string
+ isValid func(int) bool
+ }{
+ {name: "all_valid", isValid: func(int) bool { return true }},
+ {name: "clustered_10_percent_valid", isValid: func(i int) bool
{ return i < length/10 }},
+ {name: "clustered_50_percent_valid", isValid: func(i int) bool
{ return i < length/2 }},
+ {name: "strided_10_percent_valid", isValid: func(i int) bool {
return i%10 == 0 }},
+ {name: "strided_90_percent_valid", isValid: func(i int) bool {
return i%10 != 0 }},
+ {name: "alternating", isValid: func(i int) bool { return i%2 ==
0 }},
+ }
+
+ for _, benchmark := range benchmarks {
+ b.Run(benchmark.name, func(b *testing.B) {
+ var validityBuffer *memory.Buffer
+ nulls := 0
+ if benchmark.name != "all_valid" {
+ validity := make([]byte,
bitutil.BytesForBits(length))
+ for i := 0; i < length; i++ {
+ if benchmark.isValid(i) {
+ bitutil.SetBit(validity, i)
+ } else {
+ nulls++
+ }
+ }
+ validityBuffer = memory.NewBufferBytes(validity)
+ defer validityBuffer.Release()
+ }
+
+ indices := NewData(
+ arrow.PrimitiveTypes.Int32,
+ length,
+ []*memory.Buffer{validityBuffer, valuesBuffer},
+ nil,
+ nulls,
+ 0,
+ )
+ defer indices.Release()
+
+ b.ReportAllocs()
+ b.SetBytes(length *
int64(arrow.Int32Traits.BytesRequired(1)))
+ b.ResetTimer()
+ for range b.N {
+ if err := checkIndexBounds(indices, 1024); err
!= nil {
+ b.Fatal(err)
+ }
+ }
+ })
+ }
+}
