diff --git a/server/src/internal/analytics/actions/getCountAndSum.ts b/server/src/internal/analytics/actions/getCountAndSum.ts index 92e345e98..6a2e4c332 100644 --- a/server/src/internal/analytics/actions/getCountAndSum.ts +++ b/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}" : ""} diff --git a/server/tinybird/copies/events_customer_hourly_mv_backfill.pipe b/server/tinybird/copies/events_customer_hourly_mv_backfill.pipe new file mode 100644 index 000000000..19256f0ac --- /dev/null +++ b/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 diff --git a/server/tinybird/materializations/events_customer_hourly_mv.datasource b/server/tinybird/materializations/events_customer_hourly_mv.datasource new file mode 100644 index 000000000..cc89319ef --- /dev/null +++ b/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 diff --git a/server/tinybird/materializations/events_customer_hourly_mv_pipe.pipe b/server/tinybird/materializations/events_customer_hourly_mv_pipe.pipe new file mode 100644 index 000000000..225e844d2 --- /dev/null +++ b/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 diff --git a/server/tinybird/pipes/aggregate_groupable.pipe b/server/tinybird/pipes/aggregate_groupable.pipe index c289a3489..0da6fcbbf 100644 --- a/server/tinybird/pipes/aggregate_groupable.pipe +++ b/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 %} diff --git a/server/tinybird/pipes/aggregate_simple.pipe b/server/tinybird/pipes/aggregate_simple.pipe index b5dc9a086..e0f4dc90d 100644 --- a/server/tinybird/pipes/aggregate_simple.pipe +++ b/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') }}