@Rocky Hi, We are running StarRocks with 3 BE nodes. All tables are distributed using HASH(cm_id) with 32 buckets and grouped under a colocate join on cm_id (customer_id). The problem is our customers have highly uneven data. Some whale customers have 50-80 million rows while small customers have only a few thousand rows. This causes: - Tablets distributed evenly across BEs - Storage distributed severely unevenly BE-0 = 61GB | BE-1 = 15GB | BE-2 = 3.5GB - One single tablet = 38GB (one whale customer) - BE-0 RAM hitting 100% while others sit at 30-35% - Routine Load throwing OOM errors on BE-0 Root cause: HASH(cm_id) maps each customer to exactly one bucket on one BE permanently. No matter how many buckets or BE nodes we add, a whale customer's entire data stays on one BE — this is the colocate contract. Changing to HASH(cm_id, id) fixes storage balance but breaks our colocate join which we depend on for all our queries (always JOIN ON cm_id). We are stuck between: Colocate join on cm_id → fast joins, skewed storage HASH(cm_id, id) → even storage, no colocate Is there any StarRocks-native solution to achieve even storage distribution while preserving colocate join on cm_id for skewed customer datasets?

The situation you've described is a classic "Multi-tenant Skew" problem. In StarRocks,

DISTRIBUTED BY HASH(cm_id)

creates a rigid mapping where every row for a specific

cm_id

is assigned the same hash value and, therefore, must reside in the exact same bucket (tablet) on one BE. This is the physical requirement for a Colocate Join, but it effectively turns your largest whale customers into "single-node" databases. To resolve this while keeping joins fast, you have two StarRocks-native paths: Solution 1: The "Hybrid" View (Recommended for Performance) This approach allows you to keep Colocate Join for 99% of your customers (small/medium) while splitting the whales across the cluster to balance storage and RAM. 1. Split the data into two tables: * `table_main`: Distributed by

HASH(cm_id)

and part of the

colocate_with

group. Contains all standard customers. * `table_whales`: Distributed by

HASH(cm_id, id)

(or

RANDOM

in v3.1+). This splits the whale's data across all 3 BEs. 2. Create a Unified View:

sql
    CREATE VIEW customer_data AS
    SELECT * FROM table_main
    UNION ALL
    SELECT * FROM table_whales;

3. Why this works: * Small Customers: Joins on

cm_id

will prune to

table_main

. StarRocks will use the Colocate Join strategy, keeping queries sub-second. * Whale Customers: Joins will prune to

table_whales

. Because the data is balanced across all nodes, StarRocks will use a Shuffle Join or Bucket Shuffle Join. While not a "local join," the parallel compute power of all 3 BEs working on the whale's data is usually faster than one BE struggling with a 38GB tablet. ────────── Solution 2: Skew Join V2 (Recommended for Scalability) If you prefer a single table, you must change the distribution key to balance storage, and then use StarRocks' Skew Join V2 (available in v3.x) to handle the join performance. 1. Rebalance Storage: Modify the table to use a composite hash key:

DISTRIBUTED BY HASH(cm_id, some_id)

. This will immediately solve your storage skew (the 38GB tablet will be split into smaller pieces across all BEs) and stop the Routine Load OOMs. 2. Optimize the Join: Since a join on

cm_id

is no longer "Colocated," StarRocks would normally perform a standard Shuffle Join. To optimize this for whales, use the Skew Join V2 hint:

sql
    SELECT /_+ SKEW_JOIN(t1, cm_id(1001)) _/ *
    FROM table1 t1
    JOIN table2 t2 ON t1.cm_id = t2.cm_id;

* Hint Syntax: You explicitly tell the optimizer that

cm_id = 1001

is a whale. * Native Logic: StarRocks will use a Broadcast Join specifically for that value (broadcasting the small side of the join to all nodes) and a Shuffle Join for everything else. This prevents the single-BE bottleneck during query execution. ────────── Why Routine Load is OOMing The OOM on BE-0 is occurring because when you load data for the whale customer, the Coordinator BE must send all that data to the one BE that owns that specific bucket. That BE then has to process a massive amount of data in memory (indexing, sorting, and building rowsets) for a single tablet. * Fix: By changing to

HASH(cm_id, id)

, the load task is distributed. Instead of one BE receiving 100% of the whale's load, each BE receives ~33%, staying well within RAM limits. Summary of Actions * Immediate fix for OOM: Alter the table to

DISTRIBUTED BY HASH(cm_id, id) BUCKETS 48

(roughly 16 buckets per BE). * If you must have Colocate Join for small tenants: Use the Solution 1 (View) approach. * If you prioritize schema simplicity: Use Solution 2 (Skew Join V2) and enable it globally with

SET enable_optimize_skew_join_v2 = true;

. Note: Avoid increasing the bucket count (e.g., to 128) while keeping

HASH(cm_id)

. It will not help; a single

cm_id

will still map to exactly one bucket regardless of how many buckets exist. References