elsteveogrande commented on code in PR #1434:
URL: https://github.com/apache/arrow-go/pull/1434#discussion_r4428277648
##########
parquet/pqarrow/struct_reader_bench_test.go:
##########
@@ -95,6 +95,60 @@ func BenchmarkReadNestedStructSerial(b *testing.B) {
}
}
+func BenchmarkGetFieldReaderNestedStructSerial(b *testing.B) {
+ const (
+ nrows = 1024
+ rowGroupSize = 64
+ )
+
+ for _, nchildren := range []int{1, 8, 32, 128} {
+ b.Run(fmt.Sprintf("children=%d", nchildren), func(b *testing.B)
{
+ mem := memory.DefaultAllocator
+ tbl := makeWideNestedInt32Table(mem, nchildren, nrows)
+ defer tbl.Release()
+
+ var buf bytes.Buffer
+ writerProps :=
parquet.NewWriterProperties(parquet.WithCompression(compress.Codecs.Uncompressed))
+ if err := pqarrow.WriteTable(tbl, &buf, rowGroupSize,
writerProps, pqarrow.DefaultWriterProps()); err != nil {
+ b.Fatal(err)
+ }
+
+ pf, err :=
file.NewParquetReader(bytes.NewReader(buf.Bytes()))
+ if err != nil {
+ b.Fatal(err)
+ }
+ defer pf.Close()
+
+ reader, err := pqarrow.NewFileReader(pf,
pqarrow.ArrowReadProperties{
+ BatchSize: nrows,
+ Parallel: false,
+ }, mem)
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ rowGroups := make([]int, nrows/rowGroupSize)
+ for i := range rowGroups {
+ rowGroups[i] = i
+ }
+ includedLeaves := make(map[int]bool, nchildren)
+ for i := 0; i < nchildren; i++ {
+ includedLeaves[i] = true
+ }
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for range b.N {
Review Comment:
Done in 7c114ed: `for b.Loop()`, and the `b.ResetTimer()` is gone.
##########
parquet/pqarrow/file_reader.go:
##########
@@ -652,36 +652,47 @@ func (fr *FileReader) getReader(ctx context.Context,
field *SchemaField, arrowFi
childReaders := make([]*ColumnReader, len(field.Children))
childFields := make([]arrow.Field, len(field.Children))
- // Get child field readers concurrently
- // 'getReader' causes a read operation. Issue the 'reads'
concurrently
- // When reading structs with large numbers of columns, the
serial load is very slow.
- // This is especially true when reading Cloud Storage. Loading
concurrently
- // greatly improves performance.
- g, gctx := errgroup.WithContext(ctx)
if !fr.Props.Parallel {
- g.SetLimit(1)
- }
-
- for n, child := range field.Children {
- n, child := n, child
- g.Go(func() error {
- reader, err := fr.getReader(gctx, &child,
*child.Field)
+ // Serial reads build the child readers in order,
without a goroutine per child.
+ for n, child := range field.Children {
+ reader, err := fr.getReader(ctx, &child,
*child.Field)
if err != nil {
- return err
+ return nil, err
Review Comment:
Added `TestGetFieldReaderSerialReturnsFirstChildError` in 7c114ed. With
`Parallel: false` it fails the second child's read and checks that
`GetFieldReader` returns that error without reading the third child's chunk. It
fails with your `continue` change, and on `main`, which still reads the third
child after the second fails.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
