feat(trends): add moving average smoothing_intervals param

.devcontainer/.env

@@ -21,3 +21,6 @@ REDIS_URL=redis://redis:6379
 CLICKHOUSE_HOST=clickhouse
 CLICKHOUSE_DATABASE=posthog_test
 CLICKHOUSE_VERIFY=False
+
+# Setup redis config
+REDIS_URL=redis://redis:6379/

ee/clickhouse/queries/test/__snapshots__/test_trends.ambr

@@ -449,7 +449,7 @@
              AND (has(['finance'], replaceRegexpAll(JSONExtractRaw(group_properties_0, 'industry'), '^"|"$', ''))) )
         GROUP BY toStartOfDay(timestamp))
      group by day_start
-     order by day_start) SETTINGS timeout_before_checking_execution_speed = 60
+     order by day_start SETTINGS allow_experimental_window_functions = 1) SETTINGS timeout_before_checking_execution_speed = 60
   '
 ---
 # name: TestClickhouseTrends.test_trend_breakdown_user_props_with_filter_with_partial_property_pushdowns

ee/clickhouse/queries/trends/total_volume.py

@@ -67,7 +67,27 @@ def _total_volume_query(self, entity: Entity, filter: Filter, team: Team) -> Tup
 
             null_sql = NULL_SQL.format(trunc_func=trunc_func, interval_func=interval_func)
             params["interval"] = filter.interval
-            final_query = AGGREGATE_SQL.format(null_sql=null_sql, content_sql=content_sql)
+
+            # If we have a smoothing interval > 1 then add in the sql to
+            # handling rolling average. Else just do a sum. This is possibly an
+            # nessacary optimization.
+            if filter.smoothing_intervals > 1:
+                smoothing_operation = f"""
+                    AVG(SUM(total))
+                    OVER (
+                        ORDER BY day_start
+                        ROWS BETWEEN {filter.smoothing_intervals - 1} PRECEDING
+                        AND CURRENT ROW
+                    )"""
+            else:
+                smoothing_operation = "SUM(total)"
+
+            final_query = AGGREGATE_SQL.format(
+                null_sql=null_sql,
+                content_sql=content_sql,
+                smoothing_operation=smoothing_operation,
+                aggregate="count" if filter.smoothing_intervals < 2 else "floor(count)",
+            )
             return final_query, params, self._parse_total_volume_result(filter, entity, team.id)
 
     def _parse_total_volume_result(self, filter: Filter, entity: Entity, team_id: int) -> Callable:

ee/clickhouse/sql/trends/volume.py

@@ -20,8 +20,16 @@
 """
 
 AGGREGATE_SQL = """
-SELECT groupArray(day_start) as date, groupArray(count) as data FROM (
-    SELECT SUM(total) AS count, day_start from ({null_sql} UNION ALL {content_sql}) group by day_start order by day_start
+SELECT groupArray(day_start) as date, groupArray({aggregate}) as data FROM (
+    SELECT {smoothing_operation} AS count, day_start
+    from (
+        {null_sql}
+        UNION ALL
+        {content_sql}
+    )
+    group by day_start
+    order by day_start
+    SETTINGS allow_experimental_window_functions = 1
 )
 SETTINGS timeout_before_checking_execution_speed = 60
 """

ee/clickhouse/views/test/__snapshots__/test_clickhouse_trends.ambr

@@ -76,7 +76,7 @@
              AND timestamp <= '2012-01-15 23:59:59' )
         GROUP BY toStartOfDay(timestamp))
      group by day_start
-     order by day_start) SETTINGS timeout_before_checking_execution_speed = 60
+     order by day_start SETTINGS allow_experimental_window_functions = 1) SETTINGS timeout_before_checking_execution_speed = 60
   '
 ---
 # name: ClickhouseTestTrends.test_insight_trends_basic.1
@@ -138,7 +138,7 @@
              AND (has(['val'], replaceRegexpAll(JSONExtractRaw(e.properties, 'key'), '^"|"$', ''))) )
         GROUP BY toStartOfDay(timestamp))
      group by day_start
-     order by day_start) SETTINGS timeout_before_checking_execution_speed = 60
+     order by day_start SETTINGS allow_experimental_window_functions = 1) SETTINGS timeout_before_checking_execution_speed = 60
   '
 ---
 # name: ClickhouseTestTrends.test_insight_trends_clean_arg.1
@@ -200,7 +200,7 @@
              AND timestamp <= '2012-01-15 23:59:59' )
         GROUP BY toStartOfDay(timestamp))
      group by day_start
-     order by day_start) SETTINGS timeout_before_checking_execution_speed = 60
+     order by day_start SETTINGS allow_experimental_window_functions = 1) SETTINGS timeout_before_checking_execution_speed = 60
   '
 ---
 # name: ClickhouseTestTrends.test_insight_trends_cumulative.1
@@ -272,7 +272,7 @@
            GROUP BY person_id)
         GROUP BY toStartOfDay(timestamp))
      group by day_start
-     order by day_start) SETTINGS timeout_before_checking_execution_speed = 60
+     order by day_start SETTINGS allow_experimental_window_functions = 1) SETTINGS timeout_before_checking_execution_speed = 60
   '
 ---
 # name: ClickhouseTestTrends.test_insight_trends_cumulative.3
@@ -547,7 +547,7 @@
              AND (NOT has([''], "$group_0")) )
         GROUP BY toStartOfDay(timestamp))
      group by day_start
-     order by day_start) SETTINGS timeout_before_checking_execution_speed = 60
+     order by day_start SETTINGS allow_experimental_window_functions = 1) SETTINGS timeout_before_checking_execution_speed = 60
   '
 ---
 # name: ClickhouseTestTrendsGroups.test_aggregating_by_group.1

ee/clickhouse/views/test/test_clickhouse_stickiness.py

@@ -1,14 +1,14 @@
 from datetime import timedelta
 from uuid import uuid4
 
+from django.test.client import Client
 from freezegun.api import freeze_time
 
 from ee.clickhouse.models.event import create_event
 from ee.clickhouse.models.group import create_group
 from ee.clickhouse.queries.stickiness.clickhouse_stickiness import ClickhouseStickiness
 from ee.clickhouse.util import ClickhouseTestMixin, snapshot_clickhouse_queries
 from posthog.api.test.test_stickiness import get_stickiness_time_series_ok, stickiness_test_factory
-from posthog.api.test.test_trends import get_people_from_url_ok
 from posthog.models.action import Action
 from posthog.models.action_step import ActionStep
 from posthog.models.person import Person
@@ -34,6 +34,12 @@ def _create_person(**kwargs):
     return Person(id=person.uuid)
 
 
+def get_people_from_url_ok(client: Client, url: str):
+    response = client.get("/" + url)
+    assert response.status_code == 200, response.content
+    return response.json()["results"][0]["people"]
+
+
 class TestClickhouseStickiness(ClickhouseTestMixin, stickiness_test_factory(ClickhouseStickiness, _create_event, _create_person, _create_action, get_earliest_timestamp)):  # type: ignore
     @snapshot_clickhouse_queries
     def test_filter_by_group_properties(self):