perf: tune async batch iterator

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

@@ -87,7 +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 = make([]parquet.Value, 0, l)
 	}
 	p = p[:len(values)]
 	for i, v := range values {