| 121 | } |
| 122 | |
| 123 | func NewParquetQueryable( |
| 124 | config Config, |
| 125 | storageCfg cortex_tsdb.BlocksStorageConfig, |
| 126 | limits *validation.Overrides, |
| 127 | blockStorageQueryable *BlocksStoreQueryable, |
| 128 | logger log.Logger, |
| 129 | reg prometheus.Registerer, |
| 130 | ) (storage.Queryable, error) { |
| 131 | bucketClient, err := createCachingBucketClient(context.Background(), storageCfg, nil, "parquet-querier", logger, reg) |
| 132 | if err != nil { |
| 133 | return nil, err |
| 134 | } |
| 135 | |
| 136 | manager, err := services.NewManager(blockStorageQueryable) |
| 137 | if err != nil { |
| 138 | return nil, err |
| 139 | } |
| 140 | |
| 141 | cache, err := parquetutil.NewParquetShardCache[parquet_storage.ParquetShard](&config.ParquetShardCache, "parquet-shards", extprom.WrapRegistererWith(prometheus.Labels{"component": "querier"}, reg)) |
| 142 | if err != nil { |
| 143 | return nil, err |
| 144 | } |
| 145 | |
| 146 | cDecoder := schema.NewPrometheusParquetChunksDecoder(chunkenc.NewPool()) |
| 147 | |
| 148 | parquetQueryableOpts := []queryable.QueryableOpts{ |
| 149 | queryable.WithHonorProjectionHints(config.HonorProjectionHints), |
| 150 | queryable.WithRowCountLimitFunc(func(ctx context.Context) int64 { |
| 151 | // Ignore error as this shouldn't happen. |
| 152 | // If failed to resolve tenant we will just use the default limit value. |
| 153 | userID, _ := users.TenantID(ctx) |
| 154 | return int64(limits.ParquetMaxFetchedRowCount(userID)) |
| 155 | }), |
| 156 | queryable.WithChunkBytesLimitFunc(func(ctx context.Context) int64 { |
| 157 | // Ignore error as this shouldn't happen. |
| 158 | // If failed to resolve tenant we will just use the default limit value. |
| 159 | userID, _ := users.TenantID(ctx) |
| 160 | return int64(limits.ParquetMaxFetchedChunkBytes(userID)) |
| 161 | }), |
| 162 | queryable.WithDataBytesLimitFunc(func(ctx context.Context) int64 { |
| 163 | // Ignore error as this shouldn't happen. |
| 164 | // If failed to resolve tenant we will just use the default limit value. |
| 165 | userID, _ := users.TenantID(ctx) |
| 166 | return int64(limits.ParquetMaxFetchedDataBytes(userID)) |
| 167 | }), |
| 168 | queryable.WithMaterializedLabelsFilterCallback(materializedLabelsFilterCallback), |
| 169 | queryable.WithMaterializedSeriesCallback(func(ctx context.Context, series storage.ChunkSeries) error { |
| 170 | queryLimiter := limiter.QueryLimiterFromContextWithFallback(ctx) |
| 171 | chkCount := 0 |
| 172 | chunkSize := 0 |
| 173 | lblSize := 0 |
| 174 | lblAdapter := cortexpb.FromLabelsToLabelAdapters(series.Labels()) |
| 175 | for _, lbl := range lblAdapter { |
| 176 | lblSize += lbl.Size() |
| 177 | } |
| 178 | iter := series.Iterator(nil) |
| 179 | for iter.Next() { |
| 180 | chk := iter.At() |