Merge pull request #1882 from useautumn/feat/events-customer-hourly-mv

customer-leading events_customer_hourly_mv for per-customer aggregates

server/src/internal/analytics/actions/getCountAndSum.ts

@@ -129,7 +129,7 @@ export const getCountAndSum = async ({
 		`
 		: `
 			SELECT event_name, sum(event_count) as count, sum(total_value) as sum
-			FROM ${useOrgRollup ? "events_org_hourly_mv" : "events_hourly_no_properties_two_mv"}
+			FROM ${useOrgRollup ? "events_org_hourly_mv" : "events_customer_hourly_mv"}
 			WHERE org_id = {org_id:String} AND env = {env:String}
 				${!useOrgRollup && !params.aggregateAll ? "AND customer_id = {customer_id:String}" : ""}
 				${!useOrgRollup && params.entity_id ? "AND entity_id = {entity_id:String}" : ""}

server/tinybird/copies/events_customer_hourly_mv_backfill.pipe

@@ -0,0 +1,34 @@
+DESCRIPTION >
+    History backfill for events_customer_hourly_mv (deployed with BACKFILL skip). Fills the window before
+    the MV's forward-materialization seam, in bounded half-open [start_date, end_date) chunks on on-demand
+    compute
+    (tb copy run events_customer_hourly_mv_backfill --on-demand-compute --param start_date=... --param end_date=...).
+    SQL mirrors events_customer_hourly_mv_pipe EXACTLY (same projection + GROUP BY, no ARRAY JOIN) so
+    backfilled rows are identical to forward-materialized ones. No fan-out, so chunks can be wide.
+
+    Seam safety: MergeTree does NOT dedup — keep every end_date <= the MV's actual promote time
+    (capture it empirically: min(hour) once forward events land), else overlapping hours double-count.
+    Only ~100 days of history is needed to cover the max servable window (the 92-day custom_range clamp
+    plus billing-cycle headroom); backfilling further back is wasted storage.
+
+NODE migrate
+SQL >
+    %
+    SELECT
+        org_id,
+        env,
+        customer_id,
+        event_name,
+        coalesce(entity_id, '') as entity_id,
+        internal_product_id,
+        toStartOfHour(timestamp) as hour,
+        sum(toFloat64(coalesce(value, 1))) as total_value,
+        count() as event_count
+    FROM events
+    WHERE timestamp >= {{DateTime(start_date, '2026-06-01 00:00:00')}}
+        AND timestamp < {{DateTime(end_date, '2026-06-02 00:00:00')}}
+    GROUP BY org_id, env, customer_id, event_name, entity_id, internal_product_id, hour
+
+TYPE COPY
+TARGET_DATASOURCE events_customer_hourly_mv
+COPY_MODE append

server/tinybird/materializations/events_customer_hourly_mv.datasource

@@ -0,0 +1,31 @@
+DESCRIPTION >
+    Customer-leading hourly rollup of `events`, keyed (org_id, env, customer_id, entity_id, event_name, hour).
+    Same grain and projection as events_hourly_no_properties_two_mv, but with customer_id LEADING the sort key
+    so a per-customer (or per-customer+entity) query prunes straight to that customer's slice instead of
+    scanning the whole org's hourly partitions — events_hourly_no_properties_two_mv leads with `hour`, so a
+    per-customer query there cannot prune by customer and scans the entire org/env date window.
+
+    Serves the per-customer ungrouped timeseries + count/sum and the group-by customer_id/entity_id/plan_id
+    paths once the read pipes (aggregate_simple, aggregate_groupable, getCountAndSum) are repointed at it.
+    Totals reconcile exactly with events_hourly_no_properties_two_mv because both pre-aggregate one row per
+    event at insert and the read pipe re-sums by (period, event_name, group_value).
+
+    BACKFILL skip: deploy empty + forward-materialization trigger only (no deploy-time repopulate of the
+    existing events history); fill history in bounded chunks via events_customer_hourly_mv_backfill.
+
+SCHEMA >
+    `org_id` String,
+    `env` String,
+    `customer_id` String,
+    `event_name` String,
+    `entity_id` String DEFAULT '',
+    `internal_product_id` Nullable(String),
+    `hour` DateTime,
+    `total_value` Float64,
+    `event_count` UInt64
+
+ENGINE "MergeTree"
+ENGINE_PARTITION_KEY "toYYYYMM(hour)"
+ENGINE_SORTING_KEY "org_id, env, customer_id, entity_id, event_name, hour"
+
+BACKFILL skip

server/tinybird/materializations/events_customer_hourly_mv_pipe.pipe

@@ -0,0 +1,22 @@
+DESCRIPTION >
+    Materializes events into customer-leading hourly aggregates without properties.
+    Projection + GROUP BY are identical to events_hourly_no_properties_two_mv_pipe, so the two rollups
+    are row-for-row equivalent; only the destination sort key differs (customer-leading here).
+
+NODE materialize
+SQL >
+    SELECT
+        org_id,
+        env,
+        customer_id,
+        event_name,
+        coalesce(entity_id, '') as entity_id,
+        internal_product_id,
+        toStartOfHour(timestamp) as hour,
+        sum(toFloat64(coalesce(value, 1))) as total_value,
+        count() as event_count
+    FROM events
+    GROUP BY org_id, env, customer_id, event_name, entity_id, internal_product_id, hour
+
+TYPE materialized
+DATASOURCE events_customer_hourly_mv

server/tinybird/pipes/aggregate_groupable.pipe

@@ -41,7 +41,7 @@ SQL >
         sum(total_value) as total_value
     FROM
         {% if use_no_props %}
-        events_hourly_no_properties_two_mv
+        events_customer_hourly_mv
         {% elif use_property_rollup %}
         events_property_mv
         {% else %}

server/tinybird/pipes/aggregate_simple.pipe

@@ -20,7 +20,7 @@ SQL >
         {% end %}
         event_name,
         sum(total_value) as total_value
-    FROM {% if not no_property_filters %}events_hourly_mv{% elif use_org_rollup %}events_org_hourly_mv{% else %}events_hourly_no_properties_two_mv{% end %}
+    FROM {% if not no_property_filters %}events_hourly_mv{% elif use_org_rollup %}events_org_hourly_mv{% else %}events_customer_hourly_mv{% end %}
     WHERE
         org_id = {{ String(org_id, '') }}
         AND env = {{ String(env, 'test') }}