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 cbbc516b fix(parquet_reader): propagate batch read errors (#1081)
cbbc516b is described below
commit cbbc516bda20cc129d9259aa3da3e885d79a2ec7
Author: Minh Vu <[email protected]>
AuthorDate: Wed Aug 5 20:20:55 2026 +0200
fix(parquet_reader): propagate batch read errors (#1081)
## What changed
Return typed column batch read errors from the parquet reader dumper and
handle them in JSON, CSV, and table output modes. Errors include the
affected column path.
## Why
All `ReadBatch` errors were discarded. Corrupt page data could stop
iteration and leave partial output without reporting why decoding
failed.
## Testing
- `go test ./parquet/cmd/parquet_reader`
---
parquet/cmd/parquet_reader/dumper.go | 49 +++++++++++++++++++------------
parquet/cmd/parquet_reader/dumper_test.go | 43 +++++++++++++++++++++++++++
parquet/cmd/parquet_reader/main.go | 18 ++++++++++--
3 files changed, 89 insertions(+), 21 deletions(-)
diff --git a/parquet/cmd/parquet_reader/dumper.go
b/parquet/cmd/parquet_reader/dumper.go
index 15d3bd79..32d377c6 100644
--- a/parquet/cmd/parquet_reader/dumper.go
+++ b/parquet/cmd/parquet_reader/dumper.go
@@ -77,40 +77,47 @@ func createDumper(reader file.ColumnChunkReader) *Dumper {
}
}
-func (dump *Dumper) readNextBatch() {
+func (dump *Dumper) readNextBatch() (err error) {
switch reader := dump.reader.(type) {
case *file.BooleanColumnChunkReader:
values := dump.valueBuffer.([]bool)
- dump.levelsBuffered, dump.valuesBuffered, _ =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
+ dump.levelsBuffered, dump.valuesBuffered, err =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
case *file.Int32ColumnChunkReader:
values := dump.valueBuffer.([]int32)
- dump.levelsBuffered, dump.valuesBuffered, _ =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
+ dump.levelsBuffered, dump.valuesBuffered, err =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
case *file.Int64ColumnChunkReader:
values := dump.valueBuffer.([]int64)
- dump.levelsBuffered, dump.valuesBuffered, _ =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
+ dump.levelsBuffered, dump.valuesBuffered, err =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
case *file.Float32ColumnChunkReader:
values := dump.valueBuffer.([]float32)
- dump.levelsBuffered, dump.valuesBuffered, _ =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
+ dump.levelsBuffered, dump.valuesBuffered, err =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
case *file.Float64ColumnChunkReader:
values := dump.valueBuffer.([]float64)
- dump.levelsBuffered, dump.valuesBuffered, _ =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
+ dump.levelsBuffered, dump.valuesBuffered, err =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
case *file.Int96ColumnChunkReader:
values := dump.valueBuffer.([]parquet.Int96)
- dump.levelsBuffered, dump.valuesBuffered, _ =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
+ dump.levelsBuffered, dump.valuesBuffered, err =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
case *file.ByteArrayColumnChunkReader:
values := dump.valueBuffer.([]parquet.ByteArray)
- dump.levelsBuffered, dump.valuesBuffered, _ =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
+ dump.levelsBuffered, dump.valuesBuffered, err =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
case *file.FixedLenByteArrayColumnChunkReader:
values := dump.valueBuffer.([]parquet.FixedLenByteArray)
- dump.levelsBuffered, dump.valuesBuffered, _ =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
+ dump.levelsBuffered, dump.valuesBuffered, err =
reader.ReadBatch(dump.batchSize, values, dump.defLevels, dump.repLevels)
}
dump.valueOffset = 0
dump.levelOffset = 0
+ return err
}
-func (dump *Dumper) hasNext() bool {
- return dump.levelOffset < dump.levelsBuffered || dump.reader.HasNext()
+func (dump *Dumper) hasNext() (bool, error) {
+ if dump.levelOffset < dump.levelsBuffered {
+ return true, nil
+ }
+ if dump.reader.HasNext() {
+ return true, nil
+ }
+ return false, dump.reader.Err()
}
const microSecondsPerDay = 24 * 3600e6
@@ -157,14 +164,20 @@ func (dump *Dumper) FormatValue(val interface{}, width
int) string {
}
}
-func (dump *Dumper) Next() (interface{}, bool) {
+func (dump *Dumper) Next() (interface{}, bool, error) {
if dump.levelOffset == dump.levelsBuffered {
- if !dump.hasNext() {
- return nil, false
+ hasNext, err := dump.hasNext()
+ if err != nil {
+ return nil, false, fmt.Errorf("reading column %s: %w",
dump.reader.Descriptor().Path(), err)
+ }
+ if !hasNext {
+ return nil, false, nil
+ }
+ if err := dump.readNextBatch(); err != nil {
+ return nil, false, fmt.Errorf("reading column %s: %w",
dump.reader.Descriptor().Path(), err)
}
- dump.readNextBatch()
if dump.levelsBuffered == 0 {
- return nil, false
+ return nil, false, nil
}
}
@@ -173,14 +186,14 @@ func (dump *Dumper) Next() (interface{}, bool) {
dump.levelOffset++
if defLevel < dump.reader.Descriptor().MaxDefinitionLevel() {
- return nil, true
+ return nil, true, nil
}
vb := reflect.ValueOf(dump.valueBuffer)
v := vb.Index(dump.valueOffset).Interface()
dump.valueOffset++
- return v, true
+ return v, true, nil
}
func dumpColIdxImpl[T parquet.ColumnTypes](cidx *metadata.TypedColumnIndex[T])
{
diff --git a/parquet/cmd/parquet_reader/dumper_test.go
b/parquet/cmd/parquet_reader/dumper_test.go
new file mode 100644
index 00000000..75c034c3
--- /dev/null
+++ b/parquet/cmd/parquet_reader/dumper_test.go
@@ -0,0 +1,43 @@
+// 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 main
+
+import (
+ "errors"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/parquet/file"
+ "github.com/stretchr/testify/require"
+)
+
+type errorColumnReader struct {
+ file.ColumnChunkReader
+ err error
+}
+
+func (r *errorColumnReader) HasNext() bool { return false }
+func (r *errorColumnReader) Err() error { return r.err }
+
+func TestDumperHasNextReportsReaderErrors(t *testing.T) {
+ want := errors.New("page read failed")
+ dump := &Dumper{reader: &errorColumnReader{err: want}}
+
+ hasNext, err := dump.hasNext()
+ require.False(t, hasNext)
+ require.ErrorIs(t, err, want)
+}
diff --git a/parquet/cmd/parquet_reader/main.go
b/parquet/cmd/parquet_reader/main.go
index 522781c9..11cb2fbf 100644
--- a/parquet/cmd/parquet_reader/main.go
+++ b/parquet/cmd/parquet_reader/main.go
@@ -329,7 +329,11 @@ func main() {
data := false
first := true
for idx, s := range scanners {
- if val, ok := s.Next(); ok {
+ val, ok, err := s.Next()
+ if err != nil {
+ log.Fatal(err)
+ }
+ if ok {
if !data {
fmt.Fprint(dataOut,
line)
}
@@ -387,7 +391,11 @@ func main() {
line += ","
}
}
- if val, ok := s.Next(); ok {
+ val, ok, err := s.Next()
+ if err != nil {
+ log.Fatal(err)
+ }
+ if ok {
if !data {
fmt.Fprint(dataOut,
line)
}
@@ -435,7 +443,11 @@ func main() {
for {
data := false
for _, s := range scanners {
- if val, ok := s.Next(); ok {
+ val, ok, err := s.Next()
+ if err != nil {
+ log.Fatal(err)
+ }
+ if ok {
if !data {
fmt.Fprint(dataOut,
line)
}
