chore: adds separate v2 engine config section (#19423)

Author: ashwanthgoli

docs/sources/shared/configuration.md

@@ -4839,39 +4839,40 @@ engine:
   # CLI flag: -querier.engine.max-count-min-sketch-heap-size
   [max_count_min_sketch_heap_size: <int> | default = 10000]
 
+engine_v2:
   # Experimental: Enable next generation query engine for supported queries.
-  # CLI flag: -querier.engine.enable-v2-engine
-  [enable_v2_engine: <boolean> | default = false]
+  # CLI flag: -querier.engine-v2.enable
+  [enable: <boolean> | default = false]
 
   # Experimental: Batch size of the next generation query engine.
-  # CLI flag: -querier.engine.batch-size
+  # CLI flag: -querier.engine-v2.batch-size
   [batch_size: <int> | default = 100]
 
   # Experimental: The number of inputs that are prefetched simultaneously by any
   # Merge node. A value of 0 means that only the currently processed input is
   # prefetched, 1 means that only the next input is prefetched, and so on. A
   # negative value means that all inputs are be prefetched in parallel.
-  # CLI flag: -querier.engine.merge-prefetch-count
+  # CLI flag: -querier.engine-v2.merge-prefetch-count
   [merge_prefetch_count: <int> | default = 0]
 
   # Configures how to read byte ranges from object storage when using the V2
   # engine.
   range_reads:
     # Experimental: maximum number of parallel reads
-    # CLI flag: -querier.engine.range-reads.max-parallelism
+    # CLI flag: -querier.engine-v2.range-reads.max-parallelism
     [max_parallelism: <int> | default = 10]
 
     # Experimental: maximum distance (in bytes) between ranges that causes them
     # to be coalesced into a single range
-    # CLI flag: -querier.engine.range-reads.coalesce-size
+    # CLI flag: -querier.engine-v2.range-reads.coalesce-size
     [coalesce_size: <int> | default = 1048576]
 
     # Experimental: maximum size of a byte range
-    # CLI flag: -querier.engine.range-reads.max-range-size
+    # CLI flag: -querier.engine-v2.range-reads.max-range-size
     [max_range_size: <int> | default = 8388608]
 
     # Experimental: minimum size of a byte range
-    # CLI flag: -querier.engine.range-reads.min-range-size
+    # CLI flag: -querier.engine-v2.range-reads.min-range-size
     [min_range_size: <int> | default = 1048576]
 
 # The maximum number of queries that can be simultaneously processed by the

pkg/engine/engine.go

@@ -2,6 +2,7 @@ package engine
 
 import (
 	"context"
+	"flag"
 	"fmt"
 	"strings"
 	"time"
@@ -35,21 +36,21 @@ var tracer = otel.Tracer("pkg/engine")
 var ErrNotSupported = errors.New("feature not supported in new query engine")
 
 // New creates a new instance of the query engine that implements the [logql.Engine] interface.
-func New(opts logql.EngineOpts, cfg metastore.Config, bucket objstore.Bucket, limits logql.Limits, reg prometheus.Registerer, logger log.Logger) *QueryEngine {
+func New(cfg Config, metastoreCfg metastore.Config, bucket objstore.Bucket, limits logql.Limits, reg prometheus.Registerer, logger log.Logger) *QueryEngine {
 	var ms metastore.Metastore
 	if bucket != nil {
 		indexBucket := bucket
-		if cfg.IndexStoragePrefix != "" {
-			indexBucket = objstore.NewPrefixedBucket(bucket, cfg.IndexStoragePrefix)
+		if metastoreCfg.IndexStoragePrefix != "" {
+			indexBucket = objstore.NewPrefixedBucket(bucket, metastoreCfg.IndexStoragePrefix)
 		}
 		ms = metastore.NewObjectMetastore(indexBucket, logger, reg)
 	}
 
-	if opts.BatchSize <= 0 {
-		panic(fmt.Sprintf("invalid batch size for query engine. must be greater than 0, got %d", opts.BatchSize))
+	if cfg.BatchSize <= 0 {
+		panic(fmt.Sprintf("invalid batch size for query engine. must be greater than 0, got %d", cfg.BatchSize))
 	}
-	if opts.RangeConfig.IsZero() {
-		opts.RangeConfig = rangeio.DefaultConfig
+	if cfg.RangeConfig.IsZero() {
+		cfg.RangeConfig = rangeio.DefaultConfig
 	}
 
 	return &QueryEngine{
@@ -58,18 +59,40 @@ func New(opts logql.EngineOpts, cfg metastore.Config, bucket objstore.Bucket, li
 		limits:    limits,
 		metastore: ms,
 		bucket:    bucket,
-		opts:      opts,
+		cfg:       cfg,
 	}
 }
 
+// Config holds the configuration options to use with the next generation Loki Query Engine.
+type Config struct {
+	// Enable the next generation Loki Query Engine for supported queries.
+	Enable bool `yaml:"enable" category:"experimental"`
+
+	// Batch size of the v2 execution engine.
+	BatchSize int `yaml:"batch_size" category:"experimental"`
+
+	// MergePrefetchCount controls the number of inputs that are prefetched simultaneously by any Merge node.
+	MergePrefetchCount int `yaml:"merge_prefetch_count" category:"experimental"`
+
+	// RangeConfig determines how to optimize range reads in the V2 engine.
+	RangeConfig rangeio.Config `yaml:"range_reads" category:"experimental" doc:"description=Configures how to read byte ranges from object storage when using the V2 engine."`
+}
+
+func (opts *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
+	f.BoolVar(&opts.Enable, prefix+"enable", false, "Experimental: Enable next generation query engine for supported queries.")
+	f.IntVar(&opts.BatchSize, prefix+"batch-size", 100, "Experimental: Batch size of the next generation query engine.")
+	f.IntVar(&opts.MergePrefetchCount, prefix+"merge-prefetch-count", 0, "Experimental: The number of inputs that are prefetched simultaneously by any Merge node. A value of 0 means that only the currently processed input is prefetched, 1 means that only the next input is prefetched, and so on. A negative value means that all inputs are be prefetched in parallel.")
+	opts.RangeConfig.RegisterFlags(prefix+"range-reads.", f)
+}
+
 // QueryEngine combines logical planning, physical planning, and execution to evaluate LogQL queries.
 type QueryEngine struct {
 	logger    log.Logger
 	metrics   *metrics
 	limits    logql.Limits
 	metastore metastore.Metastore
 	bucket    objstore.Bucket
-	opts      logql.EngineOpts
+	cfg       Config
 }
 
 // Query implements [logql.Engine].
@@ -115,7 +138,7 @@ func (e *QueryEngine) Execute(ctx context.Context, params logql.Params) (logqlmo
 
 	// Inject the range config into the context for any calls to
 	// [rangeio.ReadRanges] to make use of.
-	ctx = rangeio.WithConfig(ctx, &e.opts.RangeConfig)
+	ctx = rangeio.WithConfig(ctx, &e.cfg.RangeConfig)
 
 	logicalPlan, err := func() (*logical.Plan, error) {
 		_, span := tracer.Start(ctx, "QueryEngine.Execute.logicalPlan")
@@ -196,8 +219,8 @@ func (e *QueryEngine) Execute(ctx context.Context, params logql.Params) (logqlmo
 		timer := prometheus.NewTimer(e.metrics.execution)
 
 		cfg := executor.Config{
-			BatchSize:          int64(e.opts.BatchSize),
-			MergePrefetchCount: e.opts.MergePrefetchCount,
+			BatchSize:          int64(e.cfg.BatchSize),
+			MergePrefetchCount: e.cfg.MergePrefetchCount,
 			Bucket:             e.bucket,
 		}
 		pipeline := executor.Run(ctx, cfg, physicalPlan, logger)

pkg/engine/internal/semconv/identifier.go

@@ -126,7 +126,7 @@ func MustParseFQN(fqn string) *Identifier {
 type scopeOrigin string
 type scopeType string
 
-// Scope decribes the origin and type of an identifier.
+// Scope describes the origin and type of an identifier.
 type Scope struct {
 	Origin scopeOrigin
 	Type   scopeType

pkg/logql/bench/store_dataobj_v2_engine.go

@@ -28,16 +28,16 @@ type DataObjV2EngineStore struct {
 // NewDataObjV2EngineStore creates a new store that uses the v2 dataobj engine.
 func NewDataObjV2EngineStore(dir string, tenantID string) (*DataObjV2EngineStore, error) {
 	storageDir := filepath.Join(dir, storageDir)
-	return dataobjV2StoreWithOpts(storageDir, tenantID, logql.EngineOpts{
-		EnableV2Engine: true,
-		BatchSize:      512,
-		RangeConfig:    rangeio.DefaultConfig,
+	return dataobjV2StoreWithOpts(storageDir, tenantID, engine.Config{
+		Enable:      true,
+		BatchSize:   512,
+		RangeConfig: rangeio.DefaultConfig,
 	}, metastore.Config{
 		IndexStoragePrefix: "index/v0",
 	})
 }
 
-func dataobjV2StoreWithOpts(dataDir string, tenantID string, engineOpts logql.EngineOpts, cfg metastore.Config) (*DataObjV2EngineStore, error) {
+func dataobjV2StoreWithOpts(dataDir string, tenantID string, cfg engine.Config, metastoreCfg metastore.Config) (*DataObjV2EngineStore, error) {
 	logger := log.NewNopLogger()
 
 	// Setup filesystem client as objstore.Bucket
@@ -57,7 +57,7 @@ func dataobjV2StoreWithOpts(dataDir string, tenantID string, engineOpts logql.En
 	// or derived from the bucket structure if it's multi-tenant aware.
 	// This might require adjustment based on how pkg/engine/engine actually handles multi-tenancy
 	// with a generic objstore.Bucket.
-	queryEngine := engine.New(engineOpts, cfg, bucketClient, logql.NoLimits, nil, logger)
+	queryEngine := engine.New(cfg, metastoreCfg, bucketClient, logql.NoLimits, nil, logger)
 
 	return &DataObjV2EngineStore{
 		engine:   queryEngine,

pkg/logql/engine.go

@@ -37,7 +37,6 @@ import (
 	"github.com/grafana/loki/v3/pkg/util/constants"
 	"github.com/grafana/loki/v3/pkg/util/httpreq"
 	logutil "github.com/grafana/loki/v3/pkg/util/log"
-	"github.com/grafana/loki/v3/pkg/util/rangeio"
 	"github.com/grafana/loki/v3/pkg/util/server"
 	"github.com/grafana/loki/v3/pkg/util/validation"
 )
@@ -161,31 +160,14 @@ type EngineOpts struct {
 	// MaxCountMinSketchHeapSize is the maximum number of labels the heap for a topk query using a count min sketch
 	// can track. This impacts the memory usage and accuracy of a sharded probabilistic topk query.
 	MaxCountMinSketchHeapSize int `yaml:"max_count_min_sketch_heap_size"`
-
-	// Enable the next generation Loki Query Engine for supported queries.
-	EnableV2Engine bool `yaml:"enable_v2_engine" category:"experimental"`
-
-	// Batch size of the v2 execution engine.
-	BatchSize int `yaml:"batch_size" category:"experimental"`
-
-	// MergePrefetchCount controls the number of inputs that are prefetched simultaneously by any Merge node.
-	MergePrefetchCount int `yaml:"merge_prefetch_count" category:"experimental"`
-
-	// RangeConfig determines how to optimize range reads in the V2 engine.
-	RangeConfig rangeio.Config `yaml:"range_reads" category:"experimental" doc:"description=Configures how to read byte ranges from object storage when using the V2 engine."`
 }
 
 func (opts *EngineOpts) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 	f.DurationVar(&opts.MaxLookBackPeriod, prefix+"max-lookback-period", 30*time.Second, "The maximum amount of time to look back for log lines. Used only for instant log queries.")
 	f.IntVar(&opts.MaxCountMinSketchHeapSize, prefix+"max-count-min-sketch-heap-size", 10_000, "The maximum number of labels the heap of a topk query using a count min sketch can track.")
-	f.BoolVar(&opts.EnableV2Engine, prefix+"enable-v2-engine", false, "Experimental: Enable next generation query engine for supported queries.")
-	f.IntVar(&opts.BatchSize, prefix+"batch-size", 100, "Experimental: Batch size of the next generation query engine.")
-	f.IntVar(&opts.MergePrefetchCount, prefix+"merge-prefetch-count", 0, "Experimental: The number of inputs that are prefetched simultaneously by any Merge node. A value of 0 means that only the currently processed input is prefetched, 1 means that only the next input is prefetched, and so on. A negative value means that all inputs are be prefetched in parallel.")
 
 	// Log executing query by default
 	opts.LogExecutingQuery = true
-
-	opts.RangeConfig.RegisterFlags(prefix+"range-reads.", f)
 }
 
 func (opts *EngineOpts) applyDefault() {

pkg/loki/modules.go

@@ -577,7 +577,7 @@ func (t *Loki) initQuerier() (services.Service, error) {
 	}
 
 	var store objstore.Bucket
-	if t.Cfg.Querier.Engine.EnableV2Engine {
+	if t.Cfg.Querier.EngineV2.Enable {
 		store, err = t.createDataObjBucket("dataobj-querier")
 		if err != nil {
 			return nil, err

pkg/querier/http.go

@@ -63,14 +63,19 @@ type QuerierAPI struct {
 
 // NewQuerierAPI returns an instance of the QuerierAPI.
 func NewQuerierAPI(cfg Config, mCfg metastore.Config, querier Querier, limits querier_limits.Limits, store objstore.Bucket, reg prometheus.Registerer, logger log.Logger) *QuerierAPI {
-	return &QuerierAPI{
+	q := &QuerierAPI{
 		cfg:      cfg,
 		limits:   limits,
 		querier:  querier,
 		engineV1: logql.NewEngine(cfg.Engine, querier, limits, logger),
-		engineV2: engine.New(cfg.Engine, mCfg, store, limits, reg, logger),
 		logger:   logger,
 	}
+
+	if cfg.EngineV2.Enable {
+		q.engineV2 = engine.New(cfg.EngineV2, mCfg, store, limits, reg, logger)
+	}
+
+	return q
 }
 
 // RangeQueryHandler is a http.HandlerFunc for range queries and legacy log queries
@@ -87,7 +92,7 @@ func (q *QuerierAPI) RangeQueryHandler(ctx context.Context, req *queryrange.Loki
 		return result, err
 	}
 
-	if q.cfg.Engine.EnableV2Engine && hasDataObjectsAvailable(params.Start(), params.End()) {
+	if q.cfg.EngineV2.Enable && hasDataObjectsAvailable(params.Start(), params.End()) {
 		query := q.engineV2.Query(params)
 		result, err = query.Exec(ctx)
 		if err == nil {
@@ -126,7 +131,7 @@ func (q *QuerierAPI) InstantQueryHandler(ctx context.Context, req *queryrange.Lo
 		return logqlmodel.Result{}, err
 	}
 
-	if q.cfg.Engine.EnableV2Engine && hasDataObjectsAvailable(params.Start(), params.End()) {
+	if q.cfg.EngineV2.Enable && hasDataObjectsAvailable(params.Start(), params.End()) {
 		query := q.engineV2.Query(params)
 		result, err := query.Exec(ctx)
 		if err == nil {

pkg/querier/querier.go

@@ -25,6 +25,7 @@ import (
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc/health/grpc_health_v1"
 
+	"github.com/grafana/loki/v3/pkg/engine"
 	"github.com/grafana/loki/v3/pkg/indexgateway"
 	"github.com/grafana/loki/v3/pkg/iter"
 	"github.com/grafana/loki/v3/pkg/loghttp"
@@ -55,6 +56,7 @@ type Config struct {
 	ExtraQueryDelay           time.Duration    `yaml:"extra_query_delay,omitempty"`
 	QueryIngestersWithin      time.Duration    `yaml:"query_ingesters_within,omitempty"`
 	Engine                    logql.EngineOpts `yaml:"engine,omitempty"`
+	EngineV2                  engine.Config    `yaml:"engine_v2,omitempty" category:"experimental"`
 	MaxConcurrent             int              `yaml:"max_concurrent"`
 	QueryStoreOnly            bool             `yaml:"query_store_only"`
 	QueryIngesterOnly         bool             `yaml:"query_ingester_only"`
@@ -76,6 +78,7 @@ func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 	f.DurationVar(&cfg.ExtraQueryDelay, prefix+"extra-query-delay", 0, "Time to wait before sending more than the minimum successful query requests.")
 	f.DurationVar(&cfg.QueryIngestersWithin, prefix+"query-ingesters-within", 3*time.Hour, "Maximum lookback beyond which queries are not sent to ingester. 0 means all queries are sent to ingester.")
 	cfg.Engine.RegisterFlagsWithPrefix(prefix+"engine.", f)
+	cfg.EngineV2.RegisterFlagsWithPrefix(prefix+"engine-v2.", f)
 	f.IntVar(&cfg.MaxConcurrent, prefix+"max-concurrent", 4, "The maximum number of queries that can be simultaneously processed by the querier.")
 	f.BoolVar(&cfg.QueryStoreOnly, prefix+"query-store-only", false, "Only query the store, and not attempt any ingesters. This is useful for running a standalone querier pool operating only against stored data.")
 	f.BoolVar(&cfg.QueryIngesterOnly, prefix+"query-ingester-only", false, "When true, queriers only query the ingesters, and not stored data. This is useful when the object store is unavailable.")

pkg/querier/querier_test.go

@@ -89,9 +89,6 @@ func TestQuerier_Label_QueryTimeoutConfigFlag(t *testing.T) {
 func mockQuerierConfig() Config {
 	return Config{
 		TailMaxDuration: 1 * time.Minute,
-		Engine: logql.EngineOpts{
-			BatchSize: 1,
-		},
 	}
 }