core: no data for aggr of empty group by (#1609)

Make the behavior of using an aggregate on an empty group by result consistent with aggregate on an empty query result.
Netflix · Feb 14, 2024 · fe78706 · fe78706
1 parent d946897
commit fe78706
Show file tree

Hide file tree

Showing 3 changed files with 62 additions and 5 deletions.
diff --git a/atlas-core/src/main/scala/com/netflix/atlas/core/model/MathExpr.scala b/atlas-core/src/main/scala/com/netflix/atlas/core/model/MathExpr.scala
@@ -690,8 +690,9 @@ object MathExpr {
     def eval(context: EvalContext, data: Map[DataExpr, List[TimeSeries]]): ResultSet = {
       val rs = expr.eval(context, data)
       val ts =
-        if (rs.data.isEmpty) Nil
-        else {
+        if (rs.data.isEmpty) {
+          List(TimeSeries.noData(context.step))
+        } else {
           val aggr = aggregator(context.start, context.end)
           rs.data.foreach(aggr.update)
           val t = aggr.result()

diff --git a/atlas-core/src/test/scala/com/netflix/atlas/core/model/TimeSeriesExprSuite.scala b/atlas-core/src/test/scala/com/netflix/atlas/core/model/TimeSeriesExprSuite.scala
@@ -120,7 +120,7 @@ class TimeSeriesExprSuite extends FunSuite {
     ":true,(,name,),:by,:avg"      -> const(ts(unknownTag, "(name=unknown / count(NO TAGS))", 5)),
     ":true,(,foo,),:by"            -> const(Nil),
     ":false,(,name,),:by"          -> const(Nil),
-    "n,:has,(,n,),:by,4,:div,:sum" -> const(Nil),
+    "n,:has,(,n,),:by,4,:div,:sum" -> const(ts(Map("name" -> "NO_DATA"), "NO DATA", Double.NaN)),
     "NaN,NaN,:add"                 -> const(ts(Map("name" -> "NaN"), "(NaN + NaN)", Double.NaN)),
     "NaN,1.0,:add"                 -> const(ts(Map("name" -> "NaN"), "(NaN + 1.0)", 1.0)),
     "1.0,NaN,:add"                 -> const(ts(Map("name" -> "1.0"), "(1.0 + NaN)", 1.0)),
@@ -184,7 +184,7 @@ class TimeSeriesExprSuite extends FunSuite {
     ":true,1w,:offset"   -> const(ts(unknownTag, "name=unknown (offset=1w)", 55)),
     ":true,5,:add,1w,:offset"    -> const(ts(unknownTag, "(name=unknown (offset=1w) + 5.0)", 60)),
     "issue,283,:eq"              -> const(ts(Map("issue" -> "283"), "NO DATA", Double.NaN)),
-    ":false,(,name,),:by,:count" -> const(Nil),
+    ":false,(,name,),:by,:count" -> const(ts(Map("name" -> "NO_DATA"), "NO DATA", Double.NaN)),
     ":true,(,name,),:by,:stddev" -> const(
       ts(Map.empty[String, String], stddevLegend("NO TAGS"), 3.1622776601683795)
     ),

diff --git a/atlas-eval/src/test/scala/com/netflix/atlas/eval/stream/FinalExprEvalSuite.scala b/atlas-eval/src/test/scala/com/netflix/atlas/eval/stream/FinalExprEvalSuite.scala
@@ -163,7 +163,7 @@ class FinalExprEvalSuite extends FunSuite {
     if (expected.isNaN)
       assert(v.isNaN)
     else
-      assertEquals(v, expected)
+      assertEqualsDouble(v, expected, 1e-9)
   }
 
   test("aggregate with single datapoint per group") {
@@ -461,6 +461,62 @@ class FinalExprEvalSuite extends FunSuite {
     })
   }
 
+  test("empty aggregated group by with binary operation") {
+    val expr1 = DataExpr.GroupBy(DataExpr.Max(Query.Equal("name", "a")), List("node"))
+    val expr2 = DataExpr.GroupBy(DataExpr.Max(Query.Equal("name", "b")), List("node"))
+    val input = List(
+      sources(ds("a", s"http://atlas/graph?q=$expr1,:sum,$expr2,:sum,:add")),
+      group(
+        0, // No data for expr1
+        AggrDatapoint(0, step, expr2, "i-1", Map("name" -> "b"), 1.0)
+      ),
+      group(
+        1,
+        AggrDatapoint(0, step, expr2, "i-1", Map("name" -> "b"), 1.0)
+      )
+    )
+
+    val output = run(input)
+
+    val timeseries = output.filter(isTimeSeries)
+    assertEquals(timeseries.size, 2)
+    timeseries.foreach { env =>
+      val ts = env.message.asInstanceOf[TimeSeriesMessage]
+      checkValue(ts, 1.0)
+    }
+  }
+
+  test("empty aggregated percentile approximation with binary operation") {
+    val expr1 = MathExpr.Percentiles(
+      DataExpr.GroupBy(DataExpr.Sum(Query.Equal("name", "a")), List("percentile")),
+      List(90.0)
+    )
+    val expr2 = MathExpr.Percentiles(
+      DataExpr.GroupBy(DataExpr.Sum(Query.Equal("name", "b")), List("percentile")),
+      List(90.0)
+    )
+    val input = List(
+      sources(ds("a", s"http://atlas/graph?q=$expr1,:sum,$expr2,:sum,:add")),
+      group(
+        0, // No data for expr1
+        AggrDatapoint(0, step, expr2.expr, "i-1", Map("name" -> "b", "percentile" -> "T0000"), 1.0)
+      ),
+      group(
+        1,
+        AggrDatapoint(0, step, expr2.expr, "i-1", Map("name" -> "b", "percentile" -> "T0000"), 1.0)
+      )
+    )
+
+    val output = run(input)
+
+    val timeseries = output.filter(isTimeSeries)
+    assertEquals(timeseries.size, 2)
+    timeseries.foreach { env =>
+      val ts = env.message.asInstanceOf[TimeSeriesMessage]
+      checkValue(ts, 9e-10)
+    }
+  }
+
   test("multi-level group by with stateful operation") {
     val expr = DataExpr.GroupBy(DataExpr.Sum(Query.Equal("name", "rps")), List("node", "app"))
     val input = List(