Files
cfw-autumn/server/tinybird/pipes/aggregate_groupable.pipe

109 lines
5.2 KiB
Plaintext

DESCRIPTION >
Aggregate queries with grouping by a property key, customer_id, or entity_id.
Routes to the smallest viable MV for the query shape:
- events_hourly_no_properties_two_mv: customer_id/entity_id grouping with no filters
- events_hourly_promoted_mv: promoted property key (apiKeyId, endpoint, source) with no filters
- events_hourly_mv: fallback for arbitrary properties or filtered queries
Uses PER-BIN TOP N groups + "AUTUMN_RESERVED" bucket ((N+1) max total per bin).
N is controlled by the max_groups parameter (default 9).
Returns unpivoted data: (period, event_name, group_value, total_value, _truncated)
TOKEN "aggregate_groupable_read" READ
NODE base
DESCRIPTION >
Aggregate raw data by period, event_name, and group_value.
SQL >
%
{% set no_filters = (not defined(filter_key_0) or String(filter_key_0, '') == '') and (not defined(filter_key_1) or String(filter_key_1, '') == '') and (not defined(filter_key_2) or String(filter_key_2, '') == '') and (not defined(filter_key_3) or String(filter_key_3, '') == '') and (not defined(filter_key_4) or String(filter_key_4, '') == '') %}
{% set is_promoted_key = String(property_key, '') in ['apiKeyId', 'endpoint', 'source'] %}
{% set use_no_props = (String(group_column, 'property') in ['customer_id', 'entity_id']) and no_filters %}
{% set use_promoted = (String(group_column, 'property') == 'property') and is_promoted_key and no_filters %}
SELECT
{% if String(bin_size, 'day') == 'hour' %}
formatDateTime(hour, '%F %T') as period,
{% elif String(bin_size, 'day') == 'month' %}
formatDateTime(toStartOfMonth(hour, {{ String(timezone, 'UTC') }}), '%F %T') as period,
{% else %}
formatDateTime(toStartOfDay(hour, {{ String(timezone, 'UTC') }}), '%F %T') as period,
{% end %}
event_name,
{% if String(group_column, 'property') == 'customer_id' %}
customer_id as group_value,
{% elif String(group_column, 'property') == 'entity_id' %}
entity_id as group_value,
{% elif use_promoted %}
{{ column(String(property_key, '')) }} as group_value,
{% else %}
{{ column('properties.' + String(property_key, '')) }}::String as group_value,
{% end %}
sum(total_value) as total_value
FROM
{% if use_no_props %}
events_hourly_no_properties_two_mv
{% elif use_promoted %}
events_hourly_promoted_mv
{% else %}
events_hourly_mv
{% end %}
WHERE
org_id = {{ String(org_id, '') }}
AND env = {{ String(env, 'test') }}
AND event_name IN {{ Array(event_names, 'String', default='[]') }}
AND hour >= toDateTime({{ String(start_date, '2024-01-01 00:00:00') }})
AND hour <= toDateTime({{ String(end_date, '2024-12-31 23:59:59') }})
{% if defined(customer_id) and String(customer_id, '') != '' %}
AND customer_id = {{ String(customer_id) }}
{% end %}
{% if defined(entity_id) and String(entity_id, '') != '' %}
AND entity_id = {{ String(entity_id) }}
{% end %}
{% if defined(filter_key_0) and String(filter_key_0, '') != '' %}
AND {{ column('properties.' + String(filter_key_0, '')) }}::String = {{ String(filter_value_0, '') }}
{% end %}
{% if defined(filter_key_1) and String(filter_key_1, '') != '' %}
AND {{ column('properties.' + String(filter_key_1, '')) }}::String = {{ String(filter_value_1, '') }}
{% end %}
{% if defined(filter_key_2) and String(filter_key_2, '') != '' %}
AND {{ column('properties.' + String(filter_key_2, '')) }}::String = {{ String(filter_value_2, '') }}
{% end %}
{% if defined(filter_key_3) and String(filter_key_3, '') != '' %}
AND {{ column('properties.' + String(filter_key_3, '')) }}::String = {{ String(filter_value_3, '') }}
{% end %}
{% if defined(filter_key_4) and String(filter_key_4, '') != '' %}
AND {{ column('properties.' + String(filter_key_4, '')) }}::String = {{ String(filter_value_4, '') }}
{% end %}
{# Filter out null/empty grouping values #}
{% if String(group_column, 'property') == 'entity_id' %}
AND entity_id IS NOT NULL AND entity_id != ''
{% elif use_promoted %}
AND {{ column(String(property_key, '')) }} != ''
{% elif String(group_column, 'property') == 'property' %}
AND {{ column('properties.' + String(property_key, '')) }}::String IS NOT NULL
AND {{ column('properties.' + String(property_key, '')) }}::String != ''
{% end %}
GROUP BY period, event_name, group_value
NODE ranked
SQL >
SELECT
*,
row_number() OVER (PARTITION BY period, event_name ORDER BY total_value DESC) as rn
FROM base
NODE endpoint
TYPE endpoint
SQL >
%
SELECT
period,
event_name,
if(rn <= {{ Int32(max_groups, 9) }}, group_value, 'AUTUMN_RESERVED') as group_value,
sum(total_value) as total_value,
max(rn) > {{ Int32(max_groups, 9) }} as _truncated
FROM ranked
GROUP BY period, event_name, group_value
ORDER BY period, event_name, group_value