-
Notifications
You must be signed in to change notification settings - Fork 544
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Mimir query engine: binary operations between two scalars (#9277)
* Add support for binary operations between two scalars * Add changelog entry * Address PR feedback
- Loading branch information
1 parent
98f0544
commit 4627dd9
Showing
7 changed files
with
197 additions
and
61 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
102 changes: 102 additions & 0 deletions
102
pkg/streamingpromql/operators/scalar_scalar_binary_operation.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| // SPDX-License-Identifier: AGPL-3.0-only | ||
|
|
||
| package operators | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/prometheus/prometheus/promql/parser" | ||
| "github.com/prometheus/prometheus/promql/parser/posrange" | ||
|
|
||
| "github.com/grafana/mimir/pkg/streamingpromql/compat" | ||
| "github.com/grafana/mimir/pkg/streamingpromql/limiting" | ||
| "github.com/grafana/mimir/pkg/streamingpromql/types" | ||
| ) | ||
|
|
||
| type ScalarScalarBinaryOperation struct { | ||
| Left types.ScalarOperator | ||
| Right types.ScalarOperator | ||
| Op parser.ItemType | ||
| MemoryConsumptionTracker *limiting.MemoryConsumptionTracker | ||
|
|
||
| opFunc binaryOperationFunc | ||
| expressionPosition posrange.PositionRange | ||
| } | ||
|
|
||
| var _ types.ScalarOperator = &ScalarScalarBinaryOperation{} | ||
|
|
||
| func NewScalarScalarBinaryOperation( | ||
| left, right types.ScalarOperator, | ||
| op parser.ItemType, | ||
| memoryConsumptionTracker *limiting.MemoryConsumptionTracker, | ||
| expressionPosition posrange.PositionRange, | ||
| ) (*ScalarScalarBinaryOperation, error) { | ||
| f := arithmeticOperationFuncs[op] | ||
| if f == nil { | ||
| return nil, compat.NewNotSupportedError(fmt.Sprintf("binary expression with '%s'", op)) | ||
| } | ||
|
|
||
| return &ScalarScalarBinaryOperation{ | ||
| Left: left, | ||
| Right: right, | ||
| Op: op, | ||
| MemoryConsumptionTracker: memoryConsumptionTracker, | ||
|
|
||
| opFunc: f, | ||
| expressionPosition: expressionPosition, | ||
| }, nil | ||
| } | ||
|
|
||
| func (s *ScalarScalarBinaryOperation) GetValues(ctx context.Context) (types.ScalarData, error) { | ||
| leftValues, err := s.Left.GetValues(ctx) | ||
| if err != nil { | ||
| return types.ScalarData{}, err | ||
| } | ||
|
|
||
| rightValues, err := s.Right.GetValues(ctx) | ||
| if err != nil { | ||
| return types.ScalarData{}, err | ||
| } | ||
|
|
||
| // Binary operations between two scalars always produce a float value, as only arithmetic operators or comparison | ||
| // operators (with the bool keyword) are supported between two scalars. | ||
| // | ||
| // Furthermore, scalar values always have a value at each step. | ||
| // | ||
| // So we can just compute the result of each pairwise operation without examining the timestamps of each sample. | ||
| // | ||
| // We store the result in the slice from the left operator, and return the right operator's slice once we're done. | ||
| for i, left := range leftValues.Samples { | ||
| right := rightValues.Samples[i] | ||
|
|
||
| f, h, ok, err := s.opFunc(left.F, right.F, nil, nil) | ||
|
|
||
| if err != nil { | ||
| return types.ScalarData{}, err | ||
| } | ||
|
|
||
| if !ok { | ||
| panic(fmt.Sprintf("%v binary operation between two scalars (%v and %v) did not produce a result, this should never happen", s.Op.String(), left.F, right.F)) | ||
| } | ||
|
|
||
| if h != nil { | ||
| panic(fmt.Sprintf("%v binary operation between two scalars (%v and %v) produced a histogram result, this should never happen", s.Op.String(), left.F, right.F)) | ||
| } | ||
|
|
||
| leftValues.Samples[i].F = f | ||
| } | ||
|
|
||
| types.FPointSlicePool.Put(rightValues.Samples, s.MemoryConsumptionTracker) | ||
|
|
||
| return leftValues, nil | ||
| } | ||
|
|
||
| func (s *ScalarScalarBinaryOperation) ExpressionPosition() posrange.PositionRange { | ||
| return s.expressionPosition | ||
| } | ||
|
|
||
| func (s *ScalarScalarBinaryOperation) Close() { | ||
| s.Left.Close() | ||
| s.Right.Close() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters