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 066b7916 perf(parquet/pqarrow): write binary values from Arrow offsets
(#1262)
066b7916 is described below
commit 066b79161f09ae9055ce4439e4e1ab7b9b43cbdc
Author: Minh Vu <[email protected]>
AuthorDate: Wed Sep 2 20:24:13 2026 +0200
perf(parquet/pqarrow): write binary values from Arrow offsets (#1262)
## Summary
- **Writes Binary, String, LargeBinary, and LargeString values from
Arrow buffers and offsets.**
- Removes the full `[]parquet.ByteArray` allocation from non-dictionary
writes.
- Keeps nullable writes, page statistics, bloom filters, page limits,
and dictionary fallback behavior.
- Adds coverage for 32-bit and 64-bit offsets, nulls, slices, small
batches, and all byte-array encodings.
## Benchmark
64K String rows, uncompressed, dictionary disabled, Apple M1 Pro. Median
of 3 runs with 3 seconds per run.
| stats | before | after | heap/op |
| --- | ---: | ---: | ---: |
| false | 1.798 ms | 1.260 ms | 10.62 MB -> 9.04 MB |
| true | 2.343 ms | 1.861 ms | 10.62 MB -> 9.06 MB |
Benchmark command: `go test ./parquet/pqarrow -run '^$' -bench
'^BenchmarkWriteArrowBinaryOffsets$' -benchmem -benchtime=3s -count=3`
## Tests
- `go test ./parquet/...`
- `go vet ./parquet/internal/encoding ./parquet/metadata ./parquet/file
./parquet/pqarrow`
---
parquet/file/byte_array_column_writer_arrow.go | 345 +++++++++++++++++++++
.../internal/encoding/byte_array_arrow_encoder.go | 232 ++++++++++++++
.../encoding/byte_array_arrow_encoder_test.go | 190 ++++++++++++
parquet/metadata/bloom_filter_arrow_offsets.go | 100 ++++++
parquet/metadata/byte_array_arrow_offsets.go | 89 ++++++
parquet/metadata/byte_array_arrow_offsets_test.go | 114 +++++++
parquet/pqarrow/binary_offsets_bench_test.go | 75 +++++
parquet/pqarrow/binary_offsets_test.go | 234 ++++++++++++++
parquet/pqarrow/encode_arrow.go | 24 ++
9 files changed, 1403 insertions(+)
diff --git a/parquet/file/byte_array_column_writer_arrow.go
b/parquet/file/byte_array_column_writer_arrow.go
new file mode 100644
index 00000000..ce99540c
--- /dev/null
+++ b/parquet/file/byte_array_column_writer_arrow.go
@@ -0,0 +1,345 @@
+// 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 file
+
+import (
+ "fmt"
+
+ "github.com/apache/arrow-go/v18/internal/utils"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/metadata"
+)
+
+type byteArrayArrowOffset interface {
+ ~int32 | ~int64
+}
+
+type byteArrayArrowEncoder32 interface {
+ PutArrow([]byte, []int32)
+ PutArrowSpaced([]byte, []int32, []byte, int64)
+}
+
+type byteArrayArrowEncoder64 interface {
+ PutArrow64([]byte, []int64)
+ PutArrowSpaced64([]byte, []int64, []byte, int64)
+}
+
+func arrowOffsetsForBatch[T byteArrayArrowOffset](offsets []T, offset, count
int64) []T {
+ if offset < 0 || offset > int64(len(offsets)) {
+ panic("parquet: Arrow offset index is out of bounds")
+ }
+ if count == 0 {
+ if offset == int64(len(offsets)) {
+ return nil
+ }
+ return offsets[offset : offset+1]
+ }
+
+ end := offset + count + 1
+ if end > int64(len(offsets)) {
+ panic("parquet: Arrow offset range is out of bounds")
+ }
+ return offsets[offset:end]
+}
+
+// SupportsArrowOffsets reports whether the active byte-array encoder can
consume
+// Arrow's value buffer and offsets without materializing parquet.ByteArray
values.
+func (w *ByteArrayColumnChunkWriter) SupportsArrowOffsets() bool {
+ _, supports32 := w.currentEncoder.(byteArrayArrowEncoder32)
+ _, supports64 := w.currentEncoder.(byteArrayArrowEncoder64)
+ return supports32 || supports64
+}
+
+func writeArrowValues[T byteArrayArrowOffset](w *ByteArrayColumnChunkWriter,
values []byte, offsets []T, numNulls int64) {
+ switch offsets := any(offsets).(type) {
+ case []int32:
+ enc, ok := w.currentEncoder.(byteArrayArrowEncoder32)
+ if !ok {
+ panic("parquet: current byte-array encoder does not
support Arrow offsets")
+ }
+ enc.PutArrow(values, offsets)
+ if w.pageStatistics != nil {
+
w.pageStatistics.(*metadata.ByteArrayStatistics).UpdateFromArrowOffsets(values,
offsets, numNulls)
+ }
+ if w.bloomFilter != nil && w.currentEncoder.Encoding() !=
parquet.Encodings.PlainDict {
+ metadata.InsertArrowOffsetHashes(w.bloomFilter, values,
offsets)
+ }
+ case []int64:
+ enc, ok := w.currentEncoder.(byteArrayArrowEncoder64)
+ if !ok {
+ panic("parquet: current byte-array encoder does not
support Arrow offsets")
+ }
+ enc.PutArrow64(values, offsets)
+ if w.pageStatistics != nil {
+
w.pageStatistics.(*metadata.ByteArrayStatistics).UpdateFromArrowOffsets64(values,
offsets, numNulls)
+ }
+ if w.bloomFilter != nil && w.currentEncoder.Encoding() !=
parquet.Encodings.PlainDict {
+ metadata.InsertArrowOffsetHashes64(w.bloomFilter,
values, offsets)
+ }
+ }
+}
+
+func writeArrowValuesSpaced[T byteArrayArrowOffset](w
*ByteArrayColumnChunkWriter, values []byte, offsets []T, numRead, numValues
int64, validBits []byte, validBitsOffset int64) {
+ numSpaced := int64(0)
+ if len(offsets) > 0 {
+ numSpaced = int64(len(offsets) - 1)
+ }
+
+ switch offsets := any(offsets).(type) {
+ case []int32:
+ enc, ok := w.currentEncoder.(byteArrayArrowEncoder32)
+ if !ok {
+ panic("parquet: current byte-array encoder does not
support Arrow offsets")
+ }
+ if numSpaced != numRead {
+ enc.PutArrowSpaced(values, offsets, validBits,
validBitsOffset)
+ } else {
+ enc.PutArrow(values, offsets)
+ }
+ if w.pageStatistics != nil {
+ stats :=
w.pageStatistics.(*metadata.ByteArrayStatistics)
+ stats.UpdateFromArrowOffsetsSpaced(values, offsets,
validBits, validBitsOffset, numSpaced-numRead)
+ stats.IncNulls(numValues - numSpaced)
+ }
+ if w.bloomFilter != nil && w.currentEncoder.Encoding() !=
parquet.Encodings.PlainDict {
+ metadata.InsertSpacedArrowOffsetHashes(w.bloomFilter,
numRead, values, offsets, validBits, validBitsOffset)
+ }
+ case []int64:
+ enc, ok := w.currentEncoder.(byteArrayArrowEncoder64)
+ if !ok {
+ panic("parquet: current byte-array encoder does not
support Arrow offsets")
+ }
+ if numSpaced != numRead {
+ enc.PutArrowSpaced64(values, offsets, validBits,
validBitsOffset)
+ } else {
+ enc.PutArrow64(values, offsets)
+ }
+ if w.pageStatistics != nil {
+ stats :=
w.pageStatistics.(*metadata.ByteArrayStatistics)
+ stats.UpdateFromArrowOffsetsSpaced64(values, offsets,
validBits, validBitsOffset, numSpaced-numRead)
+ stats.IncNulls(numValues - numSpaced)
+ }
+ if w.bloomFilter != nil && w.currentEncoder.Encoding() !=
parquet.Encodings.PlainDict {
+ metadata.InsertSpacedArrowOffsetHashes64(w.bloomFilter,
numRead, values, offsets, validBits, validBitsOffset)
+ }
+ }
+}
+
+func writeBatchArrow[T byteArrayArrowOffset](w *ByteArrayColumnChunkWriter,
values []byte, offsets []T, defLevels, repLevels []int16) (valueOffset int64) {
+ var n int64
+ if defLevels != nil {
+ n = int64(len(defLevels))
+ } else if len(offsets) > 0 {
+ n = int64(len(offsets) - 1)
+ }
+ if n == 0 {
+ return 0
+ }
+
+ const maxSafeBatchDataSize int64 = 1 << 30
+ batchSize := w.props.WriteBatchSize()
+ maxDefLevel := w.descr.MaxDefinitionLevel()
+ requiresRowAlignment := (w.props.DataPageVersion() !=
parquet.DataPageV1 ||
+ w.props.PageIndexEnabledFor(w.descr.Path())) &&
+ repLevels != nil && w.descr.MaxRepetitionLevel() > 0
+ levelOffset := int64(0)
+ valueCount := int64(0)
+ if len(offsets) > 0 {
+ valueCount = int64(len(offsets) - 1)
+ }
+
+ if requiresRowAlignment {
+ if int64(len(repLevels)) < n {
+ panic("columnwriter: not enough repetition levels for
batch to write")
+ }
+ if repLevels[0] != 0 {
+ panic("columnwriter: row-aligned batch writing must
start at a row boundary")
+ }
+ repLevels = repLevels[:n]
+ }
+
+ for levelOffset < n {
+ remaining := n - levelOffset
+ batch := min(remaining, batchSize)
+
+ var cumDataSize int64
+ valueScan := valueOffset
+ for li := int64(0); li < batch; li++ {
+ isValue := defLevels == nil || maxDefLevel == 0 ||
defLevels[levelOffset+li] == maxDefLevel
+ if isValue && valueScan < valueCount {
+ valueSize := int64(offsets[valueScan+1]) -
int64(offsets[valueScan]) + 4
+ if cumDataSize+valueSize > maxSafeBatchDataSize
&& li > 0 {
+ batch = li
+ break
+ }
+ cumDataSize += valueSize
+ valueScan++
+ }
+ }
+
+ if requiresRowAlignment {
+ batch = alignBatchToRowBoundary(repLevels, levelOffset,
batch)
+ }
+ if batch < 1 {
+ batch = 1
+ }
+
+ toWrite := w.writeLevels(batch, levelSliceOrNil(defLevels,
levelOffset, batch),
+ levelSliceOrNil(repLevels, levelOffset, batch))
+ batchOffsets := arrowOffsetsForBatch(offsets, valueOffset,
toWrite)
+ writeArrowValues(w, values, batchOffsets, batch-toWrite)
+ if err := w.commitWriteAndCheckPageLimit(batch, toWrite); err
!= nil {
+ panic(err)
+ }
+ valueOffset += toWrite
+ w.checkDictionarySizeLimit()
+ levelOffset += batch
+ }
+ return valueOffset
+}
+
+func writeBatchSpacedArrow[T byteArrayArrowOffset](w
*ByteArrayColumnChunkWriter, values []byte, offsets []T, defLevels, repLevels
[]int16, validBits []byte, validBitsOffset int64) (valueOffset int64) {
+ length := len(defLevels)
+ if defLevels == nil {
+ if len(offsets) > 0 {
+ length = len(offsets) - 1
+ } else {
+ length = 0
+ }
+ }
+ if length == 0 {
+ return 0
+ }
+
+ const maxSafeBatchDataSize int64 = 1 << 30
+ batchSize := w.props.WriteBatchSize()
+ requiresRowAlignment := (w.props.DataPageVersion() !=
parquet.DataPageV1 ||
+ w.props.PageIndexEnabledFor(w.descr.Path())) &&
+ repLevels != nil && w.descr.MaxRepetitionLevel() > 0
+ levelOffset := int64(0)
+ n := int64(length)
+ valueCount := int64(0)
+ if len(offsets) > 0 {
+ valueCount = int64(len(offsets) - 1)
+ }
+
+ if requiresRowAlignment {
+ if int64(len(repLevels)) < n {
+ panic("columnwriter: not enough repetition levels for
batch to write")
+ }
+ if repLevels[0] != 0 {
+ panic("columnwriter: row-aligned batch writing must
start at a row boundary")
+ }
+ repLevels = repLevels[:n]
+ }
+
+ for levelOffset < n {
+ remaining := n - levelOffset
+ batch := min(remaining, batchSize)
+
+ var cumDataSize int64
+ for vi := int64(0); vi < batch && valueOffset+vi < valueCount;
vi++ {
+ valueSize := int64(offsets[valueOffset+vi+1]) -
int64(offsets[valueOffset+vi]) + 4
+ if cumDataSize+valueSize > maxSafeBatchDataSize && vi >
0 {
+ batch = vi
+ break
+ }
+ cumDataSize += valueSize
+ }
+
+ if requiresRowAlignment {
+ batch = alignBatchToRowBoundary(repLevels, levelOffset,
batch)
+ }
+ if batch < 1 {
+ batch = 1
+ }
+
+ info := w.maybeCalculateValidityBits(levelSliceOrNil(defLevels,
levelOffset, batch), batch)
+ w.writeLevelsSpaced(batch, levelSliceOrNil(defLevels,
levelOffset, batch),
+ levelSliceOrNil(repLevels, levelOffset, batch))
+ batchOffsets := arrowOffsetsForBatch(offsets, valueOffset,
info.numSpaced())
+
+ writeBits := validBits
+ writeBitsOffset := validBitsOffset + valueOffset
+ if w.bitsBuffer != nil {
+ writeBits = w.bitsBuffer.Bytes()
+ writeBitsOffset = 0
+ }
+ writeArrowValuesSpaced(w, values, batchOffsets, info.batchNum,
batch, writeBits, writeBitsOffset)
+ if err := w.commitWriteAndCheckPageLimit(batch,
info.numSpaced()); err != nil {
+ panic(err)
+ }
+ valueOffset += info.numSpaced()
+ w.checkDictionarySizeLimit()
+ levelOffset += batch
+ }
+ return valueOffset
+}
+
+// WriteBatchArrow writes Arrow binary data using its value buffer and 32-bit
offsets.
+// The active encoder must support Arrow offsets; dictionary encoders use the
existing
+// parquet.ByteArray path instead.
+func (w *ByteArrayColumnChunkWriter) WriteBatchArrow(values []byte, offsets
[]int32, defLevels, repLevels []int16) (valueOffset int64, err error) {
+ defer func() {
+ if r := recover(); r != nil {
+ err = utils.FormatRecoveredError("unknown error type",
r)
+ }
+ }()
+ if _, ok := w.currentEncoder.(byteArrayArrowEncoder32); !ok {
+ return 0, fmt.Errorf("parquet: current byte-array encoder does
not support 32-bit Arrow offsets")
+ }
+ return writeBatchArrow(w, values, offsets, defLevels, repLevels), nil
+}
+
+// WriteBatchArrow64 writes Arrow binary data using its value buffer and
64-bit offsets.
+func (w *ByteArrayColumnChunkWriter) WriteBatchArrow64(values []byte, offsets
[]int64, defLevels, repLevels []int16) (valueOffset int64, err error) {
+ defer func() {
+ if r := recover(); r != nil {
+ err = utils.FormatRecoveredError("unknown error type",
r)
+ }
+ }()
+ if _, ok := w.currentEncoder.(byteArrayArrowEncoder64); !ok {
+ return 0, fmt.Errorf("parquet: current byte-array encoder does
not support 64-bit Arrow offsets")
+ }
+ return writeBatchArrow(w, values, offsets, defLevels, repLevels), nil
+}
+
+// WriteBatchSpacedArrow writes spaced Arrow binary data using 32-bit offsets.
+func (w *ByteArrayColumnChunkWriter) WriteBatchSpacedArrow(values []byte,
offsets []int32, defLevels, repLevels []int16, validBits []byte,
validBitsOffset int64) (valueOffset int64, err error) {
+ defer func() {
+ if r := recover(); r != nil {
+ err = utils.FormatRecoveredError("unknown error type",
r)
+ }
+ }()
+ if _, ok := w.currentEncoder.(byteArrayArrowEncoder32); !ok {
+ return 0, fmt.Errorf("parquet: current byte-array encoder does
not support 32-bit Arrow offsets")
+ }
+ return writeBatchSpacedArrow(w, values, offsets, defLevels, repLevels,
validBits, validBitsOffset), nil
+}
+
+// WriteBatchSpacedArrow64 writes spaced Arrow binary data using 64-bit
offsets.
+func (w *ByteArrayColumnChunkWriter) WriteBatchSpacedArrow64(values []byte,
offsets []int64, defLevels, repLevels []int16, validBits []byte,
validBitsOffset int64) (valueOffset int64, err error) {
+ defer func() {
+ if r := recover(); r != nil {
+ err = utils.FormatRecoveredError("unknown error type",
r)
+ }
+ }()
+ if _, ok := w.currentEncoder.(byteArrayArrowEncoder64); !ok {
+ return 0, fmt.Errorf("parquet: current byte-array encoder does
not support 64-bit Arrow offsets")
+ }
+ return writeBatchSpacedArrow(w, values, offsets, defLevels, repLevels,
validBits, validBitsOffset), nil
+}
diff --git a/parquet/internal/encoding/byte_array_arrow_encoder.go
b/parquet/internal/encoding/byte_array_arrow_encoder.go
new file mode 100644
index 00000000..b5253a55
--- /dev/null
+++ b/parquet/internal/encoding/byte_array_arrow_encoder.go
@@ -0,0 +1,232 @@
+// 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 (
+ "encoding/binary"
+
+ "github.com/apache/arrow-go/v18/internal/bitutils"
+ "github.com/apache/arrow-go/v18/parquet"
+)
+
+type arrowByteArrayOffset interface {
+ ~int32 | ~int64
+}
+
+func putArrowPlain[T arrowByteArrayOffset](sink *PooledBufferWriter, values
[]byte, offsets []T) {
+ if len(offsets) < 2 {
+ return
+ }
+
+ encodedSize := 0
+ for i := 0; i < len(offsets)-1; i++ {
+ encodedSize += int(offsets[i+1]-offsets[i]) + 4
+ }
+
+ sink.Reserve(encodedSize)
+ out := sink.buf.Buf()[sink.pos : sink.pos+encodedSize]
+ for i := 0; i < len(offsets)-1; i++ {
+ start := int(offsets[i])
+ end := int(offsets[i+1])
+ binary.LittleEndian.PutUint32(out, uint32(end-start))
+ copy(out[4:], values[start:end])
+ out = out[4+end-start:]
+ }
+ sink.pos += encodedSize
+}
+
+func putArrowPlainSpaced[T arrowByteArrayOffset](sink *PooledBufferWriter,
values []byte, offsets []T, validBits []byte, validBitsOffset int64) {
+ if len(offsets) < 2 {
+ return
+ }
+ if validBits == nil {
+ putArrowPlain(sink, values, offsets)
+ return
+ }
+
+ bitutils.VisitSetBitRunsNoErr(validBits, validBitsOffset,
int64(len(offsets)-1), func(pos, length int64) {
+ putArrowPlain(sink, values, offsets[pos:pos+length+1])
+ })
+}
+
+func (enc *PlainByteArrayEncoder) PutArrow(values []byte, offsets []int32) {
+ putArrowPlain(enc.sink, values, offsets)
+}
+
+func (enc *PlainByteArrayEncoder) PutArrow64(values []byte, offsets []int64) {
+ putArrowPlain(enc.sink, values, offsets)
+}
+
+func (enc *PlainByteArrayEncoder) PutArrowSpaced(values []byte, offsets
[]int32, validBits []byte, validBitsOffset int64) {
+ putArrowPlainSpaced(enc.sink, values, offsets, validBits,
validBitsOffset)
+}
+
+func (enc *PlainByteArrayEncoder) PutArrowSpaced64(values []byte, offsets
[]int64, validBits []byte, validBitsOffset int64) {
+ putArrowPlainSpaced(enc.sink, values, offsets, validBits,
validBitsOffset)
+}
+
+func putArrowDeltaLength[T arrowByteArrayOffset](enc
*DeltaLengthByteArrayEncoder, values []byte, offsets []T, validBits []byte,
validBitsOffset int64) {
+ if len(offsets) < 2 {
+ return
+ }
+
+ if validBits == nil {
+ batchSize := 0
+ totalLen := 0
+ for i := 0; i < len(offsets)-1; i++ {
+ start := int(offsets[i])
+ end := int(offsets[i+1])
+ enc.lengths[batchSize] = int32(end - start)
+ totalLen += end - start
+ batchSize++
+ if batchSize == len(enc.lengths) {
+ enc.lengthEncoder.Put(enc.lengths[:batchSize])
+ batchSize = 0
+ }
+ }
+ if batchSize != 0 {
+ enc.lengthEncoder.Put(enc.lengths[:batchSize])
+ }
+
+ enc.sink.Reserve(totalLen)
+ for i := 0; i < len(offsets)-1; i++ {
+
enc.sink.UnsafeWrite(values[int(offsets[i]):int(offsets[i+1])])
+ }
+ return
+ }
+
+ batchSize := 0
+ totalLen := 0
+ visit := func(pos, length int64) {
+ for i := pos; i < pos+length; i++ {
+ start := int(offsets[i])
+ end := int(offsets[i+1])
+ enc.lengths[batchSize] = int32(end - start)
+ totalLen += end - start
+ batchSize++
+ if batchSize == len(enc.lengths) {
+ enc.lengthEncoder.Put(enc.lengths[:batchSize])
+ batchSize = 0
+ }
+ }
+ }
+ bitutils.VisitSetBitRunsNoErr(validBits, validBitsOffset,
int64(len(offsets)-1), visit)
+ if batchSize != 0 {
+ enc.lengthEncoder.Put(enc.lengths[:batchSize])
+ }
+
+ 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])])
+ }
+ })
+}
+
+func (enc *DeltaLengthByteArrayEncoder) PutArrow(values []byte, offsets
[]int32) {
+ putArrowDeltaLength(enc, values, offsets, nil, 0)
+}
+
+func (enc *DeltaLengthByteArrayEncoder) PutArrow64(values []byte, offsets
[]int64) {
+ putArrowDeltaLength(enc, values, offsets, nil, 0)
+}
+
+func (enc *DeltaLengthByteArrayEncoder) PutArrowSpaced(values []byte, offsets
[]int32, validBits []byte, validBitsOffset int64) {
+ putArrowDeltaLength(enc, values, offsets, validBits, validBitsOffset)
+}
+
+func (enc *DeltaLengthByteArrayEncoder) PutArrowSpaced64(values []byte,
offsets []int64, validBits []byte, validBitsOffset int64) {
+ putArrowDeltaLength(enc, values, offsets, validBits, validBitsOffset)
+}
+
+func putArrowDeltaByte[T arrowByteArrayOffset](enc *DeltaByteArrayEncoder,
values []byte, offsets []T, validBits []byte, validBitsOffset int64) {
+ if len(offsets) < 2 {
+ return
+ }
+
+ if enc.prefixEncoder == nil {
+ enc.initEncoders()
+ }
+
+ lastVal := enc.lastVal
+ if validBits == nil {
+ batchSize := 0
+ for i := 0; i < len(offsets)-1; i++ {
+ val :=
parquet.ByteArray(values[int(offsets[i]):int(offsets[i+1])])
+ prefixLength := commonPrefixLength(lastVal, val)
+ lastVal = val
+ enc.prefixLengths[batchSize] = int32(prefixLength)
+ enc.suffixes[batchSize] = val[prefixLength:]
+ batchSize++
+ if batchSize == len(enc.suffixes) {
+ enc.suffixEncoder.Put(enc.suffixes[:batchSize])
+
enc.prefixEncoder.Put(enc.prefixLengths[:batchSize])
+ clear(enc.suffixes[:batchSize])
+ batchSize = 0
+ }
+ }
+ if batchSize != 0 {
+ enc.suffixEncoder.Put(enc.suffixes[:batchSize])
+ enc.prefixEncoder.Put(enc.prefixLengths[:batchSize])
+ clear(enc.suffixes[:batchSize])
+ }
+ enc.lastVal = append(enc.lastVal[:0], lastVal...)
+ return
+ }
+
+ batchSize := 0
+ visit := func(pos, length int64) {
+ for i := pos; i < pos+length; i++ {
+ val :=
parquet.ByteArray(values[int(offsets[i]):int(offsets[i+1])])
+ prefixLength := commonPrefixLength(lastVal, val)
+ lastVal = val
+ enc.prefixLengths[batchSize] = int32(prefixLength)
+ enc.suffixes[batchSize] = val[prefixLength:]
+ batchSize++
+ if batchSize == len(enc.suffixes) {
+ enc.suffixEncoder.Put(enc.suffixes[:batchSize])
+
enc.prefixEncoder.Put(enc.prefixLengths[:batchSize])
+ clear(enc.suffixes[:batchSize])
+ batchSize = 0
+ }
+ }
+ }
+ bitutils.VisitSetBitRunsNoErr(validBits, validBitsOffset,
int64(len(offsets)-1), visit)
+ if batchSize != 0 {
+ enc.suffixEncoder.Put(enc.suffixes[:batchSize])
+ enc.prefixEncoder.Put(enc.prefixLengths[:batchSize])
+ clear(enc.suffixes[:batchSize])
+ }
+
+ enc.lastVal = append(enc.lastVal[:0], lastVal...)
+}
+
+func (enc *DeltaByteArrayEncoder) PutArrow(values []byte, offsets []int32) {
+ putArrowDeltaByte(enc, values, offsets, nil, 0)
+}
+
+func (enc *DeltaByteArrayEncoder) PutArrow64(values []byte, offsets []int64) {
+ putArrowDeltaByte(enc, values, offsets, nil, 0)
+}
+
+func (enc *DeltaByteArrayEncoder) PutArrowSpaced(values []byte, offsets
[]int32, validBits []byte, validBitsOffset int64) {
+ putArrowDeltaByte(enc, values, offsets, validBits, validBitsOffset)
+}
+
+func (enc *DeltaByteArrayEncoder) PutArrowSpaced64(values []byte, offsets
[]int64, validBits []byte, validBitsOffset int64) {
+ putArrowDeltaByte(enc, values, offsets, validBits, validBitsOffset)
+}
diff --git a/parquet/internal/encoding/byte_array_arrow_encoder_test.go
b/parquet/internal/encoding/byte_array_arrow_encoder_test.go
new file mode 100644
index 00000000..949d5e65
--- /dev/null
+++ b/parquet/internal/encoding/byte_array_arrow_encoder_test.go
@@ -0,0 +1,190 @@
+// 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/bitutil"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/stretchr/testify/require"
+)
+
+type arrowByteArrayEncoder32Test interface {
+ ByteArrayEncoder
+ PutArrow([]byte, []int32)
+ PutArrowSpaced([]byte, []int32, []byte, int64)
+}
+
+type arrowByteArrayEncoder64Test interface {
+ ByteArrayEncoder
+ PutArrow64([]byte, []int64)
+ PutArrowSpaced64([]byte, []int64, []byte, int64)
+}
+
+func arrowByteArrayInput() ([]parquet.ByteArray, []byte, []int32, []int64) {
+ values := []parquet.ByteArray{
+ []byte("prefix/000"),
+ []byte("prefix/001"),
+ {},
+ []byte("prefix/003"),
+ []byte("other"),
+ {},
+ []byte("other/longer"),
+ }
+ data := make([]byte, 0, 64)
+ offsets32 := []int32{0}
+ offsets64 := []int64{0}
+ for _, value := range values {
+ data = append(data, value...)
+ offsets32 = append(offsets32, int32(len(data)))
+ offsets64 = append(offsets64, int64(len(data)))
+ }
+ return values, data, offsets32, offsets64
+}
+
+func encodedByteArrays(t *testing.T, encoding parquet.Encoding, values
[]parquet.ByteArray, spaced bool, validBits []byte, validBitsOffset int64)
[]byte {
+ t.Helper()
+ enc := NewEncoder(parquet.Types.ByteArray, encoding, false, nil,
memory.DefaultAllocator).(ByteArrayEncoder)
+ defer enc.Release()
+ if spaced {
+ enc.PutSpaced(values, validBits, validBitsOffset)
+ } else {
+ enc.Put(values)
+ }
+ buf, err := enc.FlushValues()
+ require.NoError(t, err)
+ defer buf.Release()
+ return append([]byte(nil), buf.Bytes()...)
+}
+
+func TestByteArrayArrowEncodersMatchByteArrayInput(t *testing.T) {
+ values, data, offsets32, offsets64 := arrowByteArrayInput()
+ validBits := make([]byte, bitutil.BytesForBits(int64(len(values)+5)))
+ validBitsOffset := int64(3)
+ for _, index := range []int{0, 1, 3, 4, 6} {
+ bitutil.SetBit(validBits, int(validBitsOffset)+index)
+ }
+
+ for _, encoding := range []parquet.Encoding{
+ parquet.Encodings.Plain,
+ parquet.Encodings.DeltaLengthByteArray,
+ parquet.Encodings.DeltaByteArray,
+ } {
+ t.Run(encoding.String(), func(t *testing.T) {
+ want := encodedByteArrays(t, encoding, values, false,
nil, 0)
+ wantSpaced := encodedByteArrays(t, encoding, values,
true, validBits, validBitsOffset)
+
+ for _, width := range []string{"int32", "int64"} {
+ t.Run(width, func(t *testing.T) {
+ enc :=
NewEncoder(parquet.Types.ByteArray, encoding, false, nil,
memory.DefaultAllocator).(ByteArrayEncoder)
+ defer enc.Release()
+ if width == "int32" {
+ direct :=
enc.(arrowByteArrayEncoder32Test)
+ direct.PutArrow(data, offsets32)
+ } else {
+ direct :=
enc.(arrowByteArrayEncoder64Test)
+ direct.PutArrow64(data,
offsets64)
+ }
+ buf, err := enc.FlushValues()
+ require.NoError(t, err)
+ require.Equal(t, want, buf.Bytes())
+ buf.Release()
+
+ if width == "int32" {
+ direct :=
enc.(arrowByteArrayEncoder32Test)
+ direct.PutArrowSpaced(data,
offsets32, validBits, validBitsOffset)
+ } else {
+ direct :=
enc.(arrowByteArrayEncoder64Test)
+ direct.PutArrowSpaced64(data,
offsets64, validBits, validBitsOffset)
+ }
+ buf, err = enc.FlushValues()
+ require.NoError(t, err)
+ require.Equal(t, wantSpaced,
buf.Bytes())
+ buf.Release()
+ })
+ }
+ })
+ }
+}
+
+func TestDeltaByteArrayArrowEncoderPreservesStateAcrossBatches(t *testing.T) {
+ values, data, offsets32, offsets64 := arrowByteArrayInput()
+
+ for _, width := range []string{"int32", "int64"} {
+ t.Run(width, func(t *testing.T) {
+ enc := NewEncoder(parquet.Types.ByteArray,
parquet.Encodings.DeltaByteArray, false, nil,
memory.DefaultAllocator).(ByteArrayEncoder)
+ defer enc.Release()
+ direct32, direct64 :=
enc.(arrowByteArrayEncoder32Test), enc.(arrowByteArrayEncoder64Test)
+ if width == "int32" {
+ direct32.PutArrow(data, offsets32[:3])
+ direct32.PutArrow(data, offsets32[2:])
+ } else {
+ direct64.PutArrow64(data, offsets64[:3])
+ direct64.PutArrow64(data, offsets64[2:])
+ }
+ got, err := enc.FlushValues()
+ require.NoError(t, err)
+ defer got.Release()
+ want := encodedByteArrays(t,
parquet.Encodings.DeltaByteArray, values, false, nil, 0)
+ require.Equal(t, want, got.Bytes())
+ })
+ }
+}
+
+func TestDeltaArrowEncodersAcrossInternalBatches(t *testing.T) {
+ const nvalues = deltaByteArrayBatchSize*2 + 17
+ values := make([]parquet.ByteArray, nvalues)
+ data := make([]byte, 0, nvalues*16)
+ offsets32 := []int32{0}
+ offsets64 := []int64{0}
+ validBits := make([]byte, bitutil.BytesForBits(nvalues+1))
+ validBitsOffset := int64(1)
+ for i := range values {
+ values[i] =
parquet.ByteArray(fmt.Sprintf("partition/%03d/value", i))
+ data = append(data, values[i]...)
+ offsets32 = append(offsets32, int32(len(data)))
+ offsets64 = append(offsets64, int64(len(data)))
+ if i%5 != 0 {
+ bitutil.SetBit(validBits, int(validBitsOffset)+i)
+ }
+ }
+
+ for _, encoding := range []parquet.Encoding{
+ parquet.Encodings.DeltaLengthByteArray,
+ parquet.Encodings.DeltaByteArray,
+ } {
+ for _, width := range []string{"int32", "int64"} {
+ t.Run(encoding.String()+"/"+width, func(t *testing.T) {
+ want := encodedByteArrays(t, encoding, values,
true, validBits, validBitsOffset)
+ enc := NewEncoder(parquet.Types.ByteArray,
encoding, false, nil, memory.DefaultAllocator).(ByteArrayEncoder)
+ defer enc.Release()
+ if width == "int32" {
+
enc.(arrowByteArrayEncoder32Test).PutArrowSpaced(data, offsets32, validBits,
validBitsOffset)
+ } else {
+
enc.(arrowByteArrayEncoder64Test).PutArrowSpaced64(data, offsets64, validBits,
validBitsOffset)
+ }
+ got, err := enc.FlushValues()
+ require.NoError(t, err)
+ require.Equal(t, want, got.Bytes())
+ got.Release()
+ })
+ }
+ }
+}
diff --git a/parquet/metadata/bloom_filter_arrow_offsets.go
b/parquet/metadata/bloom_filter_arrow_offsets.go
new file mode 100644
index 00000000..1e4bfd8a
--- /dev/null
+++ b/parquet/metadata/bloom_filter_arrow_offsets.go
@@ -0,0 +1,100 @@
+// 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 metadata
+
+import "github.com/apache/arrow-go/v18/internal/bitutils"
+
+type byteArrayArrowHashOffset interface {
+ ~int32 | ~int64
+}
+
+func insertArrowOffsetHashes[T byteArrayArrowHashOffset](b BloomFilterBuilder,
values []byte, offsets []T, numValid int64, validBits []byte, validBitsOffset
int64) {
+ if len(offsets) < 2 || numValid == 0 {
+ return
+ }
+
+ h := b.Hasher()
+ var (
+ byteBatch [bloomFilterHashBatchSize][]byte
+ hashBatch [bloomFilterHashBatchSize]uint64
+ batchSize int
+ )
+ flush := func() {
+ b.InsertBulk(sum64s(h, byteBatch[:batchSize],
hashBatch[:batchSize]))
+ batchSize = 0
+ }
+ if validBits == nil {
+ for i := 0; i < len(offsets)-1; i++ {
+ byteBatch[batchSize] =
values[int(offsets[i]):int(offsets[i+1])]
+ batchSize++
+ if batchSize == len(byteBatch) {
+ flush()
+ }
+ }
+ if batchSize != 0 {
+ flush()
+ }
+ return
+ }
+
+ visit := func(pos, length int64) {
+ for i := pos; i < pos+length; i++ {
+ byteBatch[batchSize] =
values[int(offsets[i]):int(offsets[i+1])]
+ batchSize++
+ if batchSize == len(byteBatch) {
+ flush()
+ }
+ }
+ }
+
+ bitutils.VisitSetBitRunsNoErr(validBits, validBitsOffset,
int64(len(offsets)-1), visit)
+ if batchSize != 0 {
+ flush()
+ }
+}
+
+// InsertArrowOffsetHashes inserts hashes for values described by 32-bit Arrow
+// offsets without materializing parquet.ByteArray values.
+func InsertArrowOffsetHashes(b BloomFilterBuilder, values []byte, offsets
[]int32) {
+ numValues := 0
+ if len(offsets) > 0 {
+ numValues = len(offsets) - 1
+ }
+ insertArrowOffsetHashes(b, values, offsets, int64(numValues), nil, 0)
+}
+
+// InsertArrowOffsetHashes64 inserts hashes for values described by 64-bit
Arrow
+// offsets without materializing parquet.ByteArray values.
+func InsertArrowOffsetHashes64(b BloomFilterBuilder, values []byte, offsets
[]int64) {
+ numValues := 0
+ if len(offsets) > 0 {
+ numValues = len(offsets) - 1
+ }
+ insertArrowOffsetHashes(b, values, offsets, int64(numValues), nil, 0)
+}
+
+// InsertSpacedArrowOffsetHashes inserts hashes for valid values described by
+// spaced 32-bit Arrow offsets without materializing parquet.ByteArray values.
+func InsertSpacedArrowOffsetHashes(b BloomFilterBuilder, numValid int64,
values []byte, offsets []int32, validBits []byte, validBitsOffset int64) {
+ insertArrowOffsetHashes(b, values, offsets, numValid, validBits,
validBitsOffset)
+}
+
+// InsertSpacedArrowOffsetHashes64 inserts hashes for valid values described by
+// spaced 64-bit Arrow offsets without materializing parquet.ByteArray values.
+func InsertSpacedArrowOffsetHashes64(b BloomFilterBuilder, numValid int64,
values []byte, offsets []int64, validBits []byte, validBitsOffset int64) {
+ insertArrowOffsetHashes(b, values, offsets, numValid, validBits,
validBitsOffset)
+}
diff --git a/parquet/metadata/byte_array_arrow_offsets.go
b/parquet/metadata/byte_array_arrow_offsets.go
new file mode 100644
index 00000000..dbb4055d
--- /dev/null
+++ b/parquet/metadata/byte_array_arrow_offsets.go
@@ -0,0 +1,89 @@
+// 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 metadata
+
+import (
+ "github.com/apache/arrow-go/v18/internal/bitutils"
+ "github.com/apache/arrow-go/v18/parquet"
+)
+
+type byteArrayArrowOffset interface {
+ ~int32 | ~int64
+}
+
+func updateByteArrayStatisticsFromArrowOffsets[T byteArrayArrowOffset](s
*ByteArrayStatistics, values []byte, offsets []T, numNull int64, validBits
[]byte, validBitsOffset int64, spaced bool) {
+ numValues := 0
+ if len(offsets) > 0 {
+ numValues = len(offsets) - 1
+ }
+
+ s.IncNulls(numNull)
+ if spaced {
+ s.nvalues += int64(numValues) - numNull
+ } else {
+ s.nvalues += int64(numValues)
+ }
+ if numValues == 0 {
+ return
+ }
+
+ min := s.defaultMin()
+ max := s.defaultMax()
+ if validBits == nil {
+ for i := 0; i < numValues; i++ {
+ value :=
parquet.ByteArray(values[int(offsets[i]):int(offsets[i+1])])
+ min = s.minval(min, value)
+ max = s.maxval(max, value)
+ }
+ s.SetMinMax(min, max)
+ return
+ }
+
+ visit := func(pos, length int64) {
+ for i := pos; i < pos+length; i++ {
+ value :=
parquet.ByteArray(values[int(offsets[i]):int(offsets[i+1])])
+ min = s.minval(min, value)
+ max = s.maxval(max, value)
+ }
+ }
+ bitutils.VisitSetBitRunsNoErr(validBits, validBitsOffset,
int64(numValues), visit)
+ s.SetMinMax(min, max)
+}
+
+// UpdateFromArrowOffsets updates byte-array statistics from an Arrow value
buffer
+// and 32-bit offsets without materializing parquet.ByteArray values.
+func (s *ByteArrayStatistics) UpdateFromArrowOffsets(values []byte, offsets
[]int32, numNull int64) {
+ updateByteArrayStatisticsFromArrowOffsets(s, values, offsets, numNull,
nil, 0, false)
+}
+
+// UpdateFromArrowOffsets64 updates byte-array statistics from an Arrow value
buffer
+// and 64-bit offsets without materializing parquet.ByteArray values.
+func (s *ByteArrayStatistics) UpdateFromArrowOffsets64(values []byte, offsets
[]int64, numNull int64) {
+ updateByteArrayStatisticsFromArrowOffsets(s, values, offsets, numNull,
nil, 0, false)
+}
+
+// UpdateFromArrowOffsetsSpaced updates byte-array statistics from spaced Arrow
+// values using a validity bitmap and 32-bit offsets.
+func (s *ByteArrayStatistics) UpdateFromArrowOffsetsSpaced(values []byte,
offsets []int32, validBits []byte, validBitsOffset, numNull int64) {
+ updateByteArrayStatisticsFromArrowOffsets(s, values, offsets, numNull,
validBits, validBitsOffset, true)
+}
+
+// UpdateFromArrowOffsetsSpaced64 updates byte-array statistics from spaced
Arrow
+// values using a validity bitmap and 64-bit offsets.
+func (s *ByteArrayStatistics) UpdateFromArrowOffsetsSpaced64(values []byte,
offsets []int64, validBits []byte, validBitsOffset, numNull int64) {
+ updateByteArrayStatisticsFromArrowOffsets(s, values, offsets, numNull,
validBits, validBitsOffset, true)
+}
diff --git a/parquet/metadata/byte_array_arrow_offsets_test.go
b/parquet/metadata/byte_array_arrow_offsets_test.go
new file mode 100644
index 00000000..43689178
--- /dev/null
+++ b/parquet/metadata/byte_array_arrow_offsets_test.go
@@ -0,0 +1,114 @@
+// 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 metadata
+
+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/apache/arrow-go/v18/parquet/schema"
+ "github.com/stretchr/testify/require"
+)
+
+func byteArrayArrowMetadataInput() ([]parquet.ByteArray, []byte, []int32,
[]int64) {
+ values := []parquet.ByteArray{
+ []byte("alpha"),
+ []byte("beta"),
+ {},
+ []byte("gamma"),
+ []byte("delta"),
+ {},
+ }
+ data := make([]byte, 0, 32)
+ offsets32 := []int32{0}
+ offsets64 := []int64{0}
+ for _, value := range values {
+ data = append(data, value...)
+ offsets32 = append(offsets32, int32(len(data)))
+ offsets64 = append(offsets64, int64(len(data)))
+ }
+ return values, data, offsets32, offsets64
+}
+
+func TestByteArrayStatisticsArrowOffsetsMatchByteArrayInput(t *testing.T) {
+ values, data, offsets32, offsets64 := byteArrayArrowMetadataInput()
+ validBits := make([]byte, bitutil.BytesForBits(int64(len(values)+4)))
+ validBitsOffset := int64(2)
+ for _, index := range []int{0, 1, 3, 4} {
+ bitutil.SetBit(validBits, int(validBitsOffset)+index)
+ }
+
+ column := schema.NewColumn(schema.NewByteArrayNode("value",
parquet.Repetitions.Optional, -1), 1, 0)
+ want := NewByteArrayStatistics(column, memory.DefaultAllocator)
+ wantSpaced := NewByteArrayStatistics(column, memory.DefaultAllocator)
+ want.Update(values, 0)
+ wantSpaced.UpdateSpaced(values, validBits, validBitsOffset, 2)
+
+ for _, width := range []string{"int32", "int64"} {
+ t.Run(width, func(t *testing.T) {
+ got := NewByteArrayStatistics(column,
memory.DefaultAllocator)
+ gotSpaced := NewByteArrayStatistics(column,
memory.DefaultAllocator)
+ if width == "int32" {
+ got.UpdateFromArrowOffsets(data, offsets32, 0)
+ gotSpaced.UpdateFromArrowOffsetsSpaced(data,
offsets32, validBits, validBitsOffset, 2)
+ } else {
+ got.UpdateFromArrowOffsets64(data, offsets64, 0)
+ gotSpaced.UpdateFromArrowOffsetsSpaced64(data,
offsets64, validBits, validBitsOffset, 2)
+ }
+ require.Equal(t, want.Min(), got.Min())
+ require.Equal(t, want.Max(), got.Max())
+ require.Equal(t, want.NumValues(), got.NumValues())
+ require.Equal(t, want.NullCount(), got.NullCount())
+ require.Equal(t, wantSpaced.Min(), gotSpaced.Min())
+ require.Equal(t, wantSpaced.Max(), gotSpaced.Max())
+ require.Equal(t, wantSpaced.NumValues(),
gotSpaced.NumValues())
+ require.Equal(t, wantSpaced.NullCount(),
gotSpaced.NullCount())
+ })
+ }
+}
+
+func TestByteArrayBloomHashesArrowOffsetsMatchByteArrayInput(t *testing.T) {
+ values, data, offsets32, offsets64 := byteArrayArrowMetadataInput()
+ validBits := make([]byte, bitutil.BytesForBits(int64(len(values)+4)))
+ validBitsOffset := int64(2)
+ for _, index := range []int{0, 1, 3, 4} {
+ bitutil.SetBit(validBits, int(validBitsOffset)+index)
+ }
+
+ want := newBatchRecordingBloomFilter(xxhasher{})
+ InsertHashes(want, values)
+ wantSpaced := newBatchRecordingBloomFilter(xxhasher{})
+ InsertSpacedHashes(wantSpaced, 4, values, validBits, validBitsOffset)
+
+ for _, width := range []string{"int32", "int64"} {
+ t.Run(width, func(t *testing.T) {
+ got := newBatchRecordingBloomFilter(xxhasher{})
+ gotSpaced := newBatchRecordingBloomFilter(xxhasher{})
+ if width == "int32" {
+ InsertArrowOffsetHashes(got, data, offsets32)
+ InsertSpacedArrowOffsetHashes(gotSpaced, 4,
data, offsets32, validBits, validBitsOffset)
+ } else {
+ InsertArrowOffsetHashes64(got, data, offsets64)
+ InsertSpacedArrowOffsetHashes64(gotSpaced, 4,
data, offsets64, validBits, validBitsOffset)
+ }
+ require.Equal(t, flattenHashBatches(want.batches),
flattenHashBatches(got.batches))
+ require.Equal(t,
flattenHashBatches(wantSpaced.batches), flattenHashBatches(gotSpaced.batches))
+ })
+ }
+}
diff --git a/parquet/pqarrow/binary_offsets_bench_test.go
b/parquet/pqarrow/binary_offsets_bench_test.go
new file mode 100644
index 00000000..bdf6eb60
--- /dev/null
+++ b/parquet/pqarrow/binary_offsets_bench_test.go
@@ -0,0 +1,75 @@
+// 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 pqarrow_test
+
+import (
+ "bytes"
+ "strconv"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/compress"
+ "github.com/apache/arrow-go/v18/parquet/pqarrow"
+)
+
+func benchmarkBinaryTable(mem memory.Allocator, n int) (arrow.Table, int64) {
+ builder := array.NewStringBuilder(mem)
+ builder.Reserve(n)
+ var inputBytes int64
+ for i := 0; i < n; i++ {
+ value := append([]byte("partition/"), strconv.AppendInt(nil,
int64(i), 10)...)
+ builder.Append(string(value))
+ inputBytes += int64(len(value))
+ }
+ arr := builder.NewArray()
+ builder.Release()
+
+ sch := arrow.NewSchema([]arrow.Field{{Name: "value", Type:
arrow.BinaryTypes.String}}, nil)
+ col := arrow.NewColumnFromArr(sch.Field(0), arr)
+ arr.Release()
+ tbl := array.NewTable(sch, []arrow.Column{col}, int64(n))
+ col.Release()
+ return tbl, inputBytes
+}
+
+func BenchmarkWriteArrowBinaryOffsets(b *testing.B) {
+ const n = 64 * 1024
+ mem := memory.DefaultAllocator
+ tbl, inputBytes := benchmarkBinaryTable(mem, n)
+ defer tbl.Release()
+
+ for _, stats := range []bool{false, true} {
+ b.Run("stats="+strconv.FormatBool(stats), func(b *testing.B) {
+ props := parquet.NewWriterProperties(
+ parquet.WithDictionaryDefault(false),
+ parquet.WithStats(stats),
+
parquet.WithCompression(compress.Codecs.Uncompressed),
+ )
+ b.SetBytes(inputBytes)
+ b.ReportAllocs()
+ for b.Loop() {
+ var buf bytes.Buffer
+ if err := pqarrow.WriteTable(tbl, &buf,
int64(n), props, pqarrow.DefaultWriterProps()); err != nil {
+ b.Fatal(err)
+ }
+ }
+ })
+ }
+}
diff --git a/parquet/pqarrow/binary_offsets_test.go
b/parquet/pqarrow/binary_offsets_test.go
new file mode 100644
index 00000000..a3a6d9ce
--- /dev/null
+++ b/parquet/pqarrow/binary_offsets_test.go
@@ -0,0 +1,234 @@
+// 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 pqarrow_test
+
+import (
+ "bytes"
+ "fmt"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/file"
+ "github.com/apache/arrow-go/v18/parquet/pqarrow"
+ "github.com/stretchr/testify/require"
+)
+
+func makeArrowBinaryOffsetsTables(t *testing.T) []struct {
+ name string
+ table arrow.Table
+} {
+ t.Helper()
+ mem := memory.DefaultAllocator
+ result := make([]struct {
+ name string
+ table arrow.Table
+ }, 0, 3)
+
+ binaryBuilder := array.NewBinaryBuilder(mem, arrow.BinaryTypes.Binary)
+ for _, value := range [][]byte{[]byte("alpha"), []byte("beta"), nil,
[]byte("gamma"), {}, []byte("delta")} {
+ if value == nil {
+ binaryBuilder.AppendNull()
+ } else {
+ binaryBuilder.Append(value)
+ }
+ }
+ binaryArray := binaryBuilder.NewArray()
+ binaryBuilder.Release()
+ binarySchema := arrow.NewSchema([]arrow.Field{{Name: "value", Type:
arrow.BinaryTypes.Binary, Nullable: true}}, nil)
+ binaryColumn := arrow.NewColumnFromArr(binarySchema.Field(0),
binaryArray)
+ binaryArray.Release()
+ result = append(result, struct {
+ name string
+ table arrow.Table
+ }{"binary", array.NewTable(binarySchema, []arrow.Column{binaryColumn},
6)})
+ binaryColumn.Release()
+
+ stringBuilder := array.NewStringBuilder(mem)
+ for _, value := range []string{"alpha", "beta", "", "gamma", "delta",
"epsilon"} {
+ stringBuilder.Append(value)
+ }
+ stringArray := stringBuilder.NewArray()
+ stringBuilder.Release()
+ stringSchema := arrow.NewSchema([]arrow.Field{{Name: "value", Type:
arrow.BinaryTypes.String, Nullable: false}}, nil)
+ stringColumn := arrow.NewColumnFromArr(stringSchema.Field(0),
stringArray)
+ stringArray.Release()
+ result = append(result, struct {
+ name string
+ table arrow.Table
+ }{"string", array.NewTable(stringSchema, []arrow.Column{stringColumn},
6)})
+ stringColumn.Release()
+
+ largeStringBuilder := array.NewLargeStringBuilder(mem)
+ for _, value := range []string{"alpha", "beta", "", "gamma", "delta",
"epsilon"} {
+ largeStringBuilder.Append(value)
+ }
+ largeStringArray := largeStringBuilder.NewArray()
+ largeStringBuilder.Release()
+ largeStringSchema := arrow.NewSchema([]arrow.Field{{Name: "value",
Type: arrow.BinaryTypes.LargeString, Nullable: false}}, nil)
+ largeStringColumn := arrow.NewColumnFromArr(largeStringSchema.Field(0),
largeStringArray)
+ largeStringArray.Release()
+ result = append(result, struct {
+ name string
+ table arrow.Table
+ }{"large-string", array.NewTable(largeStringSchema,
[]arrow.Column{largeStringColumn}, 6)})
+ largeStringColumn.Release()
+
+ return result
+}
+
+func TestWriteArrowBinaryOffsets(t *testing.T) {
+ encodings := []parquet.Encoding{
+ parquet.Encodings.Plain,
+ parquet.Encodings.DeltaLengthByteArray,
+ parquet.Encodings.DeltaByteArray,
+ }
+
+ for _, input := range makeArrowBinaryOffsetsTables(t) {
+ input := input
+ t.Run(input.name, func(t *testing.T) {
+ defer input.table.Release()
+ for _, encoding := range encodings {
+ t.Run(encoding.String(), func(t *testing.T) {
+ writerProps :=
parquet.NewWriterProperties(
+
parquet.WithDictionaryDefault(false),
+
parquet.WithEncodingFor("value", encoding),
+ parquet.WithStats(true),
+ parquet.WithBatchSize(2),
+ parquet.WithDataPageSize(32),
+
parquet.WithPageIndexEnabled(true),
+
parquet.WithBloomFilterEnabledFor("value", true),
+
parquet.WithBloomFilterNDVFor("value", input.table.NumRows()),
+ )
+ data := writeParquetTable(t,
input.table, input.table.NumRows(), writerProps)
+ got := readParquetTable(t, data,
pqarrow.ArrowReadProperties{})
+ defer got.Release()
+ if input.name == "large-string" {
+ wantArr :=
input.table.Column(0).Data().Chunk(0)
+ gotArr :=
got.Column(0).Data().Chunk(0)
+ for i := 0; i < wantArr.Len();
i++ {
+ require.Equal(t,
wantArr.IsNull(i), gotArr.IsNull(i))
+ if !wantArr.IsNull(i) {
+
require.Equal(t, binaryOffsetValue(wantArr, i), binaryOffsetValue(gotArr, i))
+ }
+ }
+ } else {
+ assertTableColumnsEqual(t,
input.table, got)
+ }
+ })
+ }
+ })
+ }
+}
+
+func binaryOffsetValue(arr arrow.Array, index int) []byte {
+ switch arr := arr.(type) {
+ case *array.Binary:
+ return arr.Value(index)
+ case *array.LargeBinary:
+ return arr.Value(index)
+ case *array.String:
+ return []byte(arr.Value(index))
+ case *array.LargeString:
+ return []byte(arr.Value(index))
+ default:
+ panic("unexpected binary array type")
+ }
+}
+
+func TestWriteArrowBinaryOffsetsWithSlice(t *testing.T) {
+ mem := memory.DefaultAllocator
+ builder := array.NewStringBuilder(mem)
+ for _, value := range []string{"zero", "one", "two", "three", "four",
"five"} {
+ builder.Append(value)
+ }
+ full := builder.NewArray()
+ builder.Release()
+ defer full.Release()
+
+ sliced := array.NewSlice(full, 1, 5)
+ defer sliced.Release()
+ field := arrow.Field{Name: "value", Type: arrow.BinaryTypes.String,
Nullable: false}
+ schema := arrow.NewSchema([]arrow.Field{field}, nil)
+ column := arrow.NewColumnFromArr(field, sliced)
+ table := array.NewTable(schema, []arrow.Column{column},
int64(sliced.Len()))
+ column.Release()
+ defer table.Release()
+
+ writerProps := parquet.NewWriterProperties(
+ parquet.WithDictionaryDefault(false),
+ parquet.WithBatchSize(2),
+ parquet.WithEncodingFor("value", parquet.Encodings.Plain),
+ )
+ data := writeParquetTable(t, table, table.NumRows(), writerProps)
+ got := readParquetTable(t, data, pqarrow.ArrowReadProperties{})
+ defer got.Release()
+
+ wantArr := table.Column(0).Data().Chunk(0)
+ gotArr := got.Column(0).Data().Chunk(0)
+ for i := 0; i < wantArr.Len(); i++ {
+ require.Equal(t, binaryOffsetValue(wantArr, i),
binaryOffsetValue(gotArr, i))
+ }
+}
+
+func TestWriteArrowBinaryOffsetsNestedNullPageIndex(t *testing.T) {
+ for _, dtype := range []arrow.DataType{arrow.BinaryTypes.Binary,
arrow.BinaryTypes.LargeBinary,
+ arrow.BinaryTypes.String, arrow.BinaryTypes.LargeString} {
+ for _, pageVersion := range
[]parquet.DataPageVersion{parquet.DataPageV1, parquet.DataPageV2} {
+ for _, enc := range
[]parquet.Encoding{parquet.Encodings.Plain,
+ parquet.Encodings.DeltaLengthByteArray,
parquet.Encodings.DeltaByteArray} {
+ t.Run(fmt.Sprintf("%s/page-%d/%s", dtype,
pageVersion, enc), func(t *testing.T) {
+ builder :=
array.NewListBuilder(memory.DefaultAllocator, dtype)
+ defer builder.Release()
+ builder.Append(true)
+ value := "aaa"
+ if dtype.ID() == arrow.BINARY ||
dtype.ID() == arrow.LARGE_BINARY {
+ value = "YWFh"
+ }
+ require.NoError(t,
builder.ValueBuilder().AppendValueFromString(value))
+ builder.ValueBuilder().AppendNull()
+ builder.AppendNull()
+ arr := builder.NewListArray()
+ defer arr.Release()
+ field := arrow.Field{Name: "value",
Type: arr.DataType(), Nullable: true}
+ column := arrow.NewColumnFromArr(field,
arr)
+ defer column.Release()
+ tbl :=
array.NewTable(arrow.NewSchema([]arrow.Field{field}, nil),
[]arrow.Column{column}, int64(arr.Len()))
+ defer tbl.Release()
+ props :=
parquet.NewWriterProperties(parquet.WithDictionaryDefault(false),
+
parquet.WithDataPageVersion(pageVersion), parquet.WithEncoding(enc),
+
parquet.WithPageIndexEnabled(true))
+ data := writeParquetTable(t, tbl,
tbl.NumRows(), props)
+ reader, err :=
file.NewParquetReader(bytes.NewReader(data))
+ require.NoError(t, err)
+ defer reader.Close()
+ rgIndex, err :=
reader.GetPageIndexReader().RowGroup(0)
+ require.NoError(t, err)
+ index, err := rgIndex.GetColumnIndex(0)
+ require.NoError(t, err)
+ require.NotNil(t, index)
+ require.Equal(t, []bool{false},
index.GetNullPages())
+ require.Equal(t, []int64{2},
index.GetNullCounts())
+ require.Equal(t,
[][]byte{[]byte("aaa")}, index.GetMinValues())
+ require.Equal(t,
[][]byte{[]byte("aaa")}, index.GetMaxValues())
+ })
+ }
+ }
+ }
+}
diff --git a/parquet/pqarrow/encode_arrow.go b/parquet/pqarrow/encode_arrow.go
index 3824aa98..dd865c02 100644
--- a/parquet/pqarrow/encode_arrow.go
+++ b/parquet/pqarrow/encode_arrow.go
@@ -661,6 +661,30 @@ func writeDenseArrow(ctx *arrowWriteContext, cw
file.ColumnChunkWriter, leafArr
valueBuf = buffer.Bytes()
}
+ if wr.SupportsArrowOffsets() {
+ switch leafArr.DataType().ID() {
+ case arrow.BINARY, arrow.STRING:
+ offsets := leafArr.(binaryarr).ValueOffsets()
+ if !maybeParentNulls && noNulls {
+ _, err = wr.WriteBatchArrow(valueBuf,
offsets, defLevels, repLevels)
+ } else {
+ _, err =
wr.WriteBatchSpacedArrow(valueBuf, offsets, defLevels, repLevels,
+ leafArr.NullBitmapBytes(),
int64(leafArr.Data().Offset()))
+ }
+ case arrow.LARGE_BINARY, arrow.LARGE_STRING:
+ offsets := leafArr.(binary64arr).ValueOffsets()
+ if !maybeParentNulls && noNulls {
+ _, err = wr.WriteBatchArrow64(valueBuf,
offsets, defLevels, repLevels)
+ } else {
+ _, err =
wr.WriteBatchSpacedArrow64(valueBuf, offsets, defLevels, repLevels,
+ leafArr.NullBitmapBytes(),
int64(leafArr.Data().Offset()))
+ }
+ default:
+ return fmt.Errorf("%w: invalid column type to
write to ByteArray: %s", arrow.ErrInvalid, leafArr.DataType().Name())
+ }
+ return err
+ }
+
data := make([]parquet.ByteArray, leafArr.Len())
switch leafArr.DataType().ID() {
case arrow.BINARY, arrow.STRING:
