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

Fix certain binary expression queries by optimizing through push down #6132

Merged
merged 5 commits into from
May 12, 2022
Merged

Fix certain binary expression queries by optimizing through push down #6132

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
1d3e906
Add test cases that fail
chaudum May 9, 2022
16920d9
Fix certain binary expressions for instant queries
chaudum May 10, 2022
fe40f44
Optimize instant vector aggregations with log extraction stage
chaudum May 10, 2022
589f92f
fixup! Add test cases that fail
chaudum May 11, 2022
f225d3e
fixup! fixup! Add test cases that fail
chaudum May 11, 2022
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
8 changes: 8 additions & 0 deletions pkg/logql/downstream_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,14 +132,19 @@ func TestRangeMappingEquivalence(t *testing.T) {

// sum
{`sum(bytes_over_time({a=~".+"}[2s]))`, time.Second},
{`sum(bytes_over_time({a=~".+"} | logfmt | line > 5 [2s]))`, time.Second},
{`sum(count_over_time({a=~".+"}[2s]))`, time.Second},
{`sum(count_over_time({a=~".+"} | logfmt | line > 5 [2s]))`, time.Second},
{`sum(sum_over_time({a=~".+"} | unwrap b [2s]))`, time.Second},
{`sum(sum_over_time({a=~".+"} | logfmt | unwrap line [2s]))`, time.Second},
{`sum(max_over_time({a=~".+"} | unwrap b [2s]))`, time.Second},
{`sum(max_over_time({a=~".+"} | unwrap b [2s]) by (a))`, time.Second},
{`sum(max_over_time({a=~".+"} | logfmt | unwrap line [2s]) by (a))`, time.Second},
{`sum(max_over_time({a=~".+"} | logfmt | unwrap line [2s]))`, time.Second},
{`sum(min_over_time({a=~".+"} | unwrap b [2s]))`, time.Second},
{`sum(min_over_time({a=~".+"} | unwrap b [2s]) by (a))`, time.Second},
{`sum(min_over_time({a=~".+"} | logfmt | unwrap line [2s]) by (a))`, time.Second},
{`sum(min_over_time({a=~".+"} | logfmt | unwrap line [2s]))`, time.Second},
{`sum(rate({a=~".+"}[2s]))`, time.Second},
{`sum(bytes_rate({a=~".+"}[2s]))`, time.Second},

Expand Down Expand Up @@ -215,9 +220,11 @@ func TestRangeMappingEquivalence(t *testing.T) {
{`min(max_over_time({a=~".+"} | unwrap b [2s]))`, time.Second},
{`min(max_over_time({a=~".+"} | unwrap b [2s]) by (a))`, time.Second},
{`min(max_over_time({a=~".+"} | logfmt | unwrap line [2s]) by (a))`, time.Second},
{`min(max_over_time({a=~".+"} | logfmt | unwrap line [2s]))`, time.Second},
{`min(min_over_time({a=~".+"} | unwrap b [2s]))`, time.Second},
{`min(min_over_time({a=~".+"} | unwrap b [2s]) by (a))`, time.Second},
{`min(min_over_time({a=~".+"} | logfmt | unwrap line [2s]) by (a))`, time.Second},
{`min(min_over_time({a=~".+"} | logfmt | unwrap line [2s]))`, time.Second},
{`min(rate({a=~".+"}[2s]))`, time.Second},
{`min(bytes_rate({a=~".+"}[2s]))`, time.Second},

Expand All @@ -237,6 +244,7 @@ func TestRangeMappingEquivalence(t *testing.T) {
// Binary operations
{`bytes_over_time({a=~".+"}[3s]) + count_over_time({a=~".+"}[5s])`, time.Second},
{`sum(count_over_time({a=~".+"}[3s]) * count(sum_over_time({a=~".+"} | unwrap b [5s])))`, time.Second},
{`sum(count_over_time({a=~".+"} | logfmt | b > 2 [3s])) / sum(count_over_time({a=~".+"} [3s]))`, time.Second},

// Multi vector aggregator layer queries
{`sum(max(bytes_over_time({a=~".+"}[3s])))`, time.Second},
Expand Down
14 changes: 13 additions & 1 deletion pkg/logql/rangemapper.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,10 +128,22 @@ func (m RangeMapper) Map(expr syntax.SampleExpr, vectorAggrPushdown *syntax.Vect
if err != nil {
return nil, err
}
// if left hand side is a noop, we need to return the original expression
// so the whole expression is a noop and thus not executed using the
// downstream engine
if e.SampleExpr.String() == lhsMapped.String() {
return e, nil
}
rhsMapped, err := m.Map(e.RHS, vectorAggrPushdown, recorder)
if err != nil {
return nil, err
}
// if right hand side is a noop, we need to return the original expression
// so the whole expression is a noop and thus not executed using the
// downstream engine
if e.RHS.String() == rhsMapped.String() {
return e, nil
}
e.SampleExpr = lhsMapped
e.RHS = rhsMapped
return e, nil
Expand Down Expand Up @@ -331,7 +343,7 @@ func (m RangeMapper) mapRangeAggregationExpr(expr *syntax.RangeAggregationExpr,

// We cannot execute downstream queries that would potentially produce a huge amount of series
// and therefore would very likely fail.
if expr.Grouping == nil && hasLabelExtractionStage(expr) {
if expr.Grouping == nil && vectorAggrPushdown == nil && hasLabelExtractionStage(expr) {
return expr
}
switch expr.Operation {
Expand Down
154 changes: 118 additions & 36 deletions pkg/logql/rangemapper_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -932,6 +932,26 @@ func Test_SplitRangeVectorMapping(t *testing.T) {
)
)`,
},
{
`sum (count_over_time({app="foo"} | logfmt | duration > 10s [3m])) / sum (count_over_time({app="foo"} [3m]))`,
`(
sum (
sum without (
downstream<sum (count_over_time({app="foo"} | logfmt | duration > 10s [1m] offset 2m0s)), shard=<nil>>
++ downstream<sum (count_over_time({app="foo"} | logfmt | duration > 10s [1m] offset 1m0s)), shard=<nil>>
++ downstream<sum (count_over_time({app="foo"} | logfmt | duration > 10s [1m])), shard=<nil>>
)
)
/
sum (
sum without (
downstream<sum (count_over_time({app="foo"} [1m] offset 2m0s)), shard=<nil>>
++ downstream<sum (count_over_time({app="foo"} [1m] offset 1m0s)), shard=<nil>>
++ downstream<sum (count_over_time({app="foo"} [1m])), shard=<nil>>
)
)
)`,
},

// Multi vector aggregator layer queries
{
Expand Down Expand Up @@ -959,6 +979,98 @@ func Test_SplitRangeVectorMapping(t *testing.T) {
)
)`,
},

// outer vector aggregation is pushed down
{
`sum(min_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`sum(
min without(
downstream<sum(min_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 2m0s)), shard=<nil>>
++ downstream<sum(min_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 1m0s)), shard=<nil>>
++ downstream<sum(min_over_time({app="foo"} | logfmt | unwrap bar [1m])), shard=<nil>>
)
)`,
},
{
`sum(max_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`sum(
max without(
downstream<sum(max_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 2m0s)), shard=<nil>>
++ downstream<sum(max_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 1m0s)), shard=<nil>>
++ downstream<sum(max_over_time({app="foo"} | logfmt | unwrap bar [1m])), shard=<nil>>
)
)`,
},
{
`sum(sum_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`sum(
sum without(
downstream<sum(sum_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 2m0s)), shard=<nil>>
++ downstream<sum(sum_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 1m0s)), shard=<nil>>
++ downstream<sum(sum_over_time({app="foo"} | logfmt | unwrap bar [1m])), shard=<nil>>
)
)`,
},
{
`min(min_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`min(
min without(
downstream<min(min_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 2m0s)), shard=<nil>>
++ downstream<min(min_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 1m0s)), shard=<nil>>
++ downstream<min(min_over_time({app="foo"} | logfmt | unwrap bar [1m])), shard=<nil>>
)
)`,
},
{
`min(max_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`min(
max without(
downstream<min(max_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 2m0s)), shard=<nil>>
++ downstream<min(max_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 1m0s)), shard=<nil>>
++ downstream<min(max_over_time({app="foo"} | logfmt | unwrap bar [1m])), shard=<nil>>
)
)`,
},
{
`min(sum_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`min(
sum without(
downstream<min(sum_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 2m0s)), shard=<nil>>
++ downstream<min(sum_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 1m0s)), shard=<nil>>
++ downstream<min(sum_over_time({app="foo"} | logfmt | unwrap bar [1m])), shard=<nil>>
)
)`,
},
{
`max(min_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`max(
min without(
downstream<max(min_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 2m0s)), shard=<nil>>
++ downstream<max(min_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 1m0s)), shard=<nil>>
++ downstream<max(min_over_time({app="foo"} | logfmt | unwrap bar [1m])), shard=<nil>>
)
)`,
},
{
`max(max_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`max(
max without(
downstream<max(max_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 2m0s)), shard=<nil>>
++ downstream<max(max_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 1m0s)), shard=<nil>>
++ downstream<max(max_over_time({app="foo"} | logfmt | unwrap bar [1m])), shard=<nil>>
)
)`,
},
{
`max(sum_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`max(
sum without(
downstream<max(sum_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 2m0s)), shard=<nil>>
++ downstream<max(sum_over_time({app="foo"} | logfmt | unwrap bar [1m] offset 1m0s)), shard=<nil>>
++ downstream<max(sum_over_time({app="foo"} | logfmt | unwrap bar [1m])), shard=<nil>>
)
)`,
},
} {
tc := tc
t.Run(tc.expr, func(t *testing.T) {
Expand Down Expand Up @@ -997,46 +1109,16 @@ func Test_SplitRangeVectorMapping_Noop(t *testing.T) {

// should be noop if inner range aggregation includes a stage for label extraction such as `| json` or `| logfmt`
// because otherwise the downstream queries would result in too many series
{
`sum(min_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`sum(min_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
},
{
`sum(max_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`sum(max_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
},
{
`sum(sum_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`sum(sum_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
},
{
`min(min_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`min(min_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
},
{
`min(max_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`min(max_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
},
{
`min(sum_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`min(sum_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
},
{
`max(min_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`max(min_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
},
{
`max(max_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`max(max_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
},
{
`max(sum_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
`max(sum_over_time({app="foo"} | logfmt | unwrap bar [3m]))`,
},
{
`max_over_time({app="foo"} | json | unwrap bar [3m])`,
`max_over_time({app="foo"} | json | unwrap bar [3m])`,
},

// if one side of a binary expression is a noop, the full query is a noop as well
{
`sum by (foo) (sum_over_time({app="foo"} | json | unwrap bar [3m])) / sum_over_time({app="foo"} | json | unwrap bar [6m])`,
`(sum by (foo) (sum_over_time({app="foo"} | json | unwrap bar [3m])) / sum_over_time({app="foo"} | json | unwrap bar [6m]))`,
},
} {
tc := tc
t.Run(tc.expr, func(t *testing.T) {
Expand Down