Starrocks Aggregation Optimizer

StarRocks Aggregation Optimizer

When to Use

Load this skill when the user needs to:

• Diagnose slow GROUP BY queries or aggregation bottlenecks in StarRocks
• Understand or control one-phase vs two-phase aggregation planning
• Optimize COUNT DISTINCT queries on high-cardinality columns
• Design pre-aggregation into Aggregate Key tables or async materialized views
• Use HLL or BITMAP sketches to approximate distinct counts at scale
• Tune aggregation memory usage to prevent OOM on large groupings
• Read and interpret EXPLAIN output for AggregationNode, AggregateNode, pre-aggregation flags
• Understand when spill-to-disk is triggered and how to configure it

Aggregation Architecture

StarRocks uses a two-tier aggregation pipeline across the distributed cluster:

┌───────────────────────────────────────────────────────────────────┐
│  BE node (per tablet)                                             │
│  ┌──────────────┐    ┌─────────────────┐                         │
│  │  OlapScanNode│ →  │ Local AggNode   │  (partial aggregation)  │
│  │  preagg=on   │    │ (per-BE reduce) │                         │
│  └──────────────┘    └────────┬────────┘                         │
└───────────────────────────────┼───────────────────────────────────┘
                                │ Exchange (network shuffle)
                                ▼
┌───────────────────────────────────────────────────────────────────┐
│  FE / Coordinator                                                 │
│  ┌──────────────────────────────┐                                 │
│  │  Global AggNode              │  (final merge / aggregation)   │
│  │  (merge partial results)     │                                 │
│  └──────────────────────────────┘                                 │
└───────────────────────────────────────────────────────────────────┘

Three levels where aggregation work happens

The goal is to eliminate as many rows as possible before they cross the network. The optimizer selects one-phase or two-phase strategy based on statistics and distribution properties.

One-Phase vs Two-Phase Aggregation

One-Phase Aggregation

One-phase aggregation occurs when all rows belonging to a GROUP BY key are guaranteed to reside on the same BE. This happens when:

• The GROUP BY key is identical to or a prefix of the distribution (hash bucket) key
• The distribution key guarantees that hash(group_by_cols) % buckets maps each group to a single BE

With one-phase aggregation:

• No shuffle exchange is needed — each BE aggregates its own complete, non-overlapping slice of the data
• The coordinator collects already-final results from each BE and unions them
• EXPLAIN shows a single AggregationNode (or ONE_PHASE_LOCAL_AGGREGATE label) above the scan

EXPLAIN SELECT user_id, SUM(amount)
FROM orders                          -- DISTRIBUTED BY HASH(user_id)
GROUP BY user_id;

-- EXPLAIN output (simplified):
PLAN FRAGMENT 0
  OUTPUT EXPRS: user_id | sum(amount)
  PARTITION: UNPARTITIONED

  RESULT SINK

  2:AGGREGATE (update finalize)      ← single AggregationNode, finalize on BE
    group by: user_id
    output: sum(amount)
  1:OlapScanNode
    TABLE: orders
    PREAGGREGATION: ON

Key signals in EXPLAIN:

• AGGREGATE (update finalize) — partial AND final aggregation done on same BE
• PREAGGREGATION: ON — storage-level merge active
• No EXCHANGE node between scan and aggregate

Two-Phase Aggregation

Two-phase aggregation occurs when the GROUP BY keys do not match the distribution key, so rows for the same group can live on multiple BEs:

1. Phase 1 (Partial): each BE builds a local partial aggregate hash table over its own tablets
2. Exchange: partial results are shuffled by GROUP BY key hash to the correct BE
3. Phase 2 (Final/Merge): each target BE receives all partial results for its assigned keys and performs the final merge

EXPLAIN SELECT region, COUNT(*), SUM(revenue)
FROM orders                          -- DISTRIBUTED BY HASH(user_id)
GROUP BY region;

-- EXPLAIN output (simplified):
PLAN FRAGMENT 0
  OUTPUT EXPRS: region | count(*) | sum(revenue)
  PARTITION: UNPARTITIONED

  RESULT SINK

  4:AGGREGATE (merge finalize)       ← Phase 2: global merge on coordinator
    group by: region
    output: count(*), sum(revenue)
  3:EXCHANGE                         ← network shuffle partitioned by region

PLAN FRAGMENT 1
  PARTITION: HASH_PARTITIONED: region

  STREAM DATA SINK
    EXCHANGE ID: 03
    HASH_PARTITIONED: region

  2:AGGREGATE (update serialize)     ← Phase 1: partial agg on each BE
    group by: region
    output: count(*), sum(revenue)
  1:OlapScanNode
    TABLE: orders
    PREAGGREGATION: ON

Key signals:

• AGGREGATE (update serialize) — Phase 1 partial result, serialized for network
• EXCHANGE node with HASH_PARTITIONED: <group_by_cols> — shuffle by group key
• AGGREGATE (merge finalize) — Phase 2 final merge

CBO Selection Logic

The Cost-Based Optimizer (CBO) chooses based on:

Force a specific strategy when the CBO gets it wrong:

-- Force two-phase (useful when CBO picks one-phase but data is skewed)
SET new_planner_agg_stage = 2;

-- Force one-phase
SET new_planner_agg_stage = 1;

-- Let CBO decide (default)
SET new_planner_agg_stage = 0;

Pre-Aggregation Pushdown

What Pre-Aggregation Is

For Aggregate Key tables, StarRocks stores pre-merged rowsets: at load time and compaction time, rows with identical key columns are merged using the declared aggregate functions (SUM, MAX, MIN, REPLACE, HLL_UNION, BITMAP_UNION, etc.).

At scan time, the storage layer can return one merged row per key instead of all raw rowsets — this is called pre-aggregation. It eliminates rows before they ever reach the query engine pipeline, dramatically reducing CPU and memory usage downstream.

Reading PREAGGREGATION in EXPLAIN

OlapScanNode
  TABLE: dws_user_metrics
  PREAGGREGATION: ON        ← storage merges rowsets at scan

vs.

OlapScanNode
  TABLE: dws_user_metrics
  PREAGGREGATION: OFF       ← full raw rowsets returned to engine

When PREAGGREGATION: ON, the scan node emits at most one row per distinct key combination — massive reduction for tables with high load frequency.

Reasons Pre-Aggregation Turns OFF

Pre-aggregation cannot be applied when the query prevents the storage layer from safely merging:

Diagnosis and fix:

-- Check if preagg is ON
EXPLAIN SELECT date, SUM(page_views), SUM(click_count)
FROM dws_daily_stats
GROUP BY date;
-- Look for: PREAGGREGATION: ON

-- This turns preagg OFF — AVG requires raw values:
EXPLAIN SELECT date, AVG(session_duration)
FROM dws_daily_stats   -- session_duration stored as SUM
GROUP BY date;
-- PREAGGREGATION: OFF

-- Fix: store both SUM and COUNT separately in Aggregate Key table
-- Then compute AVG = SUM / COUNT in the query
CREATE TABLE dws_daily_stats (
    date              DATE           NOT NULL,
    user_id           BIGINT         NOT NULL,
    page_views        BIGINT         SUM     DEFAULT '0',
    click_count       BIGINT         SUM     DEFAULT '0',
    session_dur_sum   BIGINT         SUM     DEFAULT '0',  -- SUM for AVG numerator
    session_dur_cnt   BIGINT         SUM     DEFAULT '0'   -- COUNT for AVG denominator
)
AGGREGATE KEY(date, user_id)
DISTRIBUTED BY HASH(user_id) BUCKETS 32;

-- Query AVG correctly:
SELECT date, session_dur_sum / NULLIF(session_dur_cnt, 0) AS avg_session_sec
FROM dws_daily_stats
GROUP BY date;
-- PREAGGREGATION: ON

Aggregate Key Table Auto-Merge

The Aggregate Key table continuously merges rowsets in background compaction. Query-time behavior:

Load batch 1: (user=1, date='2025-01-01', views=100)
Load batch 2: (user=1, date='2025-01-01', views=50)

During compaction → merged: (user=1, date='2025-01-01', views=150)

At query time (before compaction):
  Scan returns 2 rowsets → engine applies SUM → 150
At query time (after compaction):
  Scan returns 1 merged row → engine reads 150 directly
  PREAGGREGATION: ON in both cases — but after compaction it's faster

Force compaction to ensure pre-aggregation efficiency:

-- Manual compaction on a specific partition
ALTER TABLE dws_daily_stats COMPACT PARTITION p20250101;

-- Check compaction status
SHOW PROC '/statuses/be_compaction_tasks';

GROUP BY on Distribution Key

When the GROUP BY clause matches the distribution key, the optimizer recognizes that all rows for each group are co-located on a single BE — no shuffle is necessary.

Identifying Local Aggregation in EXPLAIN

-- Table: DISTRIBUTED BY HASH(user_id)
EXPLAIN SELECT user_id, SUM(amount), COUNT(*) AS txn_count
FROM orders
GROUP BY user_id;

EXPLAIN output signals for local aggregation:

• Single AggregationNode with no preceding EXCHANGE
• Label ONE_PHASE_LOCAL_AGGREGATE or update finalize on same fragment as scan
• Fragment partition type matches distribution key

PLAN FRAGMENT 1
  PARTITION: HASH_PARTITIONED: user_id    ← matches distribution key

  2:AGGREGATE (update finalize)            ← local agg, no shuffle
    group by: user_id
    output: sum(amount), count(*)
  1:OlapScanNode
    TABLE: orders
    PREAGGREGATION: ON

Design for Local Aggregation

Choose the distribution key to match the most frequent GROUP BY pattern:

-- Metric table: most queries GROUP BY user_id
CREATE TABLE dws_user_daily_metrics (
    dt              DATE           NOT NULL,
    user_id         BIGINT         NOT NULL,
    page_views      BIGINT         SUM     DEFAULT '0',
    session_count   BIGINT         SUM     DEFAULT '0',
    revenue         DECIMAL(18,2)  SUM     DEFAULT '0.00'
)
AGGREGATE KEY(dt, user_id)
DISTRIBUTED BY HASH(user_id) BUCKETS 64   -- user_id = distribution key
PARTITION BY RANGE(dt) (
    START ("2025-01-01") END ("2026-01-01") EVERY (INTERVAL 1 DAY)
)
PROPERTIES ("replication_num" = "3");

-- This query gets LOCAL aggregation (no network shuffle):
SELECT user_id, SUM(revenue) AS total_revenue
FROM dws_user_daily_metrics
WHERE dt BETWEEN '2025-01-01' AND '2025-03-31'
GROUP BY user_id;

When the primary aggregation pattern involves multiple columns:

-- If most queries GROUP BY (region, user_id), set distribution key to user_id
-- (region usually has low cardinality — putting it first causes skew)
DISTRIBUTED BY HASH(user_id) BUCKETS 64
-- GROUP BY region, user_id → still two-phase, but user_id hashing helps partial agg

COUNT DISTINCT Optimization

The Problem with Naive COUNT DISTINCT

COUNT(DISTINCT col) on a high-cardinality column is expensive:

1. All rows must be read and de-duplicated
2. De-duplication requires a large hash set in memory
3. In two-phase mode, partial results cannot be merged — COUNT DISTINCT is not decomposable

Multi-Phase COUNT DISTINCT

StarRocks 3.x implements a three-phase strategy for high-cardinality COUNT DISTINCT:

Phase 1 (Partial, per-BE):
  For each (group_key, distinct_col), emit one row
  → reduces to unique (group, value) pairs per BE

Phase 2 (Shuffle):
  Shuffle by (group_key, distinct_col_bucket)
  → each target BE receives a subset of distinct values

Phase 3 (Final merge):
  Count distinct values per group_key

Enable bucket-based parallelism:

-- Session variables for COUNT DISTINCT bucketing
SET enable_distinct_column_bucketization = true;
SET count_distinct_column_buckets = 1024;   -- split distinct space into 1024 virtual buckets

With bucketing enabled, Phase 2 shuffle distributes distinct values across more BEs, reducing the per-node memory footprint.

EXPLAIN with COUNT DISTINCT and bucketing:

EXPLAIN SELECT user_id, COUNT(DISTINCT session_id)
FROM events
GROUP BY user_id;

-- Shows 3 AggregationNode operators in the plan:
-- AggregateNode (update serialize)     ← Phase 1 local partial
-- EXCHANGE HASH_PARTITIONED            ← shuffle by (user_id, session_id bucket)
-- AggregateNode (merge serialize)      ← Phase 2 bucket-level partial count
-- EXCHANGE HASH_PARTITIONED            ← shuffle by user_id
-- AggregateNode (merge finalize)       ← Phase 3 final COUNT

Multiple COUNT DISTINCT Columns

Multiple COUNT DISTINCT on different columns in the same query triggers rewrite to avoid cross-product:

-- Problematic: StarRocks rewrites this internally
SELECT
    dt,
    COUNT(DISTINCT user_id)    AS dau,
    COUNT(DISTINCT session_id) AS sessions
FROM events
GROUP BY dt;

-- StarRocks 3.x rewrites as UNION of two separate COUNT DISTINCT queries
-- or uses multi-column bucketing — let the optimizer handle it
-- Verify with EXPLAIN

-- Alternative: use HLL for approximate counts (see HLL section)
SELECT
    dt,
    approx_count_distinct(user_id)    AS dau_approx,
    approx_count_distinct(session_id) AS sessions_approx
FROM events
GROUP BY dt;

approx_count_distinct()

approx_count_distinct(col) uses HyperLogLog internally. Key properties:

• ~2% relative error (configurable via HLL precision)
• 10–100× faster than exact COUNT(DISTINCT)
• Works on any data type (strings, integers, UUIDs)
• Does not require a special table type — works on Duplicate Key tables

-- Fast approximate DAU with ~2% error
SELECT
    DATE(event_ts)          AS dt,
    approx_count_distinct(user_id)   AS dau,
    approx_count_distinct(device_id) AS unique_devices
FROM ods.page_events
WHERE event_ts >= '2025-01-01'
GROUP BY DATE(event_ts)
ORDER BY dt;

HLL Approximate Aggregation

HLL (HyperLogLog) sketches stored natively in Aggregate Key tables enable approximate COUNT DISTINCT with:

• ~1–2% relative error
• ~10× storage reduction compared to storing raw distinct values
• O(1) merge cost — HLL sketches are mergeable by XOR

Aggregate Key Table with HLL Column

CREATE TABLE dws_user_daily_hll (
    dt          DATE    NOT NULL,
    region      VARCHAR(64),
    -- HLL columns for approximate distinct counts
    user_hll    HLL     HLL_UNION   NOT NULL,   -- approximate distinct users
    device_hll  HLL     HLL_UNION   NOT NULL    -- approximate distinct devices
)
AGGREGATE KEY(dt, region)
DISTRIBUTED BY HASH(region) BUCKETS 16
PARTITION BY RANGE(dt) (
    START ("2025-01-01") END ("2026-01-01") EVERY (INTERVAL 1 DAY)
)
PROPERTIES ("replication_num" = "3");

Loading HLL Data

-- INSERT using hll_hash() to create HLL sketch from a raw value
INSERT INTO dws_user_daily_hll
SELECT
    DATE(event_ts)    AS dt,
    region,
    hll_hash(user_id)   AS user_hll,
    hll_hash(device_id) AS device_hll
FROM ods.page_events
WHERE event_ts >= CURRENT_DATE - INTERVAL 1 DAY;

With StreamLoad from a Routine Load or Spark connector, use the hll_hash function mapping in the column expression.

Querying HLL Columns

-- HLL_UNION_AGG merges sketches across rows, HLL_CARDINALITY returns the estimate
SELECT
    dt,
    region,
    HLL_UNION_AGG(user_hll)                         AS raw_hll_sketch,
    HLL_CARDINALITY(HLL_UNION_AGG(user_hll))        AS approx_dau,
    HLL_CARDINALITY(HLL_UNION_AGG(device_hll))      AS approx_unique_devices
FROM dws_user_daily_hll
WHERE dt BETWEEN '2025-01-01' AND '2025-01-31'
GROUP BY dt, region
ORDER BY dt, region;

-- Roll up across regions — HLL merges are O(1) per sketch:
SELECT
    dt,
    HLL_CARDINALITY(HLL_UNION_AGG(user_hll)) AS total_dau
FROM dws_user_daily_hll
WHERE dt BETWEEN '2025-01-01' AND '2025-01-31'
GROUP BY dt
ORDER BY dt;

HLL Functions Reference

HLL vs Exact COUNT DISTINCT

BITMAP Aggregation

BITMAP provides exact COUNT DISTINCT for integer ID columns by storing a compressed bitset. Each bit position represents whether that integer ID appears in the group.

When to Use BITMAP vs HLL

Aggregate Key Table with BITMAP Column

CREATE TABLE dws_user_daily_bitmap (
    dt          DATE        NOT NULL,
    region      VARCHAR(64) NOT NULL,
    active_users BITMAP     BITMAP_UNION  NOT NULL   -- exact distinct user_id count
)
AGGREGATE KEY(dt, region)
DISTRIBUTED BY HASH(region) BUCKETS 16
PARTITION BY RANGE(dt) (
    START ("2025-01-01") END ("2026-01-01") EVERY (INTERVAL 1 DAY)
)
PROPERTIES ("replication_num" = "3");

Loading BITMAP Data

-- to_bitmap() converts integer to single-bit BITMAP
INSERT INTO dws_user_daily_bitmap
SELECT
    DATE(event_ts)    AS dt,
    region,
    to_bitmap(user_id) AS active_users
FROM ods.page_events
WHERE event_ts >= CURRENT_DATE - INTERVAL 1 DAY
  AND user_id IS NOT NULL
  AND user_id >= 0;   -- to_bitmap requires non-negative integers

For non-integer IDs (e.g., UUIDs, strings), map to integer IDs first using a dimension table, then use to_bitmap().

Querying BITMAP Columns

-- Exact DAU per day/region
SELECT
    dt,
    region,
    BITMAP_UNION_COUNT(active_users) AS exact_dau   -- exact COUNT DISTINCT
FROM dws_user_daily_bitmap
WHERE dt BETWEEN '2025-01-01' AND '2025-01-31'
GROUP BY dt, region;

-- Rollup across regions — BITMAP_UNION merges sets:
SELECT
    dt,
    BITMAP_UNION_COUNT(BITMAP_UNION(active_users)) AS total_dau
FROM dws_user_daily_bitmap
WHERE dt BETWEEN '2025-01-01' AND '2025-01-31'
GROUP BY dt;

-- Set operations: users active on both Jan 1 and Jan 2
SELECT BITMAP_AND_COUNT(a.active_users, b.active_users) AS retained
FROM dws_user_daily_bitmap a
JOIN dws_user_daily_bitmap b ON a.region = b.region
WHERE a.dt = '2025-01-01' AND b.dt = '2025-01-02'
  AND a.region = 'US';

BITMAP Functions Reference

Aggregation Spill

When Aggregation Spills to Disk

For large GROUP BY operations where the hash table cannot fit in BE memory, StarRocks 3.x supports spilling intermediate aggregation data to local disk. Without spill, large aggregations cause BE OOM and query failure.

Spill is triggered when aggregate hash table memory exceeds a threshold, writing sorted run files to the BE's spill directory (spill_storage_volume).

Spill Configuration Variables

Session-level (per query):

-- Enable automatic spill when memory threshold is exceeded
SET spill_mode = 'auto';        -- spill only when needed (default: 'none')
-- or
SET spill_mode = 'force';       -- always spill (for testing or guaranteed-OOM queries)

-- Memory threshold per operator before spill triggers (bytes)
-- Default: 2 GB
SET spill_operator_min_memory_bytes = 2147483648;

-- Size of each in-memory spill partition before flush to disk (bytes)
-- Default: 1 MB
SET spill_mem_table_size = 1048576;

BE configuration (be.conf):

# Directory for spill temporary files (should be on fast SSD)
spill_storage_volume = /data/starrocks/spill

# Max total disk space for spill across all queries (bytes)
# Default: unlimited (0)
spill_storage_limit = 107374182400   # 100 GB

Spill vs Memory Tuning Decision

Monitoring Spill in EXPLAIN ANALYZE

-- After running with spill, check query profile for spill metrics:
-- In the query profile (SHOW QUERY PROFILE '<query_id>'):
-- SpillBytes: bytes written to disk by spill
-- SpillWriteTime: time spent writing spill files
-- SpillReadTime: time spent reading spill files back
-- If SpillBytes > 0 and SpillWriteTime dominates → memory is the bottleneck

EXPLAIN Aggregation Analysis

Full EXPLAIN Walkthrough

EXPLAIN SELECT
    region,
    DATE(event_ts)            AS dt,
    COUNT(*)                  AS events,
    SUM(revenue)              AS total_revenue,
    COUNT(DISTINCT user_id)   AS dau
FROM ods.page_events
WHERE event_ts >= '2025-01-01'
GROUP BY region, DATE(event_ts);

Sample annotated output:

PLAN FRAGMENT 0                              ← coordinator fragment
  OUTPUT EXPRS: region | dt | events | total_revenue | dau
  PARTITION: UNPARTITIONED

  RESULT SINK

  8:AGGREGATE (merge finalize)               ← Phase 3: final COUNT DISTINCT merge
    group by: region, dt
    output: count(*), sum(revenue), count(distinct user_id)
  7:EXCHANGE                                 ← shuffle by (region, dt)

PLAN FRAGMENT 1                              ← intermediate fragment
  PARTITION: HASH_PARTITIONED: region, dt

  6:AGGREGATE (merge serialize)              ← Phase 2: partial distinct count
    group by: region, dt, user_id            ← user_id still in group for dedup
    output: count(*), sum(revenue)
  5:EXCHANGE                                 ← shuffle by (region, dt, user_id_bucket)

PLAN FRAGMENT 2                              ← BE scan fragment
  PARTITION: HASH_PARTITIONED: region, dt

  4:AGGREGATE (update serialize)             ← Phase 1: partial agg on each BE
    group by: region, dt, user_id
    output: count(*), sum(revenue)
  3:OlapScanNode
      TABLE: page_events
      PREAGGREGATION: OFF                    ← OFF because table is Duplicate Key
      partitions: 365/365 selected
      rollup: page_events
      tabletRatio: 32/32                     ← all 32 buckets scanned
      cardinality: 1200000000
      avgRowSize: 128.0
      numNodes: 8

Key EXPLAIN Signals for Aggregation

Rows-In vs Rows-Out Ratio

Check the aggregation efficiency via the ratio of input rows (scan cardinality) to output rows (final aggregate result):

-- Use EXPLAIN COSTS or query profile for row count estimates:
EXPLAIN COSTS SELECT region, SUM(revenue) FROM orders GROUP BY region;

-- If AggregationNode shows:
--   rows-in:  100 000 000
--   rows-out:     5 000
-- → 20 000:1 reduction — aggregation is highly effective

-- If rows-in ≈ rows-out → low selectivity — consider HLL/BITMAP or MV

Design Patterns

Decision Matrix: Where to Perform Aggregation

Pre-Aggregation in Aggregate Key Table Pattern

-- Tier 1: hourly pre-aggregated metrics
CREATE TABLE dws_events_hourly (
    event_hour      DATETIME       NOT NULL,
    region          VARCHAR(64)    NOT NULL,
    event_type      VARCHAR(64)    NOT NULL,
    event_count     BIGINT         SUM     DEFAULT '0',
    revenue_sum     DECIMAL(18,2)  SUM     DEFAULT '0.00',
    user_hll        HLL            HLL_UNION NOT NULL
)
AGGREGATE KEY(event_hour, region, event_type)
DISTRIBUTED BY HASH(region) BUCKETS 16
PARTITION BY RANGE(event_hour) (
    START ("2025-01-01") END ("2026-01-01") EVERY (INTERVAL 1 DAY)
)
PROPERTIES ("replication_num" = "3");

-- Tier 2: daily pre-aggregated metrics (aggregated from Tier 1)
CREATE TABLE dws_events_daily (
    event_date      DATE           NOT NULL,
    region          VARCHAR(64)    NOT NULL,
    event_count     BIGINT         SUM     DEFAULT '0',
    revenue_sum     DECIMAL(18,2)  SUM     DEFAULT '0.00',
    user_hll        HLL            HLL_UNION NOT NULL
)
AGGREGATE KEY(event_date, region)
DISTRIBUTED BY HASH(region) BUCKETS 16
PARTITION BY RANGE(event_date) (
    START ("2025-01-01") END ("2026-01-01") EVERY (INTERVAL 1 DAY)
)
PROPERTIES ("replication_num" = "3");

-- Load Tier 2 from Tier 1 using HLL_UNION:
INSERT INTO dws_events_daily
SELECT
    DATE(event_hour)       AS event_date,
    region,
    SUM(event_count)       AS event_count,
    SUM(revenue_sum)       AS revenue_sum,
    HLL_UNION(user_hll)    AS user_hll
FROM dws_events_hourly
WHERE event_hour >= CURRENT_DATE - INTERVAL 1 DAY
  AND event_hour <  CURRENT_DATE
GROUP BY DATE(event_hour), region;

Async Materialized View for Complex Aggregations

-- Async MV for weekly cohort retention — complex aggregation not suited for Aggregate Key
CREATE MATERIALIZED VIEW mv_weekly_retention
DISTRIBUTED BY HASH(cohort_week) BUCKETS 8
REFRESH ASYNC EVERY (INTERVAL 1 DAY)
AS
SELECT
    DATE_TRUNC('week', first_event_date)  AS cohort_week,
    DATE_TRUNC('week', event_date)        AS activity_week,
    COUNT(DISTINCT user_id)               AS retained_users
FROM (
    SELECT
        user_id,
        MIN(DATE(event_ts))  OVER (PARTITION BY user_id) AS first_event_date,
        DATE(event_ts)                                    AS event_date
    FROM ods.page_events
) cohort_data
GROUP BY
    DATE_TRUNC('week', first_event_date),
    DATE_TRUNC('week', event_date);

-- StarRocks CBO will transparently rewrite matching queries to use this MV

Anti-Patterns

1. COUNT DISTINCT on High-Cardinality Column Without Sketch

-- BAD: Forces full deduplication hash set in memory; OOM on billions of rows
SELECT dt, COUNT(DISTINCT user_id) AS dau
FROM ods.page_events
GROUP BY dt;

-- GOOD option A: Use approx_count_distinct (2% error, 100× faster)
SELECT dt, approx_count_distinct(user_id) AS dau_approx
FROM ods.page_events
GROUP BY dt;

-- GOOD option B: Use HLL column in Aggregate Key table (1% error, O(1) merge)
SELECT dt, HLL_CARDINALITY(HLL_UNION_AGG(user_hll)) AS dau_approx
FROM dws_events_daily
GROUP BY dt;

-- GOOD option C: Use BITMAP column for exact result with integer IDs
SELECT dt, BITMAP_UNION_COUNT(active_users) AS exact_dau
FROM dws_user_daily_bitmap
GROUP BY dt;

2. GROUP BY Without Statistics Leading to Bad Plan

-- Symptom: CBO chooses two-phase when one-phase would be faster
-- (or vice versa) because NDV estimates are wrong

-- Fix: run ANALYZE to collect column statistics
ANALYZE TABLE dws_events_hourly UPDATE HISTOGRAM ON region, event_type;
ANALYZE TABLE orders;

-- Check statistics exist:
SHOW STATS META WHERE `Table` = 'orders';

-- Check column histogram:
SHOW HISTOGRAM META WHERE `Table` = 'orders' AND `Column` = 'region';

3. Pre-Aggregation Disabled by WHERE Filter on Value Column

-- BAD: WHERE on non-key column disables preagg on Aggregate Key table
SELECT event_date, SUM(revenue)
FROM dws_daily_revenue           -- Aggregate Key table
WHERE revenue > 1000             -- revenue is a value column → preagg OFF
GROUP BY event_date;

-- FIX A: Move filter to key columns
SELECT event_date, SUM(revenue)
FROM dws_daily_revenue
WHERE event_date >= '2025-01-01' -- event_date is a key column → preagg ON
GROUP BY event_date;

-- FIX B: If filtering on value column is required, use Duplicate Key + query-time agg
--        or filter after aggregation:
SELECT event_date, total_revenue
FROM (
    SELECT event_date, SUM(revenue) AS total_revenue
    FROM dws_daily_revenue
    GROUP BY event_date
) t
WHERE total_revenue > 1000;
-- PREAGGREGATION: ON in inner query

4. Mismatched Aggregate Function Disabling Pre-Aggregation

-- BAD: Table stores SUM(revenue) but query computes AVG — preagg can't help
CREATE TABLE dws_daily_avg_wrong (
    dt       DATE           NOT NULL,
    user_id  BIGINT         NOT NULL,
    revenue  DECIMAL(18,2)  SUM DEFAULT '0.00'   -- stored as SUM
)
AGGREGATE KEY(dt, user_id);

SELECT dt, AVG(revenue) FROM dws_daily_avg_wrong GROUP BY dt;
-- PREAGGREGATION: OFF — engine must read raw rows to compute AVG

-- GOOD: Store both SUM and COUNT; compute AVG at query time
CREATE TABLE dws_daily_avg_correct (
    dt            DATE           NOT NULL,
    user_id       BIGINT         NOT NULL,
    revenue_sum   DECIMAL(18,2)  SUM DEFAULT '0.00',
    revenue_cnt   BIGINT         SUM DEFAULT '0'
)
AGGREGATE KEY(dt, user_id);

SELECT dt, SUM(revenue_sum) / NULLIF(SUM(revenue_cnt), 0) AS avg_revenue
FROM dws_daily_avg_correct
GROUP BY dt;
-- PREAGGREGATION: ON

5. Large Spill Without Root Cause Investigation

-- BAD: Enabling spill as a permanent fix without understanding root cause
SET spill_mode = 'force';   -- on by default for all queries

-- GOOD: Profile first, then choose the right solution
-- Step 1: Check if GROUP BY key has high or low cardinality
-- Step 2: If low cardinality → one-phase should work; check if stats are stale
-- Step 3: If high cardinality → consider HLL/BITMAP or MV rollup
-- Step 4: If result set is large and memory is genuinely insufficient → spill is correct

-- Only set spill at session level for specific queries:
SET spill_mode = 'auto';
SELECT region, user_segment, COUNT(*), SUM(revenue)
FROM fact_orders_large
GROUP BY region, user_segment;
SET spill_mode = 'none';   -- restore default after

6. Using BITMAP on Non-Integer or Negative IDs

-- BAD: to_bitmap() only accepts non-negative BIGINT
SELECT to_bitmap(user_id) FROM events WHERE user_id < 0;   -- returns NULL or error
SELECT to_bitmap(session_uuid) FROM events;                 -- won't work on VARCHAR

-- GOOD: Map to integer ID first
-- For UUID sessions: maintain a session_id → integer_id mapping table
-- For negative IDs: shift by offset → to_bitmap(user_id + 2147483648)
-- Or use HLL (supports any type):
SELECT hll_hash(session_uuid) FROM events;

References

• StarRocks documentation — Query acceleration: aggregation optimization
• StarRocks documentation — Aggregate Key table: pre-aggregation and compaction behavior
• StarRocks documentation — HLL functions and approximate COUNT DISTINCT
• StarRocks documentation — BITMAP functions and exact COUNT DISTINCT
• StarRocks documentation — Query spill configuration (spill_mode, spill_operator_min_memory_bytes)
• StarRocks documentation — EXPLAIN output interpretation: fragment plans, AggregationNode labels
• StarRocks documentation — Session variables: new_planner_agg_stage, enable_distinct_column_bucketization, count_distinct_column_buckets
• StarRocks documentation — Async Materialized Views for transparent query rewrite
• Related skills: starrocks-ddl-table-types (Aggregate Key table design), starrocks-explain-plan (EXPLAIN reading), starrocks-materialized-views (async MV setup), starrocks-bucketing (distribution key selection for local aggregation)