Make binary pushdown optional (#358)

Pushing down binary operations with no vector matching makes sense
for Thanos and we should do have it enabled by default. However, we use
the engine to query other metric sources and here we see issues where
matching series between two metrics are not present in the same engine.

To provide an escape hatch while we figure out a more generic solution,
this commit makes the binary pushdown optional and enabled by default,
but can be disabled by setting a flag.

Signed-off-by: Filip Petkovski <[email protected]>

logicalplan/distribute.go

@@ -146,7 +146,8 @@ var distributiveAggregations = map[parser.ItemType]struct{}{
 // DistributedExecutionOptimizer produces a logical plan suitable for
 // distributed Query execution.
 type DistributedExecutionOptimizer struct {
-	Endpoints api.RemoteEndpoints
+	Endpoints          api.RemoteEndpoints
+	SkipBinaryPushdown bool
 }
 
 func (m DistributedExecutionOptimizer) Optimize(plan parser.Expr, opts *query.Options) (parser.Expr, annotations.Annotations) {
@@ -176,7 +177,7 @@ func (m DistributedExecutionOptimizer) Optimize(plan parser.Expr, opts *query.Op
 
 	TraverseBottomUp(nil, &plan, func(parent, current *parser.Expr) (stop bool) {
 		// If the current operation is not distributive, stop the traversal.
-		if !isDistributive(current) {
+		if !isDistributive(current, m.SkipBinaryPushdown) {
 			return true
 		}
 
@@ -202,7 +203,7 @@ func (m DistributedExecutionOptimizer) Optimize(plan parser.Expr, opts *query.Op
 		}
 
 		// If the parent operation is distributive, continue the traversal.
-		if isDistributive(parent) {
+		if isDistributive(parent, m.SkipBinaryPushdown) {
 			return false
 		}
 
@@ -422,14 +423,14 @@ func numSteps(start, end time.Time, step time.Duration) int64 {
 	return (end.UnixMilli()-start.UnixMilli())/step.Milliseconds() + 1
 }
 
-func isDistributive(expr *parser.Expr) bool {
+func isDistributive(expr *parser.Expr, skipBinaryPushdown bool) bool {
 	if expr == nil {
 		return false
 	}
 
 	switch aggr := (*expr).(type) {
 	case *parser.BinaryExpr:
-		return isBinaryExpressionWithOneConstantSide(aggr) || isBinaryExpressionWithDistributableMatching(aggr)
+		return isBinaryExpressionWithOneConstantSide(aggr) || (!skipBinaryPushdown && isBinaryExpressionWithDistributableMatching(aggr))
 	case *parser.AggregateExpr:
 		// Certain aggregations are currently not supported.
 		if _, ok := distributiveAggregations[aggr.Op]; !ok {

logicalplan/distribute_avg.go

@@ -9,11 +9,13 @@ import (
 
 // DistributeAvgOptimizer rewrites an AVG aggregation into a SUM/COUNT aggregation so that
 // it can be executed in a distributed manner.
-type DistributeAvgOptimizer struct{}
+type DistributeAvgOptimizer struct {
+	SkipBinaryPushdown bool
+}
 
 func (r DistributeAvgOptimizer) Optimize(plan parser.Expr, _ *query.Options) (parser.Expr, annotations.Annotations) {
 	TraverseBottomUp(nil, &plan, func(parent, current *parser.Expr) (stop bool) {
-		if !isDistributiveOrAverage(current) {
+		if !isDistributiveOrAverage(current, r.SkipBinaryPushdown) {
 			return true
 		}
 		// If the current node is avg(), distribute the operation and
@@ -39,18 +41,18 @@ func (r DistributeAvgOptimizer) Optimize(plan parser.Expr, _ *query.Options) (pa
 			}
 			return true
 		}
-		return !isDistributiveOrAverage(parent)
+		return !isDistributiveOrAverage(parent, r.SkipBinaryPushdown)
 	})
 	return plan, nil
 }
 
-func isDistributiveOrAverage(expr *parser.Expr) bool {
+func isDistributiveOrAverage(expr *parser.Expr, skipBinaryPushdown bool) bool {
 	if expr == nil {
 		return false
 	}
 	var isAvg bool
 	if aggr, ok := (*expr).(*parser.AggregateExpr); ok {
 		isAvg = aggr.Op == parser.AVG
 	}
-	return isDistributive(expr) || isAvg
+	return isDistributive(expr, skipBinaryPushdown) || isAvg
 }