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 153ec5bf fix(parquet/pqarrow): release partial readers when building a
field reader fails (#1435)
153ec5bf is described below
commit 153ec5bf362c2c6322fbf1f3d5f87ecb1ffe4746
Author: Steve O'Brien <[email protected]>
AuthorDate: Mon Nov 2 13:40:39 2026 -0500
fix(parquet/pqarrow): release partial readers when building a field reader
fails (#1435)
### Rationale for this change
Two error paths while building a field reader leave readers unreleased:
- In `getReader`, when one child of a struct fails to build, the readers
already built for its siblings are dropped without `Release`. With
`errgroup.WithContext` and `SetLimit(1)`, the children after the failing
one are still built too, since `getReader` does not check the canceled
context. Their page readers' buffers leak: 32 KiB for a 3-child struct
of `int32`s in the new test.
- `newLeafReader` returns a live `*ColumnReader` together with its
`nextRowGroup` error. Callers check the error and drop the reader, so
nobody releases it.
`GetFieldReaders` already releases its partial readers on error; this
applies the same cleanup inside `getReader`.
### What changes are included in this PR?
- `getReader` releases the struct's already-built child readers before
returning a child's error.
- `newLeafReader` releases its reader and returns `nil` when
`nextRowGroup` fails.
- Tests:
- `TestGetFieldReaderReleasesStructChildrenOnError` fails a struct's
second child and checks that no allocations remain, with `Parallel`
false and true. On `main` it reports two 16 KiB leaks from
`serializedPageReader.init`.
- `TestGetFieldReaderReturnsNilReaderOnError` checks that a failing leaf
returns a nil reader. On `main` it gets a `*leafReader`.
Both reuse `failingReaderAt` from
`TestGetFieldReadersReleasesPartialReadersOnError`.
This overlaps #1434, which adds a serial loop for struct children in the
same function. Whichever lands second should release the built children
in that loop too.
### Are these changes tested?
Yes:
- `PARQUET_TEST_DATA=<parquet-testing-data> go test ./parquet/...`
- `go test -race ./parquet/pqarrow -run
'^(TestGetFieldReader|TestGetFieldReaders|TestRecordReader|TestPartialStructColumnRead|TestStructReaderSerial)'
-count=1`
- `go vet ./parquet/pqarrow`, `git diff --check`
### Are there any user-facing changes?
`GetFieldReader` now returns a nil reader with its error when a leaf
column fails to open, where it used to return an unreleased reader.
Struct reads that fail no longer leak memory from a non-Go allocator.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
parquet/pqarrow/column_readers.go | 7 ++-
parquet/pqarrow/file_reader.go | 5 ++
parquet/pqarrow/file_reader_test.go | 96 +++++++++++++++++++++++++++++++++++++
3 files changed, 106 insertions(+), 2 deletions(-)
diff --git a/parquet/pqarrow/column_readers.go
b/parquet/pqarrow/column_readers.go
index b90072ea..237df7d8 100644
--- a/parquet/pqarrow/column_readers.go
+++ b/parquet/pqarrow/column_readers.go
@@ -67,8 +67,11 @@ func newLeafReader(rctx *readerCtx, field *arrow.Field,
input *columnIterator, l
}
ret.refCount.Add(1)
- err := ret.nextRowGroup(0)
- return &ColumnReader{ret}, err
+ if err := ret.nextRowGroup(0); err != nil {
+ ret.Release()
+ return nil, err
+ }
+ return &ColumnReader{ret}, nil
}
func (lr *leafReader) Retain() {
diff --git a/parquet/pqarrow/file_reader.go b/parquet/pqarrow/file_reader.go
index 36382a6a..d0b5f9b3 100644
--- a/parquet/pqarrow/file_reader.go
+++ b/parquet/pqarrow/file_reader.go
@@ -678,6 +678,11 @@ func (fr *FileReader) getReader(ctx context.Context, field
*SchemaField, arrowFi
})
}
if err = g.Wait(); err != nil {
+ for _, rdr := range childReaders {
+ if rdr != nil {
+ rdr.Release()
+ }
+ }
return nil, err
}
diff --git a/parquet/pqarrow/file_reader_test.go
b/parquet/pqarrow/file_reader_test.go
index 737dff2a..e9f1ecbb 100644
--- a/parquet/pqarrow/file_reader_test.go
+++ b/parquet/pqarrow/file_reader_test.go
@@ -251,6 +251,102 @@ func TestGetFieldReadersReleasesPartialReadersOnError(t
*testing.T) {
require.Zero(t, mem.CurrentAlloc())
}
+func TestGetFieldReaderReturnsNilReaderOnError(t *testing.T) {
+ schema := arrow.NewSchema([]arrow.Field{
+ {Name: "first", Type: arrow.PrimitiveTypes.Int32},
+ {Name: "second", Type: arrow.PrimitiveTypes.Int32},
+ }, nil)
+ record, _, err := array.RecordFromJSON(memory.DefaultAllocator, schema,
+ strings.NewReader(`[{"first": 1, "second": 2}]`))
+ require.NoError(t, err)
+ defer record.Release()
+
+ var buf bytes.Buffer
+ writer, err := pqarrow.NewFileWriter(schema, &buf, nil,
pqarrow.DefaultWriterProps())
+ require.NoError(t, err)
+ require.NoError(t, writer.Write(record))
+ require.NoError(t, writer.Close())
+
+ readErr := errors.New("read second column")
+ source := &failingReaderAt{
+ Reader: bytes.NewReader(buf.Bytes()),
+ failOffset: -1,
+ err: readErr,
+ }
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ parquetReader, err := file.NewParquetReader(source,
+ file.WithReadProps(parquet.NewReaderProperties(mem)))
+ require.NoError(t, err)
+ defer parquetReader.Close()
+
+ column, err := parquetReader.MetaData().RowGroup(0).ColumnChunk(1)
+ require.NoError(t, err)
+ source.failOffset = column.DataPageOffset()
+ if column.HasDictionaryPage() && column.DictionaryPageOffset() > 0 {
+ source.failOffset = column.DictionaryPageOffset()
+ }
+
+ reader, err := pqarrow.NewFileReader(parquetReader,
pqarrow.ArrowReadProperties{}, mem)
+ require.NoError(t, err)
+ fieldReader, err := reader.GetFieldReader(context.Background(), 1,
map[int]bool{1: true}, []int{0})
+ require.ErrorIs(t, err, readErr)
+ require.Nil(t, fieldReader, "a reader returned with an error is never
released by callers")
+}
+
+func TestGetFieldReaderReleasesStructChildrenOnError(t *testing.T) {
+ structType := arrow.StructOf(
+ arrow.Field{Name: "first", Type: arrow.PrimitiveTypes.Int32},
+ arrow.Field{Name: "second", Type: arrow.PrimitiveTypes.Int32},
+ arrow.Field{Name: "third", Type: arrow.PrimitiveTypes.Int32},
+ )
+ schema := arrow.NewSchema([]arrow.Field{{Name: "nested", Type:
structType}}, nil)
+ record, _, err := array.RecordFromJSON(memory.DefaultAllocator, schema,
+ strings.NewReader(`[{"nested": {"first": 1, "second": 2,
"third": 3}}]`))
+ require.NoError(t, err)
+ defer record.Release()
+
+ var buf bytes.Buffer
+ writer, err := pqarrow.NewFileWriter(schema, &buf, nil,
pqarrow.DefaultWriterProps())
+ require.NoError(t, err)
+ require.NoError(t, writer.Write(record))
+ require.NoError(t, writer.Close())
+
+ for _, parallel := range []bool{false, true} {
+ t.Run(fmt.Sprintf("parallel=%t", parallel), func(t *testing.T) {
+ readErr := errors.New("read second child")
+ source := &failingReaderAt{
+ Reader: bytes.NewReader(buf.Bytes()),
+ failOffset: -1,
+ err: readErr,
+ }
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ parquetReader, err := file.NewParquetReader(source,
+
file.WithReadProps(parquet.NewReaderProperties(mem)))
+ require.NoError(t, err)
+ defer parquetReader.Close()
+
+ column, err :=
parquetReader.MetaData().RowGroup(0).ColumnChunk(1)
+ require.NoError(t, err)
+ source.failOffset = column.DataPageOffset()
+ if column.HasDictionaryPage() &&
column.DictionaryPageOffset() > 0 {
+ source.failOffset =
column.DictionaryPageOffset()
+ }
+
+ reader, err := pqarrow.NewFileReader(parquetReader,
pqarrow.ArrowReadProperties{Parallel: parallel}, mem)
+ require.NoError(t, err)
+ fieldReader, err :=
reader.GetFieldReader(context.Background(), 0,
+ map[int]bool{0: true, 1: true, 2: true},
[]int{0})
+ require.ErrorIs(t, err, readErr)
+ require.Nil(t, fieldReader)
+ require.Zero(t, mem.CurrentAlloc())
+ })
+ }
+}
+
func TestRecordReaderParallel(t *testing.T) {
mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
defer mem.AssertSize(t, 0)
