raceordie690 commented on code in PR #13120:
URL: https://github.com/apache/arrow/pull/13120#discussion_r873056518
##########
go/parquet/pqarrow/column_readers.go:
##########
@@ -218,32 +222,38 @@ func (sr *structReader) GetRepLevels() ([]int16, error) {
func (sr *structReader) LoadBatch(nrecords int64) error {
var (
- // REP -- Load batches in parallel
+ // Load batches in parallel
// 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.
- wg sync.WaitGroup
- errchan chan error = make(chan error)
- err error
+ wg sync.WaitGroup
+ np int = 1 // default to serial
+ sem chan interface{}
+ errs ErrBuffer
+ err error
)
- //* Read First error from errchan and break only capturing first error
- go func() {
- for err = range errchan {
- break
- }
- }()
+ if sr.props.Parallel {
+ np = len(sr.children)
+ }
+ sem = make(chan interface{}, np)
wg.Add(len(sr.children))
for _, rdr := range sr.children {
+ sem <- nil // Acquire
go func(r *ColumnReader) {
defer wg.Done()
+ defer func() { <-sem }() // release
if err := r.LoadBatch(nrecords); err != nil {
- errchan <- err
+ errs.Append(err)
}
}(rdr)
}
wg.Wait() // wait for reads to complete
- close(errchan)
Review Comment:
That's much better. Thx! I write tons of concurrency code and never came
across that. Though I do use the semaphore code from x/sync often.
--
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]
