zeroshade commented on code in PR #1434:
URL: https://github.com/apache/arrow-go/pull/1434#discussion_r4425830406
##########
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:
Non-blocking: no test reaches this. Changing it to `continue` still passes
all of `./parquet/pqarrow`. A test where one child fails would pin the early
return.
##########
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:
Please use `for b.Loop() {` here (repo convention for benchmarks); with it
the `b.ResetTimer()` above can go, since `b.Loop` only times the loop body.
--
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]
