diff --git a/go.mod b/go.mod index 1b0a7fa..e53ee2c 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/charmbracelet/bubbletea v1.3.10 github.com/charmbracelet/lipgloss v1.1.0 github.com/spf13/cobra v1.10.1 + golang.org/x/sync v0.18.0 ) require ( @@ -48,7 +49,6 @@ require ( github.com/zeebo/xxh3 v1.0.2 // indirect golang.org/x/exp v0.0.0-20251113190631-e25ba8c21ef6 // indirect golang.org/x/mod v0.30.0 // indirect - golang.org/x/sync v0.18.0 // indirect golang.org/x/sys v0.38.0 // indirect golang.org/x/telemetry v0.0.0-20251112162317-03ef243c208a // indirect golang.org/x/text v0.31.0 // indirect diff --git a/internal/parse/rowgroup_values.go b/internal/parse/rowgroup_values.go index cb4c3de..afda5b3 100644 --- a/internal/parse/rowgroup_values.go +++ b/internal/parse/rowgroup_values.go @@ -3,10 +3,10 @@ package parse import ( "fmt" "strconv" - "sync" "github.com/apache/arrow-go/v18/parquet" "github.com/apache/arrow-go/v18/parquet/file" + "golang.org/x/sync/errgroup" ) func ReadRows(filename string, amount int) ([]string, [][]string, error) { @@ -60,37 +60,35 @@ func ReadRows(filename string, amount int) ([]string, [][]string, error) { return headers, rows, nil } -type columnValueResult struct { - index int - values []string - err error -} - func readRowGroupValues(rowGroup *file.RowGroupReader, amount int64) ([][]string, error) { - resultChan := make(chan columnValueResult, rowGroup.NumColumns()) - var wg sync.WaitGroup - for i := 0; i < rowGroup.NumColumns(); i++ { - wg.Add(1) - go func(idx int) { - defer wg.Done() - column, _ := rowGroup.Column(idx) // TODO: implement error group + numColumns := rowGroup.NumColumns() + results := make([][]string, numColumns) + + var g errgroup.Group + for i := 0; i < numColumns; i++ { + idx := i + g.Go(func() error { + column, err := rowGroup.Column(idx) + if err != nil { + return fmt.Errorf("reading column %d: %w", idx, err) + } columnValues, err := readColumnValues(column, amount) - resultChan <- columnValueResult{idx, columnValues, err} - }(i) + if err != nil { + return err + } + results[idx] = columnValues + return nil + }) + } + if err := g.Wait(); err != nil { + return nil, err } - wg.Wait() - close(resultChan) valuesPerRow := make([][]string, amount) for i := int64(0); i < amount; i++ { - valuesPerRow[i] = make([]string, rowGroup.NumColumns()) - } - for res := range resultChan { - if res.err != nil { - return nil, res.err - } - for rowIdx, val := range res.values { - valuesPerRow[rowIdx][res.index] = val + valuesPerRow[i] = make([]string, numColumns) + for col := 0; col < numColumns; col++ { + valuesPerRow[i][col] = results[col][i] } }