perf: tune async batch iterator (#3358)

Author: kolesnikovae

pkg/iter/batch_async.go

@@ -19,7 +19,7 @@ type batch[T any] struct {
 	done     chan struct{}
 }
 
-const minBatchSize = 64
+const minBatchSize = 2
 
 func NewAsyncBatchIterator[T, N any](
 	iterator Iterator[T],

pkg/phlaredb/query/repeated.go

@@ -28,12 +28,6 @@ type repeatedRowIterator[T any] struct {
 }
 
 const (
-	// Batch size specifies how many rows to be read
-	// from a column at once. Note that the batched rows
-	// are buffered in-memory, but not reference pages
-	// they were read from.
-	defaultRepeatedRowIteratorBatchSize = 32
-
 	// The value specifies how many individual values to be
 	// read (decoded) from the page.
 	//
@@ -57,8 +51,14 @@ func NewRepeatedRowIterator[T any](
 		rows: rows,
 		columns: NewMultiColumnIterator(ctx,
 			WrapWithRowNumber(rowNumbers),
-			defaultRepeatedRowIteratorBatchSize,
-			rowGroups, columns...),
+			// Batch size specifies how many rows to be read
+			// from a column at once. Note that the batched rows
+			// are buffered in-memory, but not reference pages
+			// they were read from.
+			4,
+			rowGroups,
+			columns...,
+		),
 	}
 }
 

pkg/phlaredb/query/util.go

@@ -1,6 +1,7 @@
 package query
 
 import (
+	"slices"
 	"strings"
 
 	"github.com/colega/zeropool"
@@ -86,9 +87,7 @@ var parquetValuesPool = zeropool.New(func() []parquet.Value { return nil })
 
 func CloneParquetValues(values []parquet.Value) []parquet.Value {
 	p := parquetValuesPool.Get()
-	if l := len(values); cap(p) < l {
-		p = make([]parquet.Value, 0, 2*l)
-	}
+	p = slices.Grow(p, len(values))
 	p = p[:len(values)]
 	for i, v := range values {
 		p[i] = v.Clone()