[core][fix] Timeseries apply aggregation on slotted values (#2033)

Author: aquamatthias

fixcore/fixcore/db/arango_query.py

@@ -895,9 +895,19 @@ def load_time_series(
     slotter = int(granularity.total_seconds())
     gran = ctx.add_bind_var(slotter)
     offset = start.timestamp() - ((start.timestamp() // slotter) * slotter)
+    # slot the time by averaging each single group
     query += f" LET {time_slot} = (FLOOR(d.at / @{gran}) * @{gran}) + @{ctx.add_bind_var(offset)}"
+    query += f" COLLECT group_slot={time_slot}, complete_group=d.group"
+    query += " AGGREGATE slot_avg = AVG(d.v)"
+    query += " RETURN {at: group_slot, group: complete_group, v: slot_avg}"
+
+    # short circuit: no additional grouping and aggregation is avg
+    if group_by is None and group_aggregation == "avg":
+        return query, ctx.bind_vars  # already the correct query
+
     # create the groups to collect
-    collect = [f"group_slot={time_slot}"]
+    slotted = ctx.next_crs()
+    collect = ["group_slot=d.at"]
     group = ""
     if group_by is None:
         collect.append("complete_group=d.group")
@@ -911,8 +921,10 @@ def load_time_series(
             parts.append(f"{g}: group_{g}")
         group = f"group: {{ {', '.join(parts)} }},"
 
-    query += f" COLLECT {', '.join(collect)} INTO group"
-    query += f" SORT group_slot RETURN {{at: group_slot, {group} v: {group_aggregation}(group[*].d.v)}}"
+    query = f"LET {slotted} = ( {query} )\n"
+    query += f" FOR d in {slotted} COLLECT {', '.join(collect)}"
+    query += f" AGGREGATE agg_val={group_aggregation}(d.v)"
+    query += f" SORT group_slot RETURN {{at: group_slot,{group} v: agg_val}}"
     return query, ctx.bind_vars
 
 

fixcore/tests/fixcore/db/arango_query_test.py

@@ -320,43 +320,51 @@ def test_load_time_series() -> None:
     # group_by=[] --> no group by any value
     q, bv = load_time_series("ts", "foo", now - (24 * one_hour), now, one_hour, group_by=[])
     assert (
-        q == "FOR d in `ts` FILTER d.ts==@b0 AND d.at>=@b1 AND d.at<@b2 "
+        q == "LET m1 = ( FOR d in `ts` FILTER d.ts==@b0 AND d.at>=@b1 AND d.at<@b2 "
         "LET m0 = (FLOOR(d.at / @b3) * @b3) + @b4 "
-        "COLLECT group_slot=m0 INTO group "
-        "SORT group_slot "
-        "RETURN {at: group_slot,  v: avg(group[*].d.v)}"
+        "COLLECT group_slot=m0, complete_group=d.group "
+        "AGGREGATE slot_avg = AVG(d.v) "
+        "RETURN {at: group_slot, group: complete_group, v: slot_avg} )\n "
+        "FOR d in m1 COLLECT group_slot=d.at AGGREGATE agg_val=avg(d.v) "
+        "SORT group_slot RETURN {at: group_slot, v: agg_val}"
     )
     assert bv == {"b0": "foo", "b1": 1699913600, "b2": 1700000000, "b3": 3600, "b4": 800}
     # no group by defined --> group by all values
     q, bv = load_time_series("ts", "foo", now - (24 * one_hour), now, one_hour)
     assert (
         q == "FOR d in `ts` FILTER d.ts==@b0 AND d.at>=@b1 AND d.at<@b2 "
         "LET m0 = (FLOOR(d.at / @b3) * @b3) + @b4 "
-        "COLLECT group_slot=m0, complete_group=d.group INTO group "
-        "SORT group_slot "
-        "RETURN {at: group_slot, group: complete_group, v: avg(group[*].d.v)}"
+        "COLLECT group_slot=m0, complete_group=d.group "
+        "AGGREGATE slot_avg = AVG(d.v) "
+        "RETURN {at: group_slot, group: complete_group, v: slot_avg}"
     )
     assert bv == {"b0": "foo", "b1": 1699913600, "b2": 1700000000, "b3": 3600, "b4": 800}
     # group by specific group variables
     q, bv = load_time_series("ts", "foo", now - (24 * one_hour), now, one_hour, group_by=["a", "b"])
     assert (
-        q == "FOR d in `ts` FILTER d.ts==@b0 AND d.at>=@b1 AND d.at<@b2 "
+        q == "LET m1 = ( FOR d in `ts` FILTER d.ts==@b0 AND d.at>=@b1 AND d.at<@b2 "
         "LET m0 = (FLOOR(d.at / @b3) * @b3) + @b4 "
-        "COLLECT group_slot=m0, group_a=d.group.a, group_b=d.group.b INTO group "
-        "SORT group_slot "
-        "RETURN {at: group_slot, group: { a: group_a, b: group_b }, v: avg(group[*].d.v)}"
+        "COLLECT group_slot=m0, complete_group=d.group "
+        "AGGREGATE slot_avg = AVG(d.v) "
+        "RETURN {at: group_slot, group: complete_group, v: slot_avg} )\n "
+        "FOR d in m1 "
+        "COLLECT group_slot=d.at, group_a=d.group.a, group_b=d.group.b "
+        "AGGREGATE agg_val=avg(d.v) "
+        "SORT group_slot RETURN {at: group_slot,group: { a: group_a, b: group_b }, v: agg_val}"
     )
     assert bv == {"b0": "foo", "b1": 1699913600, "b2": 1700000000, "b3": 3600, "b4": 800}
     # group by specific group variables and filter by group variables
     q, bv = load_time_series(
         "ts", "foo", now - (24 * one_hour), now, one_hour, group_by=["a", "b"], group_filter=[P("a").eq("a")]
     )
     assert (
-        q == "FOR d in `ts` FILTER d.ts==@b0 AND d.at>=@b1 AND d.at<@b2 "
-        "FILTER d.group.a==@b3 "
+        q == "LET m1 = ( FOR d in `ts` FILTER d.ts==@b0 AND d.at>=@b1 AND d.at<@b2 FILTER d.group.a==@b3 "
         "LET m0 = (FLOOR(d.at / @b4) * @b4) + @b5 "
-        "COLLECT group_slot=m0, group_a=d.group.a, group_b=d.group.b INTO group "
-        "SORT group_slot "
-        "RETURN {at: group_slot, group: { a: group_a, b: group_b }, v: avg(group[*].d.v)}"
+        "COLLECT group_slot=m0, complete_group=d.group "
+        "AGGREGATE slot_avg = AVG(d.v) RETURN {at: group_slot, group: complete_group, v: slot_avg} )\n "
+        "FOR d in m1 "
+        "COLLECT group_slot=d.at, group_a=d.group.a, group_b=d.group.b "
+        "AGGREGATE agg_val=avg(d.v) "
+        "SORT group_slot RETURN {at: group_slot,group: { a: group_a, b: group_b }, v: agg_val}"
     )
     assert bv == {"b0": "foo", "b1": 1699913600, "b2": 1700000000, "b3": "a", "b4": 3600, "b5": 800}