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 229824a3 fix(parquet/pqarrow): release extension arrays after chunking
(#1060)
229824a3 is described below
commit 229824a32ef23849346a58a7accba3697ba0344b
Author: Minh Vu <[email protected]>
AuthorDate: Fri Aug 7 17:37:31 2026 +0200
fix(parquet/pqarrow): release extension arrays after chunking (#1060)
### Rationale for this change
The extension reader creates arrays and passes them to NewChunked, which
retains each chunk. The original references were not released, and
partially built chunks could also leak when a later chunk failed.
### What changes are included in this PR?
Release each extension array after NewChunked takes ownership, and
release already-built chunks while unwinding an error.
### Are these changes tested?
- `go test ./parquet/pqarrow -run
TestExtensionReaderBuildArrayReleasesPartialChunksOnPanic`
### Are there any user-facing changes?
No API changes. This corrects the reported behavior while preserving the
existing ownership and compatibility contracts.
---
parquet/pqarrow/file_reader.go | 7 ++
parquet/pqarrow/file_reader_extension_test.go | 158 ++++++++++++++++++++++++++
2 files changed, 165 insertions(+)
diff --git a/parquet/pqarrow/file_reader.go b/parquet/pqarrow/file_reader.go
index 96af9f77..37b847fc 100644
--- a/parquet/pqarrow/file_reader.go
+++ b/parquet/pqarrow/file_reader.go
@@ -135,6 +135,13 @@ func (er *extensionReader) BuildArray(boundedLen int64)
(*arrow.Chunked, error)
extType := er.fieldWithExt.Type.(arrow.ExtensionType)
newChunks := make([]arrow.Array, len(chkd.Chunks()))
+ defer func() {
+ for _, c := range newChunks {
+ if c != nil {
+ c.Release()
+ }
+ }
+ }()
for i, c := range chkd.Chunks() {
newChunks[i] = array.NewExtensionArrayWithStorage(extType, c)
}
diff --git a/parquet/pqarrow/file_reader_extension_test.go
b/parquet/pqarrow/file_reader_extension_test.go
new file mode 100644
index 00000000..76429e8a
--- /dev/null
+++ b/parquet/pqarrow/file_reader_extension_test.go
@@ -0,0 +1,158 @@
+// 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
+
+import (
+ "reflect"
+ "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/stretchr/testify/require"
+)
+
+type mismatchingExtensionType struct {
+ arrow.ExtensionBase
+ storageTypeCalls int
+}
+
+type mismatchingExtensionArray struct {
+ array.ExtensionArrayBase
+}
+
+type stableExtensionType struct {
+ arrow.ExtensionBase
+}
+
+type stableExtensionArray struct {
+ array.ExtensionArrayBase
+}
+
+func (*stableExtensionType) StorageType() arrow.DataType { return
arrow.PrimitiveTypes.Int32 }
+
+func (*stableExtensionType) ArrayType() reflect.Type {
+ return reflect.TypeFor[stableExtensionArray]()
+}
+
+func (*stableExtensionType) ExtensionName() string { return "test.stable" }
+
+func (*stableExtensionType) ExtensionEquals(other arrow.ExtensionType) bool {
+ _, ok := other.(*stableExtensionType)
+ return ok
+}
+
+func (*stableExtensionType) Serialize() string { return "" }
+
+func (*stableExtensionType) Deserialize(arrow.DataType, string)
(arrow.ExtensionType, error) {
+ return &stableExtensionType{
+ ExtensionBase: arrow.ExtensionBase{Storage:
arrow.PrimitiveTypes.Int32},
+ }, nil
+}
+
+func (t *mismatchingExtensionType) StorageType() arrow.DataType {
+ t.storageTypeCalls++
+ if t.storageTypeCalls > 2 {
+ return arrow.PrimitiveTypes.Int64
+ }
+ return arrow.PrimitiveTypes.Int32
+}
+
+func (*mismatchingExtensionType) ArrayType() reflect.Type {
+ return reflect.TypeFor[mismatchingExtensionArray]()
+}
+
+func (*mismatchingExtensionType) ExtensionName() string { return
"test.mismatching" }
+
+func (*mismatchingExtensionType) ExtensionEquals(other arrow.ExtensionType)
bool {
+ _, ok := other.(*mismatchingExtensionType)
+ return ok
+}
+
+func (*mismatchingExtensionType) Serialize() string { return "" }
+
+func (*mismatchingExtensionType) Deserialize(arrow.DataType, string)
(arrow.ExtensionType, error) {
+ return &mismatchingExtensionType{
+ ExtensionBase: arrow.ExtensionBase{Storage:
arrow.PrimitiveTypes.Int32},
+ }, nil
+}
+
+type chunkedColumnReader struct {
+ colReaderImpl
+ chunks *arrow.Chunked
+}
+
+func (r *chunkedColumnReader) BuildArray(int64) (*arrow.Chunked, error) {
+ return r.chunks, nil
+}
+
+func TestExtensionReaderBuildArrayReleasesPartialChunksOnPanic(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.NewGoAllocator())
+ defer mem.AssertSize(t, 0)
+
+ b := array.NewInt32Builder(mem)
+ b.Append(1)
+ first := b.NewInt32Array()
+ b.Append(2)
+ second := b.NewInt32Array()
+ b.Release()
+
+ chunks := arrow.NewChunked(arrow.PrimitiveTypes.Int32,
[]arrow.Array{first, second})
+ first.Release()
+ second.Release()
+
+ extType := &mismatchingExtensionType{
+ ExtensionBase: arrow.ExtensionBase{Storage:
arrow.PrimitiveTypes.Int32},
+ }
+ r := extensionReader{
+ colReaderImpl: &chunkedColumnReader{chunks: chunks},
+ fieldWithExt: arrow.Field{Name: "extension", Type: extType},
+ }
+
+ require.Panics(t, func() {
+ _, _ = r.BuildArray(0)
+ })
+ require.Zero(t, mem.CurrentAlloc())
+}
+
+func TestExtensionReaderBuildArrayReleasesChunks(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.NewGoAllocator())
+ defer mem.AssertSize(t, 0)
+
+ b := array.NewInt32Builder(mem)
+ b.Append(1)
+ first := b.NewInt32Array()
+ b.Append(2)
+ second := b.NewInt32Array()
+ b.Release()
+
+ chunks := arrow.NewChunked(arrow.PrimitiveTypes.Int32,
[]arrow.Array{first, second})
+ first.Release()
+ second.Release()
+
+ extType := &stableExtensionType{
+ ExtensionBase: arrow.ExtensionBase{Storage:
arrow.PrimitiveTypes.Int32},
+ }
+ r := extensionReader{
+ colReaderImpl: &chunkedColumnReader{chunks: chunks},
+ fieldWithExt: arrow.Field{Name: "extension", Type: extType},
+ }
+
+ out, err := r.BuildArray(0)
+ require.NoError(t, err)
+ out.Release()
+}
