Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

engine: allow to override opts at query time #501

Merged
merged 2 commits into from
Dec 16, 2024
Merged

engine: allow to override opts at query time #501

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fc40cf4
engine: allow to override opts at query time
Dec 15, 2024
50340b9
engine: add more constructors
Dec 16, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 32 additions & 43 deletions engine/engine.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,14 @@ func (o Opts) getLogicalOptimizers() []logicalplan.Optimizer {
return optimizers
}

// QueryOpts implements promql.QueryOpts but allows to override more engine default options.
type QueryOpts struct {
promql.QueryOpts

// DecodingConcurrency can be used to override the DecodingConcurrency engine setting.
DecodingConcurrency int
yeya24 marked this conversation as resolved.
Show resolved Hide resolved
}

// New creates a new query engine with the given options. The query engine will
// use the storage passed in NewInstantQuery and NewRangeQuery for retrieving
// data when executing queries.
Expand Down Expand Up @@ -239,31 +247,12 @@ func (e *Engine) NewInstantQuery(ctx context.Context, q storage.Queryable, opts
if err != nil {
return nil, err
}

if opts == nil {
opts = promql.NewPrometheusQueryOpts(false, e.lookbackDelta)
}
if opts.LookbackDelta() <= 0 {
opts = promql.NewPrometheusQueryOpts(opts.EnablePerStepStats(), e.lookbackDelta)
}

// determine sorting order before optimizers run, we do this by looking for "sort"
// and "sort_desc" and optimize them away afterwards since they are only needed at
// the presentation layer and not when computing the results.
resultSort := newResultSort(expr)

qOpts := &query.Options{
Start: ts,
End: ts,
Step: 0,
StepsBatch: stepsBatch,
LookbackDelta: opts.LookbackDelta(),
EnablePerStepStats: e.enablePerStepStats && opts.EnablePerStepStats(),
ExtLookbackDelta: e.extLookbackDelta,
EnableAnalysis: e.enableAnalysis,
NoStepSubqueryIntervalFn: e.noStepSubqueryIntervalFn,
DecodingConcurrency: e.decodingConcurrency,
}
qOpts := e.makeQueryOpts(ts, ts, 0, opts)
if qOpts.StepsBatch > 64 {
return nil, ErrStepsBatchTooLarge
}
Expand Down Expand Up @@ -308,13 +297,6 @@ func (e *Engine) NewInstantQueryFromPlan(ctx context.Context, q storage.Queryabl
}
defer e.activeQueryTracker.Delete(idx)

if opts == nil {
opts = promql.NewPrometheusQueryOpts(false, e.lookbackDelta)
}
if opts.LookbackDelta() <= 0 {
opts = promql.NewPrometheusQueryOpts(opts.EnablePerStepStats(), e.lookbackDelta)
}

qOpts := e.makeQueryOpts(ts, ts, 0, opts)
if qOpts.StepsBatch > 64 {
return nil, ErrStepsBatchTooLarge
Expand Down Expand Up @@ -371,12 +353,6 @@ func (e *Engine) NewRangeQuery(ctx context.Context, q storage.Queryable, opts pr
if expr.Type() != parser.ValueTypeVector && expr.Type() != parser.ValueTypeScalar {
return nil, errors.Newf("invalid expression type %q for range query, must be Scalar or instant Vector", parser.DocumentedType(expr.Type()))
}
if opts == nil {
opts = promql.NewPrometheusQueryOpts(false, e.lookbackDelta)
}
if opts.LookbackDelta() <= 0 {
opts = promql.NewPrometheusQueryOpts(opts.EnablePerStepStats(), e.lookbackDelta)
}
qOpts := e.makeQueryOpts(start, end, step, opts)
if qOpts.StepsBatch > 64 {
return nil, ErrStepsBatchTooLarge
Expand Down Expand Up @@ -420,12 +396,6 @@ func (e *Engine) NewRangeQueryFromPlan(ctx context.Context, q storage.Queryable,
}
defer e.activeQueryTracker.Delete(idx)

if opts == nil {
opts = promql.NewPrometheusQueryOpts(false, e.lookbackDelta)
}
if opts.LookbackDelta() <= 0 {
opts = promql.NewPrometheusQueryOpts(opts.EnablePerStepStats(), e.lookbackDelta)
}
qOpts := e.makeQueryOpts(start, end, step, opts)
if qOpts.StepsBatch > 64 {
return nil, ErrStepsBatchTooLarge
Expand Down Expand Up @@ -462,19 +432,38 @@ func (e *Engine) NewRangeQueryFromPlan(ctx context.Context, q storage.Queryable,
}

func (e *Engine) makeQueryOpts(start time.Time, end time.Time, step time.Duration, opts promql.QueryOpts) *query.Options {
qOpts := &query.Options{
res := &query.Options{
Start: start,
End: end,
Step: step,
StepsBatch: stepsBatch,
LookbackDelta: opts.LookbackDelta(),
EnablePerStepStats: e.enablePerStepStats && opts.EnablePerStepStats(),
LookbackDelta: e.lookbackDelta,
EnablePerStepStats: e.enablePerStepStats,
ExtLookbackDelta: e.extLookbackDelta,
EnableAnalysis: e.enableAnalysis,
NoStepSubqueryIntervalFn: e.noStepSubqueryIntervalFn,
DecodingConcurrency: e.decodingConcurrency,
}
return qOpts
if opts == nil {
return res
}

if opts.LookbackDelta() > 0 {
res.LookbackDelta = opts.LookbackDelta()
}
if opts.EnablePerStepStats() {
res.EnablePerStepStats = opts.EnablePerStepStats()
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This maybe should be &= but im not sure if original logic was a bug or not!

}

extOpts, ok := opts.(*QueryOpts)
if !ok {
return res
}

if extOpts.DecodingConcurrency != 0 {
res.DecodingConcurrency = extOpts.DecodingConcurrency
}
return res
}

func (e *Engine) storageScanners(queryable storage.Queryable, qOpts *query.Options, lplan logicalplan.Plan) (engstorage.Scanners, error) {
Expand Down
Loading