@Rocky in a shared data model, we are seeing fairly high network traffic between compute nodes; is there any tuning that can be done to reduce the network traffic?

In a StarRocks Shared-data architecture (Compute Nodes + Remote Storage), high network traffic between compute nodes is typically caused by data shuffling during joins or aggregations, or by the Data Cache Sharing feature. Here are the primary tuning strategies to reduce inter-node network traffic: 1. Implement Colocate Joins The most effective way to eliminate network traffic during joins is to ensure that the data being joined resides on the same compute node. * How it works: By using the

colocate_with

property during table creation, you can force related tables to distribute their buckets to the same set of nodes based on the same distribution key. * Tuning:

sql
    -- Example for two tables
    CREATE TABLE table1 (...) DISTRIBUTED BY HASH(id) BUCKETS 32
    PROPERTIES("colocate_with" = "group_join_1");

    CREATE TABLE table2 (...) DISTRIBUTED BY HASH(id) BUCKETS 32
    PROPERTIES("colocate_with" = "group_join_1");

* Benefit: Joins on the distribution key (

id

) will be performed locally on each CN, completely avoiding the network shuffle. 2. Optimize Join Strategies (Broadcast vs. Shuffle) If Colocate Join isn't possible, StarRocks chooses between Broadcast and Shuffle joins. * Broadcast Join: Sends the entire small table to every node. If one table is very small, this is often cheaper than shuffling a huge table. * Shuffle Join: Redistributes both tables across the network based on the join key. * Tuning: Check your

EXPLAIN

plan. If you see a massive shuffle that shouldn't be there, you can hint the optimizer:

sql
    SELECT * FROM large_table JOIN [broadcast] small_table ON ...

3. Review Data Cache Sharing In shared-data mode, StarRocks has a feature called Cache Sharing. If a node needs data that isn't in its local cache, it can fetch it from another node's cache via the network instead of going to S3/Object Storage. * The Trade-off: This improves query latency but increases inter-node network traffic. * Tuning: If your internal network is a bottleneck and you'd prefer nodes to pull directly from remote storage (which uses external bandwidth), you can disable it:

sql
    SET enable_datacache_sharing = false;

Note: This might increase your S3/OSS latency and costs. 4. Leverage Runtime Filters Runtime filters are generated during the join process to prune data at the scan stage, significantly reducing the amount of data that needs to be sent over the network. * Tuning: Ensure they are enabled (default is usually fine) and consider increasing the filter type to

global

if you have complex multi-stage joins.

sql
    SET runtime_filter_type = 'GLOBAL';

5. Proper Bucketing (Distribution) If your distribution keys don't match your frequent join or aggregation keys, StarRocks must shuffle the data every time a query runs. * Tuning: Align your

DISTRIBUTED BY HASH(column)

with the columns you most frequently use in

JOIN

conditions or

GROUP BY

clauses. This allows the execution engine to perform "Shuffle-less" or "Local" operations more often. 6. Network Compression StarRocks uses LZ4 by default to compress data sent between nodes (via BRPC). While usually optimized, you can verify if compression is active. * BE/CN Configuration: Check

brpc_common_request_compression

in your CN configuration file. It is typically set to

snappy

or

lz4

. 7. Global Dictionary for Low-Cardinality Strings If you are shuffling large volumes of string data, using a Global Dictionary allows StarRocks to shuffle integer IDs instead of raw strings. * Benefit: This can reduce the bytes transferred over the network by several multiples for low-cardinality columns. Summary Recommendation: