Comparison in measure filters

proto/rill/runtime/v1/queries.proto

@@ -480,19 +480,31 @@
 }
 
 message MeasureFilter {
+  MeasureFilterExpression expression = 1;
+}
+
+message MeasureFilterNode {
   oneof entry {
-    MeasureFilterEntry MeasureFilterEntry = 1;
+    MeasureFilterMeasure MeasureFilterMeasure = 1;
     MeasureFilterExpression MeasureFilterExpression = 2;
+    google.protobuf.Value Value = 3;
+    MeasureFilterNode MeasureFilterNode = 4;
   }
 }
 
-message MeasureFilterEntry {
+message MeasureFilterMeasure {
   enum ColumnType {
     COLUMN_TYPE_UNSPECIFIED = 0;
     COLUMN_TYPE_PREVIOUS = 1;
     COLUMN_TYPE_DELTA_ABSOLUTE = 2;
     COLUMN_TYPE_DELTA_RELATIVE = 3;
   }
+
+  MetricsViewAggregationMeasure measure = 1;
+  ColumnType column_type = 2;
+}
+
+message MeasureFilterExpression {
   enum OperationType {
     OPERATION_TYPE_UNSPECIFIED = 0;
     OPERATION_TYPE_EQUALS = 1;
@@ -501,23 +513,13 @@
     OPERATION_TYPE_LESSER_OR_EQUALS = 4;
     OPERATION_TYPE_GREATER = 5;
     OPERATION_TYPE_GREATER_OR_EQUALS = 6;
+    OPERATION_TYPE_OR = 7;
+    OPERATION_TYPE_AND = 8;
+    OPERATION_TYPE_BETWEEN = 9;
   }
 
-  MetricsViewAggregationMeasure measure = 1;
+  repeated MeasureFilterNode entries = 1;
   OperationType operation_type = 2;
-  google.protobuf.Value value = 3;
-  ColumnType column_type = 4;
-}
-
-message MeasureFilterExpression {
-  enum Joiner {
-    JOINER_UNSPECIFIED = 0;
-    JOINER_OR = 2;
-    JOINER_AND = 3;
-  }
-
-  repeated MeasureFilter entries = 1;
-  Joiner joiner = 2;
 }
 
 // **********

runtime/queries/metricsview.go

@@ -292,69 +292,125 @@
 	return fmt.Sprintf("AND (%s) ", condsClause), args, nil
 }
 
