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 8ed90de5 fix(parquet/pqarrow): propagate level builder errors (#1049)
8ed90de5 is described below
commit 8ed90de5da00a81f8a78493dbb1aba7f2758e252
Author: Minh Vu <[email protected]>
AuthorDate: Wed Aug 5 19:38:58 2026 +0200
fix(parquet/pqarrow): propagate level builder errors (#1049)
## Summary
- return errors from multipath level builder construction instead of
reporting a successful column write
- release builders created for earlier chunks when a later chunk fails
- release partially initialized builder state on visitor errors
- leave the physical column position unchanged after builder
construction fails
## Testing
- `go test ./parquet/pqarrow`
The regression test uses a valid dictionary chunk followed by a
dictionary containing a null, which reaches the previously swallowed
error path, verifies checked allocator cleanup, and successfully retries
the same physical column with supported data.
---
parquet/pqarrow/encode_arrow.go | 14 +++++++++--
parquet/pqarrow/encode_dictionary_test.go | 42 +++++++++++++++++++++++++++++++
parquet/pqarrow/path_builder.go | 1 +
3 files changed, 55 insertions(+), 2 deletions(-)
diff --git a/parquet/pqarrow/encode_arrow.go b/parquet/pqarrow/encode_arrow.go
index c7f089c2..73b52a34 100644
--- a/parquet/pqarrow/encode_arrow.go
+++ b/parquet/pqarrow/encode_arrow.go
@@ -126,6 +126,11 @@ func newArrowColumnWriter(data *arrow.Chunked, offset,
size int64, manifest *Sch
isNullable = nullableRoot(manifest, schemaField)
builders := make([]*multipathLevelBuilder, 0)
+ releaseBuilders := func() {
+ for _, bldr := range builders {
+ bldr.Release()
+ }
+ }
for values < size {
chunk := data.Chunk(chunkIdx)
available := int64(chunk.Len() - int(chunkOffset))
@@ -134,17 +139,22 @@ func newArrowColumnWriter(data *arrow.Chunked, offset,
size int64, manifest *Sch
// the chunk offset will be 0 here except for possibly the
first chunk
// because of the above advancing logic
arrToWrite := array.NewSlice(chunk, chunkOffset,
chunkOffset+chunkWriteSize)
- defer arrToWrite.Release()
if arrToWrite.Len() > 0 {
bldr, err := newMultipathLevelBuilder(arrToWrite,
isNullable)
+ arrToWrite.Release()
if err != nil {
- return arrowColumnWriter{}, nil
+ releaseBuilders()
+ return arrowColumnWriter{}, err
}
if leafCount != bldr.leafCount() {
+ bldr.Release()
+ releaseBuilders()
return arrowColumnWriter{}, fmt.Errorf("data
type leaf_count != builder leaf_count: %d - %d", leafCount, bldr.leafCount())
}
builders = append(builders, bldr)
+ } else {
+ arrToWrite.Release()
}
if chunkWriteSize == available {
diff --git a/parquet/pqarrow/encode_dictionary_test.go
b/parquet/pqarrow/encode_dictionary_test.go
index e92b2587..b027f7f0 100644
--- a/parquet/pqarrow/encode_dictionary_test.go
+++ b/parquet/pqarrow/encode_dictionary_test.go
@@ -21,6 +21,7 @@ package pqarrow_test
import (
"bytes"
"context"
+ "errors"
"fmt"
"math"
"strings"
@@ -39,6 +40,47 @@ import (
"github.com/stretchr/testify/suite"
)
+func TestWriteColumnChunkedPropagatesLevelBuilderError(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ dictType := &arrow.DictionaryType{IndexType:
arrow.PrimitiveTypes.Int32, ValueType: arrow.BinaryTypes.String}
+ validIndices, _, err := array.FromJSON(mem, arrow.PrimitiveTypes.Int32,
strings.NewReader(`[0]`))
+ require.NoError(t, err)
+ defer validIndices.Release()
+ validDictionary, _, err := array.FromJSON(mem,
arrow.BinaryTypes.String, strings.NewReader(`["valid"]`))
+ require.NoError(t, err)
+ defer validDictionary.Release()
+ valid := array.NewDictionaryArray(dictType, validIndices,
validDictionary)
+ defer valid.Release()
+
+ invalidIndices, _, err := array.FromJSON(mem,
arrow.PrimitiveTypes.Int32, strings.NewReader(`[0]`))
+ require.NoError(t, err)
+ defer invalidIndices.Release()
+ invalidDictionary, _, err := array.FromJSON(mem,
arrow.BinaryTypes.String, strings.NewReader(`[null]`))
+ require.NoError(t, err)
+ defer invalidDictionary.Release()
+ invalid := array.NewDictionaryArray(dictType, invalidIndices,
invalidDictionary)
+ defer invalid.Release()
+
+ values := arrow.NewChunked(dictType, []arrow.Array{valid, invalid})
+ defer values.Release()
+ schema := arrow.NewSchema([]arrow.Field{{Name: "values", Type:
dictType, Nullable: true}}, nil)
+
+ var output bytes.Buffer
+ writer, err := pqarrow.NewFileWriter(schema, &output,
+ parquet.NewWriterProperties(parquet.WithAllocator(mem)),
+ pqarrow.NewArrowWriterProperties(pqarrow.WithAllocator(mem)))
+ require.NoError(t, err)
+ require.NoError(t, writer.NewRowGroupChecked())
+
+ err = writer.WriteColumnChunked(values, 0, int64(values.Len()))
+ require.Error(t, err)
+ assert.True(t, errors.Is(err, arrow.ErrNotImplemented))
+ require.NoError(t, writer.WriteColumnData(valid))
+ require.NoError(t, writer.Close())
+}
+
func (ps *ParquetIOTestSuite) TestSingleColumnOptionalDictionaryWrite() {
for _, dt := range fullTypeList {
// skip tests for bool as we don't do dictionaries for it
diff --git a/parquet/pqarrow/path_builder.go b/parquet/pqarrow/path_builder.go
index 991710b3..b10318a4 100644
--- a/parquet/pqarrow/path_builder.go
+++ b/parquet/pqarrow/path_builder.go
@@ -536,6 +536,7 @@ func newMultipathLevelBuilder(arr arrow.Array,
fieldNullable bool) (*multipathLe
builder: pathBuilder{nullableInParent: fieldNullable, paths:
make([]pathInfo, 0), refCount: utils.NewRefCount(1)},
}
if err := ret.builder.Visit(arr); err != nil {
+ ret.builder.Release()
return nil, err
}
arr.Data().Retain()
