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 7c210e36 perf(parquet): decode nullable plain values into valid runs
(#1314)
7c210e36 is described below
commit 7c210e364d8b54ff1328acdd7c6edff2a46bd720
Author: Minh Vu <[email protected]>
AuthorDate: Tue Sep 15 21:14:27 2026 +0200
perf(parquet): decode nullable plain values into valid runs (#1314)
## What does this change?
- `PlainDecoder.DecodeSpaced` used to decode all non-null values into a
dense prefix first, then move them into their bitmap positions.
- For simple validity patterns, decode each valid run directly into its
final output range.
- Keep the existing dense plus spaced expansion path for highly
fragmented validity.
- Use the reverse bitmap reader so trailing-null pages stop after
finding the dense prefix.
- Add edge-case tests and a focused benchmark.
## Benchmark
Apple M1 Pro, Go 1.26.3, 65,536 `int32` values, 8 runs, 300ms per
sample:
| Validity pattern | Before | After |
| --- | ---: | ---: |
| leading null | 11.6 us/op | 7.1 us/op |
| clustered nulls | 8.6 us/op | 7.2 us/op |
| trailing null | 5.6 us/op | 5.6 us/op |
Random 10% nulls and alternating validity keep the existing fallback
path. All cases stayed at **0 B/op** and **0 allocs/op**.
## Tests
- `PARQUET_TEST_DATA=parquet-testing/data
ARROW_TEST_DATA=arrow-testing/data go test -p 2 ./... -count=1`
- `go test -race -p 2 ./parquet/internal/encoding ./parquet/file`
- `go vet -p 2 ./parquet/internal/encoding ./parquet/file`
---
.../plain_decoder_spaced_benchmark_test.go | 96 ++++++++++++++++++
.../internal/encoding/plain_decoder_spaced_test.go | 108 +++++++++++++++++++++
parquet/internal/encoding/plain_encoding_types.go | 83 +++++++++++++++-
3 files changed, 282 insertions(+), 5 deletions(-)
diff --git a/parquet/internal/encoding/plain_decoder_spaced_benchmark_test.go
b/parquet/internal/encoding/plain_decoder_spaced_benchmark_test.go
new file mode 100644
index 00000000..14b02627
--- /dev/null
+++ b/parquet/internal/encoding/plain_decoder_spaced_benchmark_test.go
@@ -0,0 +1,96 @@
+// 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 encoding
+
+import (
+ "fmt"
+ "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/apache/arrow-go/v18/parquet"
+)
+
+func BenchmarkPlainDecoderDecodeSpaced(b *testing.B) {
+ const size = 1 << 16
+
+ patterns := []struct {
+ name string
+ valid func(int) bool
+ }{
+ {name: "trailing_null", valid: func(i int) bool { return i !=
size-1 }},
+ {name: "leading_null", valid: func(i int) bool { return i != 0
}},
+ {name: "clustered_nulls", valid: func(i int) bool { return i <
size/2-32 || i >= size/2+32 }},
+ {name: "random_10pct_nulls", valid: randomValidity(size, 10)},
+ {name: "alternating", valid: func(i int) bool { return i%2 == 0
}},
+ }
+
+ for _, pattern := range patterns {
+ pattern := pattern
+ b.Run(fmt.Sprintf("int32/%s", pattern.name), func(b *testing.B)
{
+ data, validBits, nullCount :=
newPlainInt32SpacedInput(size, pattern.valid)
+ out := make([]int32, size)
+ dec := NewDecoder(parquet.Types.Int32,
parquet.Encodings.Plain, nil, memory.DefaultAllocator).(Int32Decoder)
+
+ b.ReportAllocs()
+ b.SetBytes(int64(size * arrow.Int32SizeBytes))
+ b.ResetTimer()
+ for b.Loop() {
+ if err := dec.SetData(size-nullCount, data);
err != nil {
+ b.Fatal(err)
+ }
+ n, err := dec.DecodeSpaced(out, nullCount,
validBits, 0)
+ if err != nil {
+ b.Fatal(err)
+ }
+ if n != size {
+ b.Fatalf("expected %d values, got %d",
size, n)
+ }
+ }
+ })
+ }
+}
+
+func newPlainInt32SpacedInput(size int, valid func(int) bool) (data, validBits
[]byte, nullCount int) {
+ validBits = make([]byte, bitutil.BytesForBits(int64(size)))
+ values := make([]int32, 0, size)
+ for i := 0; i < size; i++ {
+ if valid(i) {
+ bitutil.SetBit(validBits, i)
+ values = append(values, int32(i))
+ } else {
+ nullCount++
+ }
+ }
+
+ enc := NewEncoder(parquet.Types.Int32, parquet.Encodings.Plain, false,
nil, memory.DefaultAllocator).(Int32Encoder)
+ enc.Put(values)
+ buf, _ := enc.FlushValues()
+ defer buf.Release()
+ return append([]byte(nil), buf.Bytes()...), validBits, nullCount
+}
+
+func randomValidity(size, nullPercent int) func(int) bool {
+ state := uint32(1)
+ valid := make([]bool, size)
+ for i := range valid {
+ state = state*1664525 + 1013904223
+ valid[i] = state%100 >= uint32(nullPercent)
+ }
+ return func(i int) bool { return valid[i] }
+}
diff --git a/parquet/internal/encoding/plain_decoder_spaced_test.go
b/parquet/internal/encoding/plain_decoder_spaced_test.go
new file mode 100644
index 00000000..bab202bb
--- /dev/null
+++ b/parquet/internal/encoding/plain_decoder_spaced_test.go
@@ -0,0 +1,108 @@
+// 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 encoding
+
+import (
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow/bitutil"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestPlainDecoderDecodeSpaced(t *testing.T) {
+ cases := []struct {
+ name string
+ valid []bool
+ offset int64
+ }{
+ {
+ name: "leading null",
+ valid: []bool{false, true, true, true, true, true},
+ },
+ {
+ name: "internal null",
+ valid: []bool{true, true, false, true, true, true},
+ },
+ {
+ name: "trailing null",
+ valid: []bool{true, true, true, true, true, false},
+ },
+ {
+ name: "clustered nulls",
+ valid: []bool{true, true, false, false, false, true,
true, true},
+ },
+ {
+ name: "fragmented validity",
+ valid: []bool{true, false, true, false, true, false,
true, false, true, false, true, false},
+ },
+ {
+ name: "highly fragmented validity",
+ valid: []bool{true, false, true, false, true, false,
true, false, true, false, true, false, true, false, true, false, true, false,
true, false},
+ },
+ {
+ name: "offset",
+ valid: []bool{false, true, true, false, true, true},
+ offset: 3,
+ },
+ {
+ name: "all null",
+ valid: []bool{false, false, false, false},
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ validBits := make([]byte,
bitutil.BytesForBits(tc.offset+int64(len(tc.valid))))
+ physical := make([]int32, 0, len(tc.valid))
+ nullCount := 0
+ for i, valid := range tc.valid {
+ if valid {
+ bitutil.SetBit(validBits,
int(tc.offset)+i)
+ physical = append(physical, int32(i))
+ } else {
+ nullCount++
+ }
+ }
+
+ enc := NewEncoder(parquet.Types.Int32,
parquet.Encodings.Plain, false, nil, memory.DefaultAllocator).(Int32Encoder)
+ enc.Put(physical)
+ data, err := enc.FlushValues()
+ require.NoError(t, err)
+ defer data.Release()
+
+ dec := NewDecoder(parquet.Types.Int32,
parquet.Encodings.Plain, nil, memory.DefaultAllocator).(Int32Decoder)
+ require.NoError(t, dec.SetData(len(physical),
data.Bytes()))
+
+ out := make([]int32, len(tc.valid))
+ for i := range out {
+ out[i] = -1
+ }
+
+ n, err := dec.DecodeSpaced(out, nullCount, validBits,
tc.offset)
+ require.NoError(t, err)
+ assert.Equal(t, len(tc.valid), n)
+ for i, valid := range tc.valid {
+ if valid {
+ assert.Equal(t, int32(i), out[i])
+ }
+ }
+ })
+ }
+}
diff --git a/parquet/internal/encoding/plain_encoding_types.go
b/parquet/internal/encoding/plain_encoding_types.go
index f026e7ed..dd8119f5 100644
--- a/parquet/internal/encoding/plain_encoding_types.go
+++ b/parquet/internal/encoding/plain_encoding_types.go
@@ -96,14 +96,76 @@ func (dec *PlainDecoder[T]) Decode(out []T) (int, error) {
}
func (dec *PlainDecoder[T]) DecodeSpaced(out []T, nullCount int, validBits
[]byte, validBitsOffset int64) (int, error) {
+ if nullCount == 0 {
+ if err := dec.decodeDense(out, len(out)); err != nil {
+ return 0, err
+ }
+ return len(out), nil
+ }
+
toread := len(out) - nullCount
- values, err := dec.Decode(out[:toread])
- if err != nil {
- return 0, err
+ if toread == 0 {
+ return len(out), nil
}
- if values != toread {
- return 0, errors.New("parquet: number of values / definition
levels read did not match")
+ if dec.bitSetReader == nil {
+ dec.bitSetReader =
bitutils.NewReverseSetBitRunReader(validBits, validBitsOffset, int64(len(out)))
+ } else {
+ dec.bitSetReader.Reset(validBits, validBitsOffset,
int64(len(out)))
+ }
+
+ const maxDirectDecodeRuns = 8
+ var validRuns [maxDirectDecodeRuns]bitutils.SetBitRun
+ runCount := 0
+ decodedPos := int64(0)
+ needsExpansion := false
+ for {
+ run := dec.bitSetReader.NextRun()
+ if run.Length == 0 {
+ break
+ }
+ if runCount == len(validRuns) {
+ return dec.decodeSpacedDense(out, nullCount, validBits,
validBitsOffset)
+ }
+ validRuns[runCount] = run
+ runCount++
+ densePos := int64(toread) - decodedPos - run.Length
+ if run.Pos != densePos {
+ needsExpansion = true
+ }
+ decodedPos += run.Length
+ if decodedPos >= int64(toread) {
+ break
+ }
+ }
+
+ if decodedPos != int64(toread) {
+ return dec.decodeSpacedDense(out, nullCount, validBits,
validBitsOffset)
+ }
+ if !needsExpansion {
+ if err := dec.decodeDense(out, toread); err != nil {
+ return 0, err
+ }
+ return len(out), nil
+ }
+
+ for i := runCount - 1; i >= 0; i-- {
+ run := validRuns[i]
+ values, err :=
dec.Decode(out[int(run.Pos):int(run.Pos+run.Length)])
+ if err != nil {
+ return 0, err
+ }
+ if values != int(run.Length) {
+ return 0, errors.New("parquet: number of values /
definition levels read did not match")
+ }
+ }
+ return len(out), nil
+}
+
+func (dec *PlainDecoder[T]) decodeSpacedDense(out []T, nullCount int,
validBits []byte, validBitsOffset int64) (int, error) {
+ toread := len(out) - nullCount
+ if err := dec.decodeDense(out, toread); err != nil {
+ return 0, err
}
nvalues := len(out)
@@ -130,6 +192,17 @@ func (dec *PlainDecoder[T]) DecodeSpaced(out []T,
nullCount int, validBits []byt
return nvalues, nil
}
+func (dec *PlainDecoder[T]) decodeDense(out []T, toread int) error {
+ values, err := dec.Decode(out[:toread])
+ if err != nil {
+ return err
+ }
+ if values != toread {
+ return errors.New("parquet: number of values / definition
levels read did not match")
+ }
+ return nil
+}
+
type (
PlainInt32Encoder = PlainEncoder[int32]
PlainInt32Decoder = PlainDecoder[int32]