-func buildHavingClause(filter *runtimev1.MeasureFilter, mv *runtimev1.MetricsViewSpec) (string, []any, error) {
+func buildHavingClause(filter *runtimev1.MeasureFilter, mv *runtimev1.MetricsViewSpec, hasComparison bool) (string, []any, error) {
+	if filter.Expression == nil {
+		return "", []any{}, nil
+	}
+	return buildHavingClauseFromExpression(filter.Expression, mv, hasComparison)
+}
+
+func buildHavingClauseFromNode(node *runtimev1.MeasureFilterNode, mv *runtimev1.MetricsViewSpec, hasComparison bool) (string, []any, error) {
 	sql := ""
 	args := make([]any, 0)
-	switch e := filter.Entry.(type) {
-	case *runtimev1.MeasureFilter_MeasureFilterEntry:
-		switch e.MeasureFilterEntry.Measure.BuiltinMeasure {
-		case runtimev1.BuiltinMeasure_BUILTIN_MEASURE_UNSPECIFIED:
-			expr, err := metricsViewMeasureExpression(mv, e.MeasureFilterEntry.Measure.Name)
-			if err != nil {
-				return "", args, err
-			}
-			sql = fmt.Sprintf(`%s %s ?`, expr, measureFilterClauseOperation(e.MeasureFilterEntry))
-			arg, err := pbutil.FromValue(e.MeasureFilterEntry.Value)
-			if err != nil {
-				return "", args, err
-			}
-			args = append(args, arg)
-			// TODO: comparison
+	switch e := node.Entry.(type) {
+	case *runtimev1.MeasureFilterNode_MeasureFilterMeasure:
+		expr, subArgs, err := buildHavingClauseFromMeasure(e.MeasureFilterMeasure, mv, hasComparison)
+		if err != nil {
+			return "", args, err
+		}
+		args = append(args, subArgs...)
+		sql += expr
 
-		case runtimev1.BuiltinMeasure_BUILTIN_MEASURE_COUNT:
-			//TODO: impl
-		case runtimev1.BuiltinMeasure_BUILTIN_MEASURE_COUNT_DISTINCT:
-			//TODO: impl
+	case *runtimev1.MeasureFilterNode_MeasureFilterExpression:
+		expr, subArgs, err := buildHavingClauseFromExpression(e.MeasureFilterExpression, mv, hasComparison)
+		if err != nil {
+			return "", args, err
 		}
+		args = append(args, subArgs...)
+		sql += expr
 
-	case *runtimev1.MeasureFilter_MeasureFilterExpression:
-		exprs := make([]string, len(e.MeasureFilterExpression.Entries))
-		for i, e := range e.MeasureFilterExpression.Entries {
-			expr, subArgs, err := buildHavingClause(e, mv)
-			if err != nil {
-				return "", args, err
+	case *runtimev1.MeasureFilterNode_Value:
+		arg, err := pbutil.FromValue(e.Value)
+		if err != nil {
+			return "", args, err
+		}
+		args = append(args, arg)
+		sql += "?"
+	}
+
+	return sql, args, nil
+}
+
+func buildHavingClauseFromMeasure(measure *runtimev1.MeasureFilterMeasure, mv *runtimev1.MetricsViewSpec, hasComparison bool) (string, []any, error) {
+	expr := ""
+	args := make([]any, 0)
+	switch measure.Measure.BuiltinMeasure {
+	case runtimev1.BuiltinMeasure_BUILTIN_MEASURE_UNSPECIFIED:
+		if measure.ColumnType != runtimev1.MeasureFilterMeasure_COLUMN_TYPE_UNSPECIFIED && !hasComparison {
+			return "", args, fmt.Errorf("comparison filter cannot be applied when it is not enabled/supported")
+		}
+
+		var ms *runtimev1.MetricsViewSpec_MeasureV2
+		for _, m := range mv.Measures {
+			if strings.EqualFold(m.Name, measure.Measure.Name) {
+				ms = m
+				break
 			}
-			args = append(args, subArgs...)
-			exprs[i] = expr
 		}
-		joiner := ""
-		switch e.MeasureFilterExpression.Joiner {
-		case runtimev1.MeasureFilterExpression_JOINER_UNSPECIFIED:
-		case runtimev1.MeasureFilterExpression_JOINER_OR:
-			joiner = " OR "
-		case runtimev1.MeasureFilterExpression_JOINER_AND:
-			joiner = " AND "
+		if ms == nil {
+			return "", args, fmt.Errorf("measure %s not found", measure.Measure.Name)
+		}
+
+		switch measure.ColumnType {
+		case runtimev1.MeasureFilterMeasure_COLUMN_TYPE_UNSPECIFIED:
+			expr += ms.Name
+		case runtimev1.MeasureFilterMeasure_COLUMN_TYPE_PREVIOUS:
+			expr += ms.Name + "__previous"
+		case runtimev1.MeasureFilterMeasure_COLUMN_TYPE_DELTA_ABSOLUTE:
+			expr += ms.Name + "__delta_abs"
+		case runtimev1.MeasureFilterMeasure_COLUMN_TYPE_DELTA_RELATIVE:
+			expr += ms.Name + "__delta_rel"
 		}
-		sql = strings.Join(exprs, joiner)
+
+	case runtimev1.BuiltinMeasure_BUILTIN_MEASURE_COUNT:
+		//TODO: impl
+	case runtimev1.BuiltinMeasure_BUILTIN_MEASURE_COUNT_DISTINCT:
+		//TODO: impl
+	}
+	return expr, args, nil
+}
+
+func buildHavingClauseFromExpression(expr *runtimev1.MeasureFilterExpression, mv *runtimev1.MetricsViewSpec, hasComparison bool) (string, []any, error) {
+	args := make([]any, 0)
+	if len(expr.Entries) < 2 {
+		return "", args, fmt.Errorf("exactly 2 entries should be provided")
+	}
+
+	leftExpr, subArgs, err := buildHavingClauseFromNode(expr.Entries[0], mv, hasComparison)
+	if err != nil {
+		return "", args, err
+	}
+	args = append(args, subArgs...)
+	rightExpr, subArgs, err := buildHavingClauseFromNode(expr.Entries[1], mv, hasComparison)
+	if err != nil {
+		return "", args, err
 	}
+	args = append(args, subArgs...)
 
+	sql := fmt.Sprintf("(%s) %s (%s)", leftExpr, measureFilterClauseOperation(expr), rightExpr)
 	return sql, args, nil
 }
 
-func measureFilterClauseOperation(e *runtimev1.MeasureFilterEntry) string {
+func measureFilterClauseOperation(e *runtimev1.MeasureFilterExpression) string {
 	switch e.OperationType {
-	case runtimev1.MeasureFilterEntry_OPERATION_TYPE_EQUALS:
+	case runtimev1.MeasureFilterExpression_OPERATION_TYPE_EQUALS:
 		return "="
-	case runtimev1.MeasureFilterEntry_OPERATION_TYPE_NOT_EQUALS:
+	case runtimev1.MeasureFilterExpression_OPERATION_TYPE_NOT_EQUALS:
 		return "!="
-	case runtimev1.MeasureFilterEntry_OPERATION_TYPE_LESSER:
+	case runtimev1.MeasureFilterExpression_OPERATION_TYPE_LESSER:
 		return "<"
-	case runtimev1.MeasureFilterEntry_OPERATION_TYPE_LESSER_OR_EQUALS:
+	case runtimev1.MeasureFilterExpression_OPERATION_TYPE_LESSER_OR_EQUALS:
 		return "<="
-	case runtimev1.MeasureFilterEntry_OPERATION_TYPE_GREATER:
+	case runtimev1.MeasureFilterExpression_OPERATION_TYPE_GREATER:
 		return ">"
-	case runtimev1.MeasureFilterEntry_OPERATION_TYPE_GREATER_OR_EQUALS:
+	case runtimev1.MeasureFilterExpression_OPERATION_TYPE_GREATER_OR_EQUALS:
 		return ">="
+	case runtimev1.MeasureFilterExpression_OPERATION_TYPE_OR:
+		return "OR"
+	case runtimev1.MeasureFilterExpression_OPERATION_TYPE_AND:
+		return "AND"
+	case runtimev1.MeasureFilterExpression_OPERATION_TYPE_BETWEEN:
+		return "BETWEEN"
 	}
 	return "=" // TODO: handle unknown operation type
 }

runtime/queries/metricsview_aggregation.go

@@ -227,7 +227,7 @@ func (q *MetricsViewAggregation) buildMetricsAggregationSQL(mv *runtimev1.Metric
 	if q.MeasureFilter != nil {
 		var havingClauseArgs []any
 		var err error
-		havingClause, havingClauseArgs, err = buildHavingClause(q.MeasureFilter, mv)
+		havingClause, havingClauseArgs, err = buildHavingClause(q.MeasureFilter, mv, false)
 		if err != nil {
 			return "", nil, err
 		}

runtime/queries/metricsview_aggregation_test.go

@@ -57,12 +57,22 @@ func TestMetricsViewAggregation_measure_filters(t *testing.T) {
 		},
 		Limit: &lmt,
 		MeasureFilter: &runtimev1.MeasureFilter{
-			Entry: &runtimev1.MeasureFilter_MeasureFilterEntry{
-				MeasureFilterEntry: &runtimev1.MeasureFilterEntry{
-					Measure:       &runtimev1.MetricsViewAggregationMeasure{Name: "measure_1"},
-					OperationType: runtimev1.MeasureFilterEntry_OPERATION_TYPE_GREATER,
-					Value:         structpb.NewNumberValue(3.25),
+			Expression: &runtimev1.MeasureFilterExpression{
+				Entries: []*runtimev1.MeasureFilterNode{
+					{
+						Entry: &runtimev1.MeasureFilterNode_MeasureFilterMeasure{
+							MeasureFilterMeasure: &runtimev1.MeasureFilterMeasure{
+								Measure: &runtimev1.MetricsViewAggregationMeasure{Name: "measure_1"},
+							},
+						},
+					},
+					{
+						Entry: &runtimev1.MeasureFilterNode_Value{
+							Value: structpb.NewNumberValue(3.25),
+						},
+					},
 				},
+				OperationType: runtimev1.MeasureFilterExpression_OPERATION_TYPE_GREATER,
 			},
 		},
 	}

runtime/queries/metricsview_comparison_toplist.go

@@ -6,14 +6,14 @@
 	"fmt"
 	"io"
 	"strings"
 	// Load IANA time zone data
 	_ "time/tzdata"
 
 	runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1"
 	"github.com/rilldata/rill/runtime"
 	"github.com/rilldata/rill/runtime/drivers"
 	"github.com/rilldata/rill/runtime/pkg/pbutil"
 	"google.golang.org/protobuf/types/known/structpb"
 )
 
 type MetricsViewComparison struct {
@@ -304,7 +304,7 @@
 	havingClause := ""
 	if q.MeasureFilter != nil {
 		var havingClauseArgs []any
-		havingClause, havingClauseArgs, err = buildHavingClause(q.MeasureFilter, mv)
+		havingClause, havingClauseArgs, err = buildHavingClause(q.MeasureFilter, mv, false)
 		if err != nil {
 			return "", nil, err
 		}
@@ -440,7 +440,7 @@
 		var labelTuple string
 		if dialect != drivers.DialectDruid {
 			columnsTuple = fmt.Sprintf(
-				"base.%[1]s, comparison.%[1]s AS %[2]s, base.%[1]s - comparison.%[1]s AS %[3]s, (base.%[1]s - comparison.%[1]s)/comparison.%[1]s::DOUBLE AS %[4]s",
+				"sum(base.%[1]s) as %[1]s, sum(comparison.%[1]s) AS %[2]s, sum(base.%[1]s - comparison.%[1]s) AS %[3]s, sum((base.%[1]s - comparison.%[1]s)/comparison.%[1]s::DOUBLE) AS %[4]s",
 				safeName(m.Name),
 				safeName(m.Name+"__previous"),
 				safeName(m.Name+"__delta_abs"),
@@ -524,6 +524,17 @@
 		args = append(args, clauseArgs...)
 	}
 
+	havingClause := ""
+	if q.MeasureFilter != nil {
+		var havingClauseArgs []any
+		havingClause, havingClauseArgs, err = buildHavingClause(q.MeasureFilter, mv, true)
+		if err != nil {
+			return "", nil, err
+		}
+		havingClause = "HAVING " + havingClause
+		args = append(args, havingClauseArgs...)
+	}
+
 	err = validateSort(q.Sort)
 	if err != nil {
 		return "", nil, err
@@ -655,6 +666,8 @@
 			) comparison
 		ON
 				base.%[2]s = comparison.%[2]s OR (base.%[2]s is null and comparison.%[2]s is null)
+		GROUP BY 1
+		%[16]s
 		ORDER BY
 			%[6]s
 		%[7]s
@@ -676,6 +689,7 @@
 			comparisonLimitClause, // 13
 			unnestClause,          // 14
 			groupByCol,            // 15
+			havingClause,          // 16
 		)
 	} else {
 		/*