55 "time"
66
77 "github.com/go-kit/log"
8+ "github.com/go-kit/log/level"
9+ "github.com/parquet-go/parquet-go"
810 "github.com/pkg/errors"
911 "github.com/prometheus-community/parquet-common/schema"
1012 "github.com/prometheus-community/parquet-common/search"
@@ -16,14 +18,17 @@ import (
1618 "github.com/prometheus/prometheus/tsdb/chunkenc"
1719 "github.com/prometheus/prometheus/util/annotations"
1820 "github.com/thanos-io/thanos/pkg/strutil"
21+ "golang.org/x/sync/errgroup"
1922
2023 "github.com/cortexproject/cortex/pkg/storage/bucket"
2124 cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
2225 "github.com/cortexproject/cortex/pkg/storage/tsdb/bucketindex"
2326 "github.com/cortexproject/cortex/pkg/tenant"
2427 "github.com/cortexproject/cortex/pkg/util"
28+ util_log "github.com/cortexproject/cortex/pkg/util/log"
2529 "github.com/cortexproject/cortex/pkg/util/multierror"
2630 "github.com/cortexproject/cortex/pkg/util/services"
31+ "github.com/cortexproject/cortex/pkg/util/validation"
2732)
2833
2934type parquetQueryableFallbackMetrics struct {
@@ -59,12 +64,15 @@ type parquetQueryableWithFallback struct {
5964
6065 // metrics
6166 metrics * parquetQueryableFallbackMetrics
67+
68+ limits * validation.Overrides
69+ logger log.Logger
6270}
6371
6472func NewParquetQueryable (
6573 config Config ,
6674 storageCfg cortex_tsdb.BlocksStorageConfig ,
67- limits BlocksStoreLimits ,
75+ limits * validation. Overrides ,
6876 blockStorageQueryable * BlocksStoreQueryable ,
6977 logger log.Logger ,
7078 reg prometheus.Registerer ,
@@ -93,18 +101,29 @@ func NewParquetQueryable(
93101 }
94102 userBkt := bucket .NewUserBucketClient (userID , bucketClient , limits )
95103
96- shards := make ([]* parquet_storage.ParquetShard , 0 , len (blocks ))
97-
98- for _ , block := range blocks {
99- // we always only have 1 shard - shard 0
100- shard , err := parquet_storage .OpenParquetShard (ctx , userBkt , block .ID .String (), 0 )
101- if err != nil {
102- return nil , err
103- }
104- shards = append (shards , shard )
104+ shards := make ([]* parquet_storage.ParquetShard , len (blocks ))
105+ errGroup := & errgroup.Group {}
106+
107+ for i , block := range blocks {
108+ errGroup .Go (func () error {
109+ // we always only have 1 shard - shard 0
110+ shard , err := parquet_storage .OpenParquetShard (ctx ,
111+ userBkt ,
112+ block .ID .String (),
113+ 0 ,
114+ parquet_storage .WithFileOptions (
115+ parquet .SkipMagicBytes (true ),
116+ parquet .ReadBufferSize (100 * 1024 ),
117+ parquet .SkipBloomFilters (true ),
118+ ),
119+ parquet_storage .WithOptimisticReader (true ),
120+ )
121+ shards [i ] = shard
122+ return err
123+ })
105124 }
106125
107- return shards , nil
126+ return shards , errGroup . Wait ()
108127 })
109128
110129 p := & parquetQueryableWithFallback {
@@ -115,6 +134,8 @@ func NewParquetQueryable(
115134 subservicesWatcher : services .NewFailureWatcher (),
116135 finder : blockStorageQueryable .finder ,
117136 metrics : newParquetQueryableFallbackMetrics (reg ),
137+ limits : limits ,
138+ logger : logger ,
118139 }
119140
120141 p .Service = services .NewBasicService (p .starting , p .running , p .stopping )
@@ -164,6 +185,8 @@ func (p *parquetQueryableWithFallback) Querier(mint, maxt int64) (storage.Querie
164185 blocksStoreQuerier : bsq ,
165186 finder : p .finder ,
166187 metrics : p .metrics ,
188+ limits : p .limits ,
189+ logger : p .logger ,
167190 }, nil
168191}
169192
@@ -181,6 +204,9 @@ type parquetQuerierWithFallback struct {
181204
182205 // metrics
183206 metrics * parquetQueryableFallbackMetrics
207+
208+ limits * validation.Overrides
209+ logger log.Logger
184210}
185211
186212func (q * parquetQuerierWithFallback ) LabelValues (ctx context.Context , name string , hints * storage.LabelHints , matchers ... * labels.Matcher ) ([]string , annotations.Annotations , error ) {
@@ -275,6 +301,18 @@ func (q *parquetQuerierWithFallback) LabelNames(ctx context.Context, hints *stor
275301}
276302
277303func (q * parquetQuerierWithFallback ) Select (ctx context.Context , sortSeries bool , hints * storage.SelectHints , matchers ... * labels.Matcher ) storage.SeriesSet {
304+ userID , err := tenant .TenantID (ctx )
305+ if err != nil {
306+ storage .ErrSeriesSet (err )
307+ }
308+
309+ if q .limits .QueryVerticalShardSize (userID ) > 1 {
310+ uLogger := util_log .WithUserID (userID , q .logger )
311+ level .Warn (uLogger ).Log ("msg" , "parquet queryable enabled but vertical sharding > 1. Falling back to the block storage" )
312+
313+ return q .blocksStoreQuerier .Select (ctx , sortSeries , hints , matchers ... )
314+ }
315+
278316 mint , maxt , limit := q .minT , q .maxT , 0
279317
280318 if hints != nil {
@@ -288,6 +326,11 @@ func (q *parquetQuerierWithFallback) Select(ctx context.Context, sortSeries bool
288326
289327 serieSets := []storage.SeriesSet {}
290328
329+ // Lets sort the series to merge
330+ if len (parquet ) > 0 && len (remaining ) > 0 {
331+ sortSeries = true
332+ }
333+
291334 if len (parquet ) > 0 {
292335 serieSets = append (serieSets , q .parquetQuerier .Select (InjectBlocksIntoContext (ctx , parquet ... ), sortSeries , hints , matchers ... ))
293336 }
0 commit comments