parquet sg: artificially cut down query and materialization concurrency

Author: francoposa

go.mod

@@ -29,7 +29,7 @@ require (
 	github.com/opentracing-contrib/go-stdlib v1.1.0 // indirect
 	github.com/opentracing/opentracing-go v1.2.1-0.20220228012449-10b1cf09e00b
 	github.com/pkg/errors v0.9.1
-	github.com/prometheus-community/parquet-common v0.0.0-20251111155302-ad37ac945d97
+	github.com/prometheus-community/parquet-common v0.0.0-20251112212740-499d487df754
 	github.com/prometheus/alertmanager v0.28.1
 	github.com/prometheus/client_golang v1.23.2
 	github.com/prometheus/client_model v0.6.2

go.sum

@@ -914,8 +914,8 @@ github.com/posener/complete v1.2.3/go.mod h1:WZIdtGGp+qx0sLrYKtIRAruyNpv6hFCicSg
 github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt9k/+g42oCprj/FisM4qX9L3sZB3upGN2ZU=
 github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE=
 github.com/pquerna/ffjson v0.0.0-20190930134022-aa0246cd15f7/go.mod h1:YARuvh7BUWHNhzDq2OM5tzR2RiCcN2D7sapiKyCel/M=
-github.com/prometheus-community/parquet-common v0.0.0-20251111155302-ad37ac945d97 h1:h+eNPOYyloibGIotLk+nMeSG51vlL4++w/iuuiO9bMY=
-github.com/prometheus-community/parquet-common v0.0.0-20251111155302-ad37ac945d97/go.mod h1:gewN7ZuOXJh0X2I57iGHyDLbLvL891P2Ynko2QM5axY=
+github.com/prometheus-community/parquet-common v0.0.0-20251112212740-499d487df754 h1:7jfGdBI7IQQbQ1EWr6ar0/aG+WjR8H+kO/1MPwinazw=
+github.com/prometheus-community/parquet-common v0.0.0-20251112212740-499d487df754/go.mod h1:gewN7ZuOXJh0X2I57iGHyDLbLvL891P2Ynko2QM5axY=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
 github.com/prometheus/client_golang v1.4.0/go.mod h1:e9GMxYsXl05ICDXkRhurwBS4Q3OK1iX/F2sw+iXX5zU=

pkg/storage/parquet/chunk_querier.go

@@ -31,7 +31,7 @@ type querierOpts struct {
 }
 
 var DefaultQuerierOpts = querierOpts{
-	concurrency:                      runtime.GOMAXPROCS(0),
+	concurrency:                      (runtime.GOMAXPROCS(0) + 1) / 2,
 	rowCountLimitFunc:                search.NoopQuotaLimitFunc,
 	chunkBytesLimitFunc:              search.NoopQuotaLimitFunc,
 	dataBytesLimitFunc:               search.NoopQuotaLimitFunc,

pkg/storage/parquet/queryable_shard.go

@@ -40,7 +40,7 @@ func newQueryableShard(
 		return nil, err
 	}
 	materializer, err := search.NewMaterializer(
-		shardSchema, chunksDecoder, shard, opts.concurrency, rowCountQuota, chunkBytesQuota, dataBytesQuota, opts.materializedSeriesCallback, opts.materializedLabelsFilterCallback, false)
+		shardSchema, chunksDecoder, shard, (opts.concurrency+1)/2, rowCountQuota, chunkBytesQuota, dataBytesQuota, opts.materializedSeriesCallback, opts.materializedLabelsFilterCallback, false)
 	if err != nil {
 		return nil, err
 	}
@@ -55,7 +55,7 @@ func newQueryableShard(
 
 func (b queryableShard) Query(ctx context.Context, sorted bool, sp *prom_storage.SelectHints, mint, maxt int64, skipChunks bool, matchers []*labels.Matcher) (prom_storage.ChunkSeriesSet, error) {
 	errGroup, ctx := errgroup.WithContext(ctx)
-	errGroup.SetLimit(b.concurrency)
+	errGroup.SetLimit((b.concurrency + 1) / 2)
 
 	rowGroupCount := len(b.shard.LabelsFile().RowGroups())
 	results := make([][]prom_storage.ChunkSeries, rowGroupCount)
@@ -128,7 +128,7 @@ func (b queryableShard) QueryIter(ctx context.Context, sorted bool, sp *prom_sto
 	err error,
 ) {
 	errGroup, groupCtx := errgroup.WithContext(ctx)
-	errGroup.SetLimit(b.concurrency)
+	errGroup.SetLimit((b.concurrency + 1) / 2)
 
 	rowGroupCount := len(b.shard.LabelsFile().RowGroups())
 	labelsSets := make([]prom_storage.ChunkSeriesSet, rowGroupCount)
@@ -222,7 +222,7 @@ func (b queryableShard) LabelNames(ctx context.Context, limit int64, matchers []
 	}
 
 	errGroup, ctx := errgroup.WithContext(ctx)
-	errGroup.SetLimit(b.concurrency)
+	errGroup.SetLimit((b.concurrency + 1) / 2)
 
 	results := make([][]string, len(b.shard.LabelsFile().RowGroups()))
 
@@ -262,7 +262,7 @@ func (b queryableShard) LabelValues(ctx context.Context, name string, limit int6
 	}
 
 	errGroup, ctx := errgroup.WithContext(ctx)
-	errGroup.SetLimit(b.concurrency)
+	errGroup.SetLimit((b.concurrency + 1) / 2)
 
 	results := make([][]string, len(b.shard.LabelsFile().RowGroups()))
 
@@ -298,7 +298,7 @@ func (b queryableShard) LabelValues(ctx context.Context, name string, limit int6
 
 func (b queryableShard) allLabelValues(ctx context.Context, name string, limit int64) ([]string, error) {
 	errGroup, ctx := errgroup.WithContext(ctx)
-	errGroup.SetLimit(b.concurrency)
+	errGroup.SetLimit((b.concurrency + 1) / 2)
 
 	results := make([][]string, len(b.shard.LabelsFile().RowGroups()))