Alibabacloud Emr Starrocks Assistant

Alibaba Cloud EMR Serverless StarRocks Development & Operations Assistant

Help users perform day-to-day table design, data ingestion, SQL writing & tuning, and health diagnostics on Alibaba Cloud EMR Serverless StarRocks. All cluster access goes through the bundled srsql CLI (pymysql-based, uses the user's own account); no MySQL client required. Non-READ SQL is classified by sqlglot and requires --yes confirmation before execution.

Scope statement: This Skill focuses on using StarRocks — development, diagnostics, and day-to-day data operations. Cluster-internal data and schema operations (DDL, DML, materialized view refresh, GRANT, etc.) are supported and execute under the user's own account, gated by sqlglot classification + --yes confirmation. Instance-lifecycle operations (create, scale, restart, configuration change, version upgrade) are control-plane operations and are not in this Skill's scope; please use the EMR Serverless console or the corresponding OpenAPI.

When to use / When not to use

When to use:

• Schema design (table model, partitioning, bucketing, sort key, indexes, storage parameters)
• Data ingestion selection (Stream/Broker/Routine Load, INSERT, Pipe, Flink/Kafka Connector, CDC)
• SQL writing, rewriting, and tuning (JOIN strategy, window functions, CTE, aggregation optimization, statistics)
• Materialized view design and operations
• Cluster health diagnostics (FE/BE/CN nodes, tablet health, compaction, warehouse, recent failed loads)

When NOT to use:

• Instance lifecycle control: create / scale / restart / config change / upgrade StarRocks instances — these are control-plane operations; use the EMR Serverless console or the corresponding OpenAPI instead
• Operating non-StarRocks products: EMR Cluster, Spark, Milvus, ClickHouse, Doris, RDS, ECS, etc.

First-time setup: install & log in

This Skill ships with the sr-connect Python CLI. See references/connect.md for details.

Assistant bootstrap protocol (instructions for Claude)

When this Skill is invoked and you anticipate running any cluster query, ensure srsql is available before asking the user for anything:

1. Run which srsql. If it returns a path, skip to step 4. 2. If missing, install it yourself: uv tool install <skill-project-root> where <skill-project-root> is the directory containing this SKILL.md and pyproject.toml (the Skill's base directory shown at invocation time; commonly ~/.claude/skills/alibabacloud-emr-starrocks-assistant/, which may be a symlink). Do not ask the user to run this — the bundled CLI is part of the Skill's capability surface, not user infrastructure. 3. If uv itself is missing (which uv fails), surface that to the user — uv is a system tool and not auto-installed. 4. Check ~/.starrocks/{profile}.cnf (default profile name: default; respect SR_PROFILE env var if set). If it exists, skip to step 5. If missing:

• First try `sr-login --from-env`. Safe to call unconditionally — it exits 2 with a clear "missing" message when the environment doesn't have the credentials it needs, and does nothing else. You do not need to inspect environment variables yourself.
• If `sr-login --from-env` exits non-zero, the user hasn't logged in yet. Give them the sr-login --host ... --user ... command and ask them to run it themselves. Do not run interactive `sr-login` yourself — it would block on a password prompt you cannot answer.

5. After both srsql is on PATH and the profile file exists, run queries via srsql -e "..." yourself.

If srsql was just installed in this session and PATH hasn't been refreshed in the user's shell, fall back to the absolute path printed by uv tool install (typically ~/.local/bin/srsql).

Chat-style rule after bootstrap succeeds: Do not echo sr-whoami / srsql -e "..." invocation syntax to the user as a "you can now run …" hint. You are the one calling these CLIs on the user's behalf — the user drives the Skill, not the binaries. Skip the post-success "next step" narration entirely and just ask what they want to do, or proceed if their intent is already clear.

Login command (give this to the user when their profile is missing)

# EMR Serverless StarRocks — both internal and public endpoints use the MySQL
# wire protocol over plain TCP; no SSL/TLS. Use the same form for either.
sr-login --host <fe-endpoint> --port 9030 --user <account>

# Verify
sr-whoami
srsql -e "SELECT CURRENT_VERSION()"

Re-running sr-login with the same --profile silently overwrites the stored credential (same semantics as docker login). Use --profile for multi-cluster:

sr-login --profile prod --host fe-prod.xxx --user app_user
SR_PROFILE=prod srsql -e "..."

Security model

This Skill has two layers:

1. FE is the authoritative permission boundary. The user supplies their own StarRocks account; whatever they're allowed to do, they're allowed to do. The Skill does not create, elevate, or rotate any accounts. 2. `srsql` is a UX gate, not a security boundary. Every statement is parsed by sqlglot (dialect starrocks):

• READ (SELECT / SHOW / DESC / EXPLAIN / WITH / …) executes directly.
• Any non-READ (INSERT / UPDATE / DELETE / DDL / GRANT / SET / USE / …) is refused unless `--yes` is passed.
• SQL sqlglot cannot parse falls back to a leading-keyword check; if still ambiguous → UNKNOWN, treated as non-READ, executable with --yes plus a soft warning.

When the user asks for a write operation:

1. Show them the SQL you intend to run. 2. Optionally preview classification via srsql --dry-run -e "...". 3. Get explicit confirmation in chat. 4. Then run with srsql --yes -e "...".

For DDL on production tables, or operations that change global cluster state (CREATE/DROP USER, ADMIN SET CONFIG, etc.), prefer to print the SQL and let the user run it themselves — even though the gate would let them run it via --yes. The gate is a safety net, not a license.

Input validation & command-injection protection

SQL passed into srsql -e "..." is assembled by the LLM and must follow these rules:

1. Identifiers (table / column / database names) are validated before interpolation: only [A-Za-z0-9_] plus backtick-quoted forms. 2. User-provided string values (search terms, label names, etc.) are not spliced into SQL directly; use parameter binding or pre-escape. 3. Never execute raw user-provided strings as SQL fragments.

Sensitive data masking

Intent routing

Disambiguation rule: When the user input is ambiguous (e.g. "ingestion is slow", "queries are slow") and context is unclear, ask one clarifying question before acting.

Five scenarios at a glance

1. Schema design

Four table models and their typical use cases:

⚠ Anti-patterns — do not produce these in DDL:

• Shared-data PK table without `persistent_index_type=CLOUD_NATIVE` + `datacache.partition_duration` — LOCAL index doesn't survive CN rebalance; no hot-data caching window. See schema/storage-properties.md.
• Setting `datacache.partition_duration` to an arbitrary "hot window" (e.g. `30 DAY`) instead of the user's stated query window — the value MUST be ≥ the query window. If the user says "查询近 N 天" / "queries the last N days", set datacache.partition_duration = "N DAY" (or larger). A value smaller than the query window guarantees cache misses on in-window queries. Do not default to 7/30/60 days when the user has given you a number.
• `storage_cooldown_time`/`storage_cooldown_ttl`/`storage_medium`/`replicated_storage` on shared-data — silently stripped or rejected by PropertyAnalyzer; use datacache.partition_duration for the cooldown effect.
• FLOAT / DOUBLE columns inside `PRIMARY KEY` — not supported; use BIGINT or DECIMAL.
• Treating "CN" as a shared-nothing signal — CN = Compute Node, which is the shared-data terminology. BE = Backend = shared-nothing.

See references/schema.md.

2. Data ingestion

⚠ Anti-patterns — do not produce these in load configs:

• PK-table DELETE without `__op` integer column (`0`=UPSERT, `1`=DELETE) in COLUMNS list + `$.__op` in `jsonpaths` — all events are silently treated as UPSERT. The `__op` contract is a pair and must be taught as a pair: the literal column name is __op, and the integer values are __op=0 for UPSERT and __op=1 for DELETE. Even when the user only asks about DELETE, your response MUST state both mappings (__op=0 → UPSERT, __op=1 → DELETE) — never one without the other. This applies on every ingestion path including Flink Connector and Kafka Connector, where the connector populates __op for the user but they still need both values to debug "DELETE not applied" / "UPSERT not applied" symptoms.
• Treating `partial_update=true` as a DELETE enabler — it controls partial-column UPSERT and has nothing to do with DELETE. If a user enables it while asking why DELETE doesn't work, flag it as misconfigured-for-intent and tell them to remove it unless they actually have a partial-column UPSERT use case. Do not validate the existing setting just because it parses.
• `COLUMNS FROM PATH AS (...)` in Routine Load — that's Broker Load's Hive-partition path syntax; not valid in Routine Load.
• `__op` values as strings (`"upsert"`/`"delete"`) — must be the integers 0 / 1.
• High-throughput CDC (≥ ~10K events/sec) without flagging TOO_MANY_VERSION risk — applies to Routine Load, Flink Connector, Kafka Connector, not just INSERT INTO VALUES. Whenever the user's scenario implies high event rate, the recommendation MUST cover: (a) the method-appropriate concurrency cap (desired_concurrent_number ≤ Kafka partitions for Routine Load; sink.parallelism ≤ Kafka partitions for Flink/Kafka Connector), AND (b) an explicit TOO_MANY_VERSION / compaction-pressure warning with the relevant flush-interval guidance.

See references/data-import.md.

3. SQL development

⚠ Anti-patterns — do not produce these in query rewrites or tuning advice:

• Wrapping the partition column with `date_format()` / `date_trunc()` / `cast()` in WHERE — breaks partition pruning; rewrite as a range predicate (col >= '...' AND col < '...').
• Tuning advice without `EXPLAIN VERBOSE` + checking `partitions=N/M` and `tabletRatio=N/M` — pruning failures (numerator == denominator) go undetected; never use plain EXPLAIN for this.
• Reading `cardinality` in EXPLAIN as the result row count — it's the CBO's row estimate. Always quantify the staleness gap using the direct comparison `cardinality` vs the user-stated total table size (e.g. "estimate 5M vs total 500M ≈ 100×"); a ratio > 10× means stats are stale → run ANALYZE TABLE.
• Estimating "real filtered rows" by guessing predicate selectivity, then comparing cardinality to that guess — you don't have runtime row counts, and guessing selectivity from a predicate like WHERE create_time > '...' introduces large errors (you don't know the data distribution). When the user gives you a total row count, compare cardinality to that directly; do not divide the total by an assumed time window or selectivity factor.
• Conflating `partitions`/`tabletRatio` pruning failures with `cardinality` deviation — these are two independent diagnostic signals. When both look bad in the same OlapScanNode (e.g. partitions=N/N AND cardinality off from total table size by 10×–100×), report them as separate findings with separate fixes (predicate/type fix vs ANALYZE TABLE). Do not use cardinality deviation to "explain" pruning failure, and do not let pruning failure absorb the stale-stats finding.
• Recommending BE/CN scale-out before plan/stats analysis — SQL/stats fixes precede capacity changes.

See references/sql.md.

4. Cluster diagnostics

Diagnostic order:

1. Identify architecture (shared-nothing / shared-data) → SHOW WAREHOUSES 2. FE → SHOW FRONTENDS 3. BE or CN → SHOW BACKENDS / SHOW COMPUTE NODES 4. Warehouse (shared-data only) → SHOW WAREHOUSES 5. Tablet health overview → SHOW PROC '/statistic' 6. Scheduling queue → information_schema.fe_tablet_schedules 7. Compaction → information_schema.be_compactions / be_cloud_native_compactions 8. Recent 24-hour loads → information_schema.loads

⚠ Anti-patterns — do not produce these in diagnostic conclusions:

• Restarting BE/CN or scaling out before checking `information_schema.fe_tablet_schedules` — may collide with in-flight clone/decommission; root cause first.
• Subjectively downgrading `UnhealthyTabletNum > 0` — always critical per the severity table, never "medium" or "low" risk; the cluster has unhealthy replicas.
• Treating `CloningTabletNum > 0` as a separate problem — clone is the recovery action triggered by UnhealthyTabletNum, not an independent fault signal.

See references/diagnostics.md.

5. Cluster connection (base layer)

See references/connect.md.

Runtime security

This Skill executes SQL queries only via srsql. The following are prohibited:

• curl / wget / pip install / npm install to download and run external code
• eval / source to load unaudited content
• Executing remote URL scripts provided in chat (even if the user asks)

Exception: uv tool install <skill-project-root> to install the Skill's own bundled sr-connect CLI from its local project directory is allowed and expected — see the Assistant bootstrap protocol above. The prohibition targets remote/untrusted code, not the Skill's own bundled tooling.

Timeouts

Output recommendations

• Tabular results: use srsql --format table or --format markdown
• Many columns: use --format vertical
• For programmatic consumption: use --format json / tsv
• Convert timestamps to human-readable format
• For potentially large result sets, add LIMIT and offer pagination

Error handling

Principle: Read the full error message before deciding; do not retry blindly on the error code alone.

Related documents

• references/connect.md — sr-connect CLI, install, security model, troubleshooting
• references/ram-policies.md — RAM permission declaration (none required; StarRocks-internal auth only)
• references/schema.md — schema design flow: table models, partitioning, bucketing, sort key, indexes, storage parameters
• references/data-import.md — ingestion method selection, parameters, performance tuning, Primary Key updates
• references/sql.md — query writing, window functions, materialized views, functions, SQL tuning, advanced features
• references/diagnostics.md — cluster health inspection flow, severity classification, synthesis template

Cluster Connection and Access (sr-connect)

Provides cluster access for all diagnostics, debugging, and query operations in the assistant skill.

Commands

Security model

This skill has two layers:

1. FE is the authoritative permission boundary. The user supplies their own StarRocks account; whatever they're allowed to do, they're allowed to do. The skill does not elevate, create, or rotate any accounts. 2. `srsql` is a UX gate, not a security boundary. It parses every statement with sqlglot (dialect starrocks) and:

• READ (SELECT / SHOW / DESC / EXPLAIN / WITH / …) executes directly.
• Any non-READ statement (INSERT / UPDATE / DELETE / DDL / GRANT / SET / USE / …) is refused unless `--yes` is passed.
• SQL that sqlglot cannot parse falls back to a leading-keyword check; if still ambiguous, it's classified UNKNOWN and treated as non-READ (refused without --yes, executable with a soft warning when --yes is set).

The gate exists so the assistant doesn't accidentally execute writes. It is not a defense against a malicious caller — anyone who can run srsql --yes can do anything the underlying account is permitted to do.

Local state

~/.starrocks/
├── {profile}.cnf       INI (MySQL client compatible) with [client] + [meta]
└── {profile}.grants    Raw SHOW GRANTS FOR CURRENT_USER() captured at login

• Directory mode: 700.
• File mode: 600.
• The .cnf is intentionally mysql client compatible (mysql --defaults-extra-file=~/.starrocks/default.cnf works), but the skill itself uses srsql, not the MySQL CLI.

First-time login

# Install uv (if not already installed)
curl -LsSf https://astral.sh/uv/install.sh | sh

# Install the CLI from the skill's project root
uv tool install .

# Log in (password is prompted interactively; not echoed, not logged).
# EMR Serverless internal and public endpoints both use plain MySQL wire
# protocol on port 9030 — no SSL/TLS. The same command works for either.
sr-login --host <fe-endpoint> --port 9030 --user <your-account>

# Verify
sr-whoami
srsql -e "SELECT CURRENT_VERSION()"

sr-login overwrites any existing profile of the same name silently — same semantics as docker login / gh auth login. To switch the stored password, just log in again.

Non-interactive login (--from-env)

For sandbox / CI environments where the platform pre-provisions credentials, call:

sr-login --from-env

The CLI picks the required values up from the environment on its own. It exits with code 2 and a clear "missing" message when those values aren't present, so it's safe to call unconditionally from a setup script or from the assistant's bootstrap path — no need to inspect or echo any variable names. --profile is still honored if multi-cluster setup is needed.

Multi-cluster

sr-login --profile prod    --host fe-prod.xxx    --user app_user
sr-login --profile staging --host fe-staging.xxx --user app_user

SR_PROFILE=prod    srsql -e "..."
SR_PROFILE=staging srsql -e "..."

Querying

# READ — runs directly
srsql -e "SHOW FRONTENDS"
srsql -e "SELECT * FROM information_schema.backends" --format json

# Non-READ — preview first with --dry-run, then re-run with --yes
srsql --dry-run -e "INSERT INTO t VALUES (1, 2)"
srsql --yes     -e "INSERT INTO t VALUES (1, 2)"

# Multi-statement: any non-READ in the batch requires --yes for the whole batch
srsql --yes -e "USE db1; INSERT INTO t SELECT * FROM s"

Supported --format values: tsv (default, LLM-friendly), json, table, markdown, vertical.

File overview

• pyproject.toml — declares 4 entry points: sr-login, sr-logout, sr-whoami, srsql
• scripts/sr_connect/login.py — sr-login flow
• scripts/sr_connect/logout.py — sr-logout flow
• scripts/sr_connect/whoami.py — sr-whoami flow
• scripts/sr_connect/query.py — srsql flow + multi-format output + classification gate
• scripts/sr_connect/classify.py — sqlglot-based SQL classifier (READ vs non-READ)
• scripts/sr_connect/connection.py — pymysql connection wrapper
• scripts/sr_connect/config.py — ~/.starrocks/{profile}.cnf + .grants read/write
• scripts/sr_connect/cli.py — click entry points

Troubleshooting

Connection troubleshooting (sr-doctor)

EMR Serverless StarRocks exposes the FE endpoint under two DNS suffixes:

sr-doctor (sr-doctor --host <host> or SR_HOST=... sr-doctor) classifies the host, TCP-probes the port, and prints what to do next. sr-login runs it automatically when its own connection attempt fails — no manual step needed in that path.

Behaviour matrix

Sample outputs

VPC endpoint unreachable:

[!] Cannot reach fe-c-xxx-internal.starrocks.aliyuncs.com:9030

Endpoint type: VPC (-internal.starrocks.aliyuncs.com)
Cause: Your client is not inside the cluster's VPC.

Fix: Switch to the public endpoint by removing "-internal" from the host:

    export SR_HOST=fe-c-xxx.starrocks.aliyuncs.com
    sr-login --from-env

If the instance has not enabled the public endpoint yet:
    Console -> EMR Serverless StarRocks -> instance details
    -> Gateway info -> Provision SLB  (prerequisite)
    -> Public address -> Enable public endpoint
    (creates a billable CLB; provisioning takes a few minutes)
    Docs: https://help.aliyun.com/zh/emr/emr-serverless-starrocks/manage-gateways

Public endpoint unreachable:

[!] Cannot reach fe-c-xxx.starrocks.aliyuncs.com:9030

Endpoint type: Public (.starrocks.aliyuncs.com)
Your public egress IP: 47.92.100.50
Suggested whitelist CIDR: 47.92.100.0/24
    (/24 buffers for NAT IP drift within a cluster)

Fix: Add the CIDR to the cluster whitelist:
    Console -> EMR Serverless -> instance details
    -> Network -> Whitelist -> Add CIDR -> Save

Then retry: sr-login --from-env

Implementation notes

• sr-doctor uses standard-library socket.create_connection for the TCP probe (5 s timeout) and urllib.request for the egress IP lookup. No extra dependencies.
• Egress IP discovery uses https://ipinfo.io/ip (plain-text response). api.ipify.org and the Aliyun ECS metadata service (100.100.100.200) were observed to be unreliable in agenthub-style sandboxes — ipinfo.io is the only one that worked there.
• All network calls are encapsulated inside the CLI; the assistant never invokes curl itself, preserving the Runtime-security boundary set out in SKILL.md.

Password input guidelines

• Enter the password interactively when prompted by sr-login. The prompt is hidden, not echoed, and doesn't appear in ps or shell history.
• Do not pass --password on the command line — it leaks via ps and shell history.

StarRocks Data Ingestion Best Practices

Required Information Checklist

Before recommending an ingestion approach, gather the following information. If any item is missing, proactively ask:

Ingestion Method Selection Decision Flow

Step 1: Identify the data source
    ├─ Kafka / Pulsar → Step 2a
    ├─ Local file → Step 2b
    ├─ HDFS / S3 / OSS / Azure / GCS → Step 2c
    ├─ MySQL / PostgreSQL (CDC) → Step 2d
    └─ StarRocks internal table / external table → Step 2e

Step 2a: Kafka/Pulsar streaming data
    ├─ Native SQL management, simple scenarios → Routine Load (see data-import/routine-load.md)
    └─ Existing Flink/Kafka Connect pipeline → Flink Connector / Kafka Connector (see data-import/connectors.md)

Step 2b: Local file
    ├─ Single file < 10GB → Stream Load (see data-import/stream-load.md)
    └─ Single file > 10GB → Split and Stream Load, or upload to object storage and use Broker Load

Step 2c: HDFS / object storage
    ├─ < tens of GB + one-time ingestion → Broker Load (see data-import/broker-load.md)
    ├─ Tens of GB ~ TB scale → Broker Load or INSERT INTO SELECT FROM FILES()
    ├─ Need to continuously watch for new files → Pipe + AUTO_INGEST (see data-import/insert-and-pipe.md)
    └─ TB-scale first-time bulk migration (Hive) → Spark Load

Step 2d: CDC real-time sync
    ├─ Flink CDC → Flink Connector (recommended, see data-import/connectors.md)
    ├─ Debezium + Kafka → Routine Load or Kafka Connector
    └─ Canal / DataX / CloudCanal → Corresponding tool (see data-import/connectors.md)

Step 2e: Internal/external table
    └─ INSERT INTO target_table SELECT ... FROM source

Detailed configuration and examples for each step are in the corresponding reference file. Read on demand.

Ingestion Method Quick Reference

Format Selection Guide

Common Anti-Patterns

Proactively check for and avoid the following when designing an ingestion plan:

Reference File Index

When you need detailed configuration and examples for a specific ingestion method, read the corresponding file:

Output Template

When providing an ingestion recommendation, use the following structured format:

## Ingestion Plan

### 1. Recommended ingestion method: {Stream Load / Broker Load / Routine Load / ...}
**Rationale:** {Why this method fits the user's data source, volume, and latency requirement}

### 2. Data format and preprocessing
**Source format:** {CSV / JSON / Parquet / ...}
**Preprocessing:** {Whether format conversion, field mapping, or data cleaning is required}

### 3. Key parameter configuration
| Parameter | Recommended value | Description |
|------|--------|------|
| ... | ... | ... |

### 4. Sample ingestion statement

​```sql
-- Complete ingestion statement or command
​```

### 5. Performance estimate
- **Estimated throughput:** {X MB/s or X 10K rows/sec}
- **Resource consumption:** {Memory / CPU / network}
- **Caveats:** {Version requirements, known limitations, compaction impact}

### 6. Monitoring and operations
- **Task state:** {SHOW LOAD / SHOW ROUTINE LOAD / SHOW PIPES}
- **Key metrics:** {Monitoring items to watch}
- **Error handling:** {Common issues and how to handle them}

Scope Note

"Slow ingestion" splits into two cases — designing a new ingestion pipeline requires performance design (this topic) vs. a previously healthy ingestion suddenly slowing down requires troubleshooting cluster-level issues (-> diagnostics.md).

Broker Load Detailed Guide

Table of Contents

1. Overview 2. SQL Syntax 3. Storage System Configuration 4. Format Parameters 5. Data Transformation 6. Best Practices 7. Monitoring and Management 8. Common Issues

---

Overview

Broker Load ingests data in batch from HDFS or cloud storage asynchronously. After submission it runs in the background; check progress via SHOW LOAD.

Key characteristics:

• Asynchronous execution; suitable for large-scale offline ingestion
• Supports CSV, Parquet, and ORC formats (JSON is not supported — use INSERT INTO ... FROM FILES() instead for JSON in object storage)
• Supports HDFS, S3, OSS, Azure Blob, GCS, MinIO, and other storage backends
• Supports wildcards to match multiple files
• Supports multi-table transactional ingestion
• Supports UPSERT / DELETE on Primary Key tables
• Supports both broker mode (HA/Kerberos) and broker-free mode

---

SQL Syntax

Basic syntax:

LOAD LABEL <database>.<label>
(
    DATA INFILE("<file_path>")
    [NEGATIVE]
    INTO TABLE <table_name>
    [PARTITION (<partition_list>)]
    [COLUMNS TERMINATED BY "<separator>"]
    [ROWS TERMINATED BY "<row_separator>"]
    [FORMAT AS "CSV|Parquet|ORC"]
    [(column_list)]
    [COLUMNS FROM PATH AS (partition_columns)]
    [SET (column_mapping)]
    [WHERE <predicate>]
)
[WITH BROKER "<broker_name>" (<broker_properties>)]
PROPERTIES (<load_properties>);

Full example — ingest CSV from OSS:

LOAD LABEL mydb.oss_load_20240101
(
    DATA INFILE("oss://my-bucket/data/2024/01/01/*.csv")
    INTO TABLE user_behavior
    COLUMNS TERMINATED BY ","
    (user_id, item_id, behavior_type, ts_str)
    SET (event_time = str_to_date(ts_str, '%Y-%m-%d %H:%i:%s'))
    WHERE behavior_type IN ('buy', 'cart')
)
WITH BROKER
(
    "fs.oss.accessKeyId" = "<ak>",
    "fs.oss.accessKeySecret" = "<sk>",
    "fs.oss.endpoint" = "oss-cn-hangzhou-internal.aliyuncs.com"
)
PROPERTIES
(
    "timeout" = "7200",
    "max_filter_ratio" = "0.01"
);

Ingest Parquet from OSS:

LOAD LABEL mydb.parquet_load
(
    DATA INFILE("oss://my-bucket/data/dt=2024-01-01/*.parquet")
    INTO TABLE user_behavior
    FORMAT AS "parquet"
    (user_id, item_id, behavior_type, event_time)
    COLUMNS FROM PATH AS (dt)
)
WITH BROKER
(
    "fs.oss.accessKeyId" = "<ak>",
    "fs.oss.accessKeySecret" = "<sk>",
    "fs.oss.endpoint" = "oss-cn-hangzhou-internal.aliyuncs.com"
)
PROPERTIES ("timeout" = "7200");

Multi-table ingestion (atomic within the same batch):

LOAD LABEL mydb.multi_table_load
(
    DATA INFILE("oss://bucket/orders/*.csv")
    INTO TABLE orders
    COLUMNS TERMINATED BY ","
    (order_id, user_id, amount, order_time),

    DATA INFILE("oss://bucket/order_items/*.csv")
    INTO TABLE order_items
    COLUMNS TERMINATED BY ","
    (item_id, order_id, product_id, quantity, price)
)
WITH BROKER (...)
PROPERTIES ("timeout" = "7200");

---

Storage System Configuration

Alibaba Cloud OSS

WITH BROKER
(
    "fs.oss.accessKeyId" = "<AccessKey ID>",
    "fs.oss.accessKeySecret" = "<AccessKey Secret>",
    "fs.oss.endpoint" = "oss-cn-<region>-internal.aliyuncs.com"
)

Caveats:

• Use the internal endpoint (-internal) to avoid external traffic charges
• Ensure BE nodes can reach OSS over the internal network
• Prefer RAM-role authorization over AK/SK

AWS S3

WITH BROKER
(
    "aws.s3.access_key" = "<access_key>",
    "aws.s3.secret_key" = "<secret_key>",
    "aws.s3.region" = "us-west-2"
)

HDFS (broker-free)

-- Simple HDFS (no HA)
DATA INFILE("hdfs://<namenode>:8020/data/*.csv")
-- No WITH BROKER clause needed

-- HDFS HA
WITH BROKER
(
    "dfs.nameservices" = "my-ha-cluster",
    "dfs.ha.namenodes.my-ha-cluster" = "nn1,nn2",
    "dfs.namenode.rpc-address.my-ha-cluster.nn1" = "nn1_host:8020",
    "dfs.namenode.rpc-address.my-ha-cluster.nn2" = "nn2_host:8020",
    "dfs.client.failover.proxy.provider.my-ha-cluster" =
        "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
)

Azure Blob Storage

WITH BROKER
(
    "azure.blob.storage_account" = "<account>",
    "azure.blob.shared_key" = "<key>"
)
-- File path: wasbs://<container>@<account>.blob.core.windows.net/path/

Google Cloud Storage

WITH BROKER
(
    "gcp.gcs.service_account_email" = "<email>",
    "gcp.gcs.service_account_private_key_id" = "<key_id>",
    "gcp.gcs.service_account_private_key" = "<private_key>"
)
-- File path: gs://<bucket>/path/

---

Format Parameters

Parquet / ORC

• Automatically maps columns to table fields by name (case-insensitive)
• If column names don't match, use (column_list) + SET for mapping
• Separator does not need to be specified

-- Mapping when Parquet column names don't match
DATA INFILE("oss://bucket/data.parquet")
INTO TABLE target_table
FORMAT AS "parquet"
(src_col1, src_col2, src_col3)
SET (
    target_col1 = src_col1,
    target_col2 = CAST(src_col2 AS INT),
    target_col3 = src_col3
)

CSV

JSON (v3.2.3+)

DATA INFILE("oss://bucket/data.json")
INTO TABLE target_table
FORMAT AS "json"
PROPERTIES (
    "jsonpaths" = '["$.id", "$.name", "$.info.age"]',
    "strip_outer_array" = "true"
)

---

Data Transformation

Similar to Stream Load. Supports:

• Column mapping and renaming
• Expression computation via the SET clause
• Row filtering via the WHERE clause
• Extraction of partition fields from the path via COLUMNS FROM PATH AS
• The NEGATIVE keyword for reverse ingestion (Aggregate table scenarios; used to undo previously ingested data)

---

Best Practices

Timeout Settings

The FE parameter broker_load_default_timeout_second controls the default value (default 14400 = 4 hours).

Concurrency Control

• FE parameter max_broker_load_job_concurrency controls the number of Broker Loads that can run simultaneously (default 5)
• Each Broker Load task is split into multiple sub-tasks distributed across BEs
• The BE parameter pipeline_dop affects the parallelism of each sub-task

File Organization

• Use wildcards to match multiple files: oss://bucket/data/2024/01/*/*.parquet
• Each file is best at 128 MB ~ 1 GB; files that are too small increase scheduling overhead
• Parquet/ORC outperforms CSV — better compression ratio with columnar storage and automatic column mapping

Memory Control

• BE parameter load_process_max_memory_limit_percent (default 30%) caps total ingestion memory
• Per-task limit via PROPERTIES "exec_mem_limit" (default 2 GB)
• Increase exec_mem_limit appropriately when ingesting large data volumes

---

Monitoring and Management

View ingestion status:

-- View recent ingestion tasks
SHOW LOAD FROM mydb ORDER BY CreateTime DESC LIMIT 10;

-- View by label
SHOW LOAD FROM mydb WHERE LABEL = "oss_load_20240101";

-- View running tasks
SHOW LOAD FROM mydb WHERE STATE = "LOADING";

Task state transitions:

PENDING → LOADING → FINISHED / CANCELLED

Cancel a task:

CANCEL LOAD FROM mydb WHERE LABEL = "oss_load_20240101";

---

Common Issues

Ecosystem Connectors Guide

Table of Contents

1. Connector Selection 2. Flink Connector 3. Flink CDC 4. Kafka Connector 5. Spark Connector 6. Spark Load 7. Other Tools

---

Connector Selection

---

Flink Connector

Overview

StarRocks provides an official Flink Connector (starrocks-connector-for-apache-flink) that writes Flink DataStream / Table API data to StarRocks via the Stream Load protocol.

Maven Dependency

<dependency>
    <groupId>com.starrocks</groupId>
    <artifactId>flink-connector-starrocks</artifactId>
    <version>${connector.version}</version>
</dependency>

For the version compatibility matrix, see the official documentation.

Flink SQL Example

-- Create the StarRocks sink table
CREATE TABLE sr_sink (
    user_id BIGINT,
    user_name STRING,
    score INT,
    PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://fe_host:9030',
    'load-url' = 'fe_host:8030',
    'database-name' = 'mydb',
    'table-name' = 'user_table',
    'username' = 'root',
    'password' = '',
    'sink.buffer-flush.max-rows' = '50000',
    'sink.buffer-flush.max-bytes' = '67108864',
    'sink.buffer-flush.interval-ms' = '5000',
    'sink.properties.format' = 'json',
    'sink.properties.strip_outer_array' = 'true'
);

-- Read from Kafka and write to StarRocks
INSERT INTO sr_sink
SELECT user_id, user_name, score FROM kafka_source;

Key Parameters

Exactly-Once Configuration

'sink.semantic' = 'exactly-once',
'sink.label-prefix' = 'flink_load'

You must also enable Flink Checkpoint; StarRocks implements 2PC via the Stream Load Transaction Interface.

Performance Tuning

• Enlarge buffers: sink.buffer-flush.max-bytes = 128MB to reduce flush frequency
• Adjust interval: sink.buffer-flush.interval-ms = 10000 to balance latency and throughput
• Parallelism (cap, not just target): sink.parallelism should be ≤ the Kafka partition count of the source topic — extra sinks idle, and over-parallelizing high-throughput CDC accelerates tablet version buildup. The Kafka partition count is the natural ceiling; match it as the upper bound, do not exceed it.
• Format choice: JSON is more flexible but slower; CSV is faster (use CSV for high throughput)

---

Flink CDC

Overview

Flink CDC is a Flink-based Change Data Capture framework that can sync changes from MySQL, PostgreSQL, Oracle, MongoDB, and other databases to StarRocks in real time.

Typical Architecture

MySQL (binlog) -> Flink CDC Source -> Flink Connector -> StarRocks (Primary Key Table)

Flink SQL Example (MySQL CDC -> StarRocks)

-- MySQL CDC Source
CREATE TABLE mysql_source (
    id BIGINT,
    name STRING,
    age INT,
    update_time TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'mysql_host',
    'port' = '3306',
    'username' = 'cdc_user',
    'password' = 'cdc_pass',
    'database-name' = 'source_db',
    'table-name' = 'user_table',
    'server-time-zone' = 'Asia/Shanghai'
);

-- StarRocks sink (Primary Key table)
CREATE TABLE sr_sink (
    id BIGINT,
    name STRING,
    age INT,
    update_time TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://fe_host:9030',
    'load-url' = 'fe_host:8030',
    'database-name' = 'target_db',
    'table-name' = 'user_table',
    'username' = 'root',
    'password' = '',
    'sink.properties.format' = 'json',
    'sink.properties.strip_outer_array' = 'true'
);

-- Real-time sync
INSERT INTO sr_sink SELECT * FROM mysql_source;

Caveats

• The target table must be a Primary Key table
• The Flink Connector automatically handles INSERT / UPDATE / DELETE events. Under the hood it still uses StarRocks' `__op` contract, which is a binary pair you must state in full: literal column __op, with `__op=0` → UPSERT and `__op=1` → DELETE. Both mappings must appear in the response (RowKind +I / +U → __op=0; RowKind -D → __op=1) — explaining only the DELETE side is incomplete. This holds whether the question is about DELETE or UPSERT; users need both values to debug "DELETE not applied" and the symmetric "UPSERT silently overwritten" failures.
• Enable Flink Checkpoint to guarantee consistency
• The full phase may consume a lot of memory; size the Flink TaskManager memory appropriately

---

Kafka Connector

Overview

The StarRocks Kafka Connector is a Kafka Connect sink connector that writes data from a Kafka topic to StarRocks.

Configuration Example

{
    "name": "starrocks-sink",
    "config": {
        "connector.class": "com.starrocks.connector.kafka.StarRocksSinkConnector",
        "topics": "user_events",
        "starrocks.http.url": "fe_host:8030",
        "starrocks.jdbc.url": "jdbc:mysql://fe_host:9030",
        "starrocks.username": "root",
        "starrocks.password": "",
        "starrocks.database.name": "mydb",
        "starrocks.table.name": "user_events",
        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter.schemas.enable": "false",
        "sink.properties.format": "json",
        "sink.properties.strip_outer_array": "true",
        "sink.buffer-flush.max-bytes": "67108864",
        "sink.buffer-flush.interval-ms": "5000"
    }
}

Supported Data Formats

• JSON: most common
• CSV: high-throughput scenarios
• Avro: with Schema Registry (v3.0+)
• Protobuf: with Schema Registry (v3.0+)

vs Routine Load

---

Spark Connector

Overview

The StarRocks Spark Connector enables batch writes from Spark to StarRocks, using Stream Load under the hood.

DataFrame Example

df.write
  .format("starrocks")
  .option("starrocks.fe.http.url", "fe_host:8030")
  .option("starrocks.fe.jdbc.url", "jdbc:mysql://fe_host:9030")
  .option("starrocks.user", "root")
  .option("starrocks.password", "")
  .option("starrocks.table.identifier", "mydb.target_table")
  .option("starrocks.write.properties.format", "csv")
  .option("starrocks.write.buffer.size", "104857600")
  .option("starrocks.write.flush.interval.ms", "10000")
  .mode("append")
  .save()

Spark SQL Example

CREATE TABLE sr_table
USING starrocks
OPTIONS (
    "starrocks.fe.http.url" = "fe_host:8030",
    "starrocks.fe.jdbc.url" = "jdbc:mysql://fe_host:9030",
    "starrocks.user" = "root",
    "starrocks.password" = "",
    "starrocks.table.identifier" = "mydb.target_table"
);

INSERT INTO sr_table SELECT * FROM hive_table WHERE dt = '2024-01-01';

---

Spark Load

Overview

Spark Load uses an external Spark cluster for ETL preprocessing, suitable for TB-scale first-time bulk migration.

Note: Spark Load does not support Primary Key tables.

Create the Spark Resource

CREATE EXTERNAL RESOURCE "spark_resource"
PROPERTIES (
    "type" = "spark",
    "spark.master" = "yarn",
    "spark.submit.deployMode" = "cluster",
    "spark.executor.memory" = "4g",
    "spark.yarn.queue" = "default",
    "working_dir" = "hdfs://namenode/tmp/starrocks_spark_load",
    "broker" = "hdfs_broker"
);

Submit Spark Load

LOAD LABEL mydb.spark_migration
(
    DATA INFILE("hdfs://namenode/hive/warehouse/source_table/*")
    INTO TABLE target_table
    FORMAT AS "parquet"
)
WITH RESOURCE "spark_resource"
PROPERTIES ("timeout" = "86400");

Applicable Scenarios

• First-time full migration of large Hive tables (TB scale)
• Need to build a global BITMAP dictionary
• An existing Spark/YARN cluster is available

---

Other Tools

DataX

Alibaba's open-source offline data sync tool, which writes via the StarRocksWriter plugin.

{
    "writer": {
        "name": "starrockswriter",
        "parameter": {
            "username": "root",
            "password": "",
            "database": "mydb",
            "table": "target_table",
            "loadUrl": ["fe_host:8030"],
            "jdbcUrl": "jdbc:mysql://fe_host:9030",
            "column": ["col1", "col2", "col3"],
            "loadProps": {
                "format": "json",
                "strip_outer_array": true
            }
        }
    }
}

CloudCanal

Cloud data sync SaaS service supporting:

• MySQL / PostgreSQL / Oracle -> StarRocks real-time sync
• Integrated full + incremental
• No code; configure via the web console
• Automatic schema migration

SMT (StarRocks Migration Tool)

StarRocks's official migration tool:

• Automatically converts other databases' DDL to StarRocks DDL
• Automatically generates DataX configuration files
• Supports MySQL, PostgreSQL, Oracle, and Hive

INSERT INTO and Pipe Detailed Guide

Table of Contents

1. INSERT INTO Overview 2. INSERT INTO VALUES 3. INSERT INTO SELECT 4. INSERT INTO SELECT FROM FILES() 5. INSERT OVERWRITE 6. Pipe Continuous Ingestion 7. Best Practices

---

INSERT INTO Overview

INSERT INTO is the most flexible ingestion method, supporting multiple data sources:

• VALUES: write a small number of data rows directly
• SELECT: ingest from internal tables, external tables, or materialized views
• SELECT FROM FILES(): ingest from cloud-storage files (v3.1+)

Executes synchronously and returns the result immediately.

---

INSERT INTO VALUES

Applicable scenarios: testing, demos, very small data writes.

INSERT INTO user_table (id, name, age, city)
VALUES
    (1, 'Alice', 30, 'Beijing'),
    (2, 'Bob', 25, 'Shanghai'),
    (3, 'Charlie', 35, 'Hangzhou');

Note: Do not use INSERT INTO VALUES at high frequency in production — each execution generates a new tablet version, which easily triggers TOO_MANY_VERSION.

---

INSERT INTO SELECT

Ingest from an internal table:

-- Ingest from a staging table into the target table
INSERT INTO target_table
SELECT * FROM staging_table WHERE dt = '2024-01-01';

-- Ingest the aggregated result of one table into another
INSERT INTO daily_summary (dt, user_count, total_amount)
SELECT
    DATE(event_time) AS dt,
    COUNT(DISTINCT user_id),
    SUM(amount)
FROM order_detail
WHERE event_time >= '2024-01-01'
GROUP BY DATE(event_time);

Ingest from an external table / catalog:

-- Ingest from a Hive catalog
INSERT INTO starrocks_table
SELECT * FROM hive_catalog.hive_db.hive_table
WHERE dt = '2024-01-01';

-- Ingest from a JDBC catalog (MySQL)
INSERT INTO starrocks_table
SELECT * FROM jdbc_catalog.mysql_db.mysql_table
WHERE id > 1000;

-- Ingest from an Iceberg catalog
INSERT INTO starrocks_table
SELECT col1, col2, col3 FROM iceberg_catalog.db.iceberg_table;

Memory control:

• Controlled via the session variable exec_mem_limit (default 2GB)
• Increase appropriately for large data volumes: SET exec_mem_limit = 8589934592; (8GB)
• query_timeout controls the timeout (default 300s); increase for large data volumes

---

INSERT INTO SELECT FROM FILES()

v3.1+ feature: ingest directly from cloud-storage files without creating an external table.

Basic syntax:

INSERT INTO target_table
SELECT * FROM FILES(
    "path" = "oss://bucket/data/2024/01/*.parquet",
    "format" = "parquet",
    "fs.oss.accessKeyId" = "<ak>",
    "fs.oss.accessKeySecret" = "<sk>",
    "fs.oss.endpoint" = "oss-cn-hangzhou-internal.aliyuncs.com"
);

Supported storage systems:

• Alibaba Cloud OSS: oss://bucket/path/
• AWS S3: s3://bucket/path/
• HDFS: hdfs://namenode:port/path/
• Azure: wasbs://container@account.blob.core.windows.net/path/
• GCS: gs://bucket/path/
• MinIO: s3://bucket/path/ (with the endpoint parameter)

Supported file formats:

• Parquet (v3.1+)
• ORC (v3.1+)
• CSV (v3.3+)

Column mapping and transformation:

-- Select specific columns and transform them
INSERT INTO target_table (user_id, event_time, amount)
SELECT
    uid,
    CAST(ts AS DATETIME),
    price * quantity
FROM FILES(
    "path" = "oss://bucket/data/*.parquet",
    "format" = "parquet",
    ...
);

Auto table creation (v3.2+):

-- Automatically create the table based on the file schema
CREATE TABLE auto_table AS
SELECT * FROM FILES(
    "path" = "oss://bucket/data/sample.parquet",
    "format" = "parquet",
    ...
);

FILES() vs Broker Load comparison:

---

INSERT OVERWRITE

Atomically replace partition data: writes into a temporary partition first and atomically swaps on success.

-- Overwrite a specific partition
INSERT OVERWRITE target_table PARTITION (p20240101)
SELECT * FROM staging_table WHERE dt = '2024-01-01';

-- Auto-inferred partitions (v3.2+)
INSERT OVERWRITE target_table
SELECT * FROM FILES(
    "path" = "oss://bucket/data/dt=2024-01-01/*.parquet",
    "format" = "parquet",
    ...
);

Typical scenarios:

• Daily full refresh of certain partitions
• T+1 report data refresh
• Data repair (re-ingest a day's data)

---

Pipe Continuous Ingestion

Overview

Pipe (v3.2+) is a long-running asynchronous ingestion mechanism that can automatically watch for new files in cloud storage and ingest them. Internally it is implemented using INSERT INTO SELECT FROM FILES().

Key characteristics:

• Automatically splits large file sets into smaller batches executed in order
• Supports AUTO_INGEST=TRUE for automatic discovery of new files
• Supports Parquet and ORC formats
• Suitable for continuous file ingestion at the 100 GB ~ TB scale

Creation Syntax

CREATE PIPE [IF NOT EXISTS] <pipe_name>
PROPERTIES (
    "AUTO_INGEST" = "TRUE",
    "POLL_INTERVAL" = "60",    -- Scan interval in seconds (default 300; lower for frequent file arrivals)
    "BATCH_SIZE" = "1GB",      -- Data volume per batch
    "BATCH_FILES" = "256"      -- Number of files per batch
)
AS INSERT INTO target_table
SELECT * FROM FILES(
    "path" = "oss://bucket/data/incoming/*.parquet",
    "format" = "parquet",
    "fs.oss.accessKeyId" = "<ak>",
    "fs.oss.accessKeySecret" = "<sk>",
    "fs.oss.endpoint" = "oss-cn-hangzhou-internal.aliyuncs.com"
);

File Discovery Mechanism

• OSS/S3: detects new/changed files via the file's ETag
• HDFS: detects via LastModifiedTime
• With AUTO_INGEST=TRUE, Pipe polls continuously; POLL_INTERVAL controls the interval
• Files already ingested are not re-ingested (deduplicated by file path + ETag)

Management and Monitoring

-- List all pipes
SHOW PIPES;

-- View pipe details
SHOW PIPE mydb.my_pipe;

-- View file ingestion status
SELECT * FROM information_schema.pipe_files
WHERE pipe_name = 'my_pipe';

-- Suspend
SUSPEND PIPE mydb.my_pipe;

-- Resume
RESUME PIPE mydb.my_pipe;

-- Re-ingest a failed file
ALTER PIPE mydb.my_pipe RETRY FILE "oss://bucket/data/failed_file.parquet";

-- Drop
DROP PIPE mydb.my_pipe;

Pipe vs Routine Load Comparison

---

Best Practices

INSERT INTO Usage Recommendations

Pipe Usage Recommendations

• File size: 128 MB ~ 1 GB per file is optimal
• BATCH_SIZE: set based on cluster memory; 512 MB ~ 2 GB is recommended
• POLL_INTERVAL: 30~60s when file arrival is frequent, 300~600s when infrequent
• Error handling: periodically check pipe_files for files with LOAD_STATE = 'ERROR' and re-ingest with RETRY FILE
• Note: every Pipe batch generates a tablet version; with frequent small files watch out for compaction pressure

Ingestion Performance Tuning Guide

Table of Contents

1. Performance Baselines 2. Memory Optimization 3. Concurrency Optimization 4. Timeout Configuration 5. Compaction Management 6. Key FE Parameters 7. Key BE Parameters 8. Resource Isolation 9. Ingestion Method Performance Comparison

---

Performance Baselines

Typical throughput reference (per BE; not absolute; varies with hardware/network/data characteristics):

Key factors influencing throughput:

• File format: Parquet/ORC > CSV > JSON
• Column count and types: more/complex columns (JSON/BITMAP) write more slowly
• Target table model: Duplicate > Primary Key > Aggregate
• Indexes: Bloom Filter / Bitmap indexes add write overhead
• Compaction state: writes slow down when versions are piling up

---

Memory Optimization

BE Ingestion Memory Management

Ingestion uses the BE's load memory pool, which shares physical memory with the query memory pool.

Global limits:

Effective memory cap = min(BE total memory x 90% x 90% x 30%, 100GB)

Per-task limit:

Tuning recommendations:

• Increase exec_mem_limit to 4~8 GB when ingesting large data volumes
• When running multiple ingestion tasks in parallel, ensure the total memory does not exceed the limit
• If BE OOM happens frequently, reduce load_process_max_memory_limit_percent

Write Buffer

• Too small -> frequent flushes, many small files, hurts query performance
• Too large -> risk of RPC timeout (tablet_writer_rpc_timeout_sec)
• Keep the default at 100 MB; in extreme cases adjust to 50~200 MB

---

Concurrency Optimization

Parallel Stream Loads

• Multiple Stream Loads can be submitted simultaneously; the BE handles them in parallel threads automatically
• Parallel ingestion into different partitions works best (no lock contention)
• Parallel ingestion into the same tablet of the same partition incurs lock waits
• Recommended concurrency: min(BE count x 2, target table partition count)

Broker Load Concurrency

• Each Broker Load is internally parallelized (work spread across BEs)
• Raise concurrency when there are many independent ingestion jobs

Routine Load Concurrency

See the concurrency and performance tuning section of routine-load.md.

Core formula:

actual_concurrent = min(alive_be_number, kafka_partition_number,
                        desired_concurrent_number, max_routine_load_task_concurrent_num)

Transaction Concurrency

• Each ingestion task occupies one transaction
• In high-frequency ingestion scenarios, watch out for hitting this limit
• New ingestions queue up when the limit is exceeded

---

Timeout Configuration

Timeout Parameters by Ingestion Method

FE-Side Timeout Parameters

BE-Side Timeout Parameters

Timeout Estimation Formula

recommended timeout = data volume (MB) / expected throughput (MB/s) x safety factor (1.5~2)

---

Compaction Management

Why Compaction Affects Ingestion

Each ingestion generates a new tablet version. If the ingestion frequency exceeds the compaction rate, versions pile up:

• More than 1000 versions -> new ingestion is rejected (TOO_MANY_VERSION)
• Queries slow down when versions pile up (must merge multiple versions)

Key BE Parameters

Tuning Recommendations

Monitoring Compaction State

-- Check the number of tablet versions
SHOW TABLET FROM table_name;

-- Tablets with too many versions
SELECT * FROM information_schema.be_tablets
WHERE num_version > 500
ORDER BY num_version DESC;

---

Key FE Parameters

---

Key BE Parameters

---

Resource Isolation

Resource Contention Between Ingestion and Queries

Ingestion and queries share the BE's CPU, memory, and IO resources. Under heavy load, they affect each other.

Isolation strategies:

1. Time-based isolation: schedule large-batch ingestion during off-peak hours 2. Resource Group (v3.1+): assign ingestion and queries to different resource groups

-- Create a resource group dedicated to ingestion
CREATE RESOURCE GROUP load_rg
TO (user = 'load_user')
WITH (
    'cpu_weight' = '4',
    'mem_limit' = '30%',
    'type' = 'normal'
);

-- Create a resource group dedicated to queries
CREATE RESOURCE GROUP query_rg
TO (user = 'query_user')
WITH (
    'cpu_weight' = '6',
    'mem_limit' = '50%',
    'type' = 'normal'
);

3. BE node isolation: in shared-data architecture, use different CN groups for ingestion and queries

---

Ingestion Method Performance Comparison

Primary Key Table Update Modes Guide

Table of Contents

1. Overview 2. UPSERT Mode 3. DELETE Mode 4. Partial Column Update 5. Conditional Update 6. Update Configuration per Ingestion Method 7. Best Practices

---

Overview

StarRocks Primary Key tables support row-level updates and deletes during ingestion, suitable for CDC sync, dimension table updates, real-time data corrections, and similar scenarios.

Supported operations:

• UPSERT: update if exists, insert if not (default behavior)
• DELETE: delete rows by primary key
• Partial column update: update only the specified columns; the rest remain unchanged
• Conditional update: update only when a condition is met

Supported ingestion methods:

• Stream Load
• Broker Load
• Routine Load
• Flink Connector / Kafka Connector

Methods that do not support update operations:

• Spark Load
• INSERT INTO (implemented via SQL semantics, not the __op mechanism)

---

UPSERT Mode

UPSERT is the default ingestion behavior for Primary Key tables — no extra configuration is required.

How it works:

• Ingested data is matched against existing rows by primary key
• Primary key exists -> update the entire row
• Primary key does not exist -> insert a new row

Stream Load example:

# UPSERT by default; no extra parameters needed
curl --location-trusted -u root: \
    -H "label:upsert_20240101" \
    -H "column_separator:," \
    -T /data/user_updates.csv \
    http://fe_host:8030/api/mydb/user_table/_stream_load

---

DELETE Mode

Use the __op field to mark a delete operation. __op = 0 means UPSERT, __op = 1 means DELETE.

Option 1: data contains an __op column

CSV data:

1,Alice,30,0
2,Bob,25,1
3,Charlie,35,0

Stream Load:

curl --location-trusted -u root: \
    -H "label:upsert_delete_20240101" \
    -H "column_separator:," \
    -H "columns:id, name, age, __op" \
    -T /data/cdc_data.csv \
    http://fe_host:8030/api/mydb/user_table/_stream_load

Option 2: the entire batch is DELETE

# The whole batch is deletes; specify via Header
curl --location-trusted -u root: \
    -H "label:delete_batch" \
    -H "column_separator:," \
    -H "columns:id, name, age" \
    -H "__op:1" \
    -T /data/delete_keys.csv \
    http://fe_host:8030/api/mydb/user_table/_stream_load

DELETE with JSON Format

// JSON data contains the __op field directly
[
    {"id": 1, "name": "Alice", "age": 30, "__op": 0},
    {"id": 2, "__op": 1}
]

curl --location-trusted -u root: \
    -H "format:json" \
    -H "strip_outer_array:true" \
    -H "columns:id, name, age, __op" \
    -T /data/cdc_data.json \
    http://fe_host:8030/api/mydb/user_table/_stream_load

---

Partial Column Update

Update only the specified columns; columns not specified keep their original values. Suited for scenarios where different data sources update different columns.

Row Mode (default, v3.0+)

Suitable when only a few columns are being updated. During ingestion, the original row is read, merged with the new values, and written.

Stream Load:

# Update only the age column; other columns unchanged
curl --location-trusted -u root: \
    -H "label:partial_update_age" \
    -H "column_separator:," \
    -H "partial_update:true" \
    -H "columns:id, age" \
    -T /data/age_updates.csv \
    http://fe_host:8030/api/mydb/user_table/_stream_load

Note: the data must contain all primary key columns.

Routine Load:

CREATE ROUTINE LOAD mydb.partial_update_job ON user_table
COLUMNS (id, score)
PROPERTIES (
    "partial_update" = "true",
    "desired_concurrent_number" = "3"
)
FROM KAFKA (...);

Column Mode (v3.1+)

Suitable when the table has many columns but only a few need to be updated. Incremental column data is appended directly and merged during compaction.

Stream Load:

curl --location-trusted -u root: \
    -H "label:column_mode_update" \
    -H "column_separator:," \
    -H "partial_update:true" \
    -H "partial_update_mode:column" \
    -H "columns:id, score" \
    -T /data/score_updates.csv \
    http://fe_host:8030/api/mydb/user_table/_stream_load

Row Mode vs Column Mode

---

Conditional Update

Performs the update only when a condition is met — useful for guaranteeing data ordering (only allow updates from newer data).

Stream Load:

# Update only when the ingested data's update_time > the existing row's update_time
curl --location-trusted -u root: \
    -H "label:conditional_update" \
    -H "column_separator:," \
    -H "merge_condition:update_time" \
    -H "columns:id, name, score, update_time" \
    -T /data/updates.csv \
    http://fe_host:8030/api/mydb/user_table/_stream_load

How it works:

• merge_condition specifies a column name
• During ingestion, compare new vs. old values: update only when new > old
• Rows that do not satisfy the condition are skipped (not counted as error rows)

Routine Load:

CREATE ROUTINE LOAD mydb.conditional_load ON user_table
COLUMNS (id, name, score, update_time)
PROPERTIES (
    "merge_condition" = "update_time"
)
FROM KAFKA (...);

Typical scenarios:

• CDC data arrives out of order; use update_time to ensure only the latest version is written
• Multiple sources writing to the same table; use a version number to prevent old data from overwriting new

---

Update Configuration per Ingestion Method

Stream Load

Broker Load

LOAD LABEL mydb.broker_upsert
(
    DATA INFILE("oss://bucket/data.csv")
    INTO TABLE pk_table
    COLUMNS TERMINATED BY ","
    (id, name, age, __op)
)
WITH BROKER (...)
PROPERTIES ("partial_update" = "true");  -- Partial column update is set in PROPERTIES

Routine Load

CREATE ROUTINE LOAD mydb.pk_update ON pk_table
COLUMNS (id, name, age, __op)
PROPERTIES (
    "partial_update" = "true",           -- Partial column update
    "merge_condition" = "update_time"    -- Conditional update
)
FROM KAFKA (...);

Flink Connector

-- Flink SQL: automatically handles INSERT/UPDATE/DELETE events
-- Primary Key table + Flink CDC -> automatic UPSERT/DELETE
CREATE TABLE sr_sink (
    id BIGINT,
    name STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'starrocks',
    ...
    'sink.properties.partial_update' = 'true',      -- Partial column update
    'sink.properties.merge_condition' = 'update_ts'  -- Conditional update
);

---

Best Practices

UPSERT / DELETE Scenarios

Performance Caveats

• Partial column update (Row mode) requires reading back the original row; write performance is 30~50% lower than full-column UPSERT
• Partial column update (Column mode) writes are fast but compaction load increases
• Conditional update requires reading the original row for comparison; slight impact on write performance
• DELETE operations in Primary Key tables are tombstones; they are truly purged during compaction
• Primary Key table write performance is roughly 50~70% of Duplicate Key tables (because of primary-key index maintenance)

Patterns to Avoid

Routine Load Detailed Guide

Table of Contents

1. Overview 2. Creation Syntax 3. Kafka Parameter Configuration 4. Data Formats 5. Concurrency and Performance Tuning 6. Management and Monitoring 7. Best Practices 8. Common Issues

---

Overview

Routine Load is a long-running consumer task that continuously pulls data from Kafka (or Pulsar) and writes it to StarRocks. The FE splits the job into multiple sub-tasks and dispatches them to BEs.

Key characteristics:

• Runs continuously; no need to manually trigger each ingestion
• Guarantees exactly-once semantics (based on Kafka offsets + StarRocks transactions)
• Supports CSV, JSON, and Avro (v3.0.1+) formats
• Supports UPSERT / DELETE on Primary Key tables
• Auto-pauses on excessive errors and can be resumed

How it works:

FE (RoutineLoadMgr) -> splits into tasks (each task corresponds to one or more Kafka partitions)
    -> dispatches to BE -> BE consumes a batch from Kafka -> writes to StarRocks -> commits offset
    -> FE schedules the next round of tasks

---

Creation Syntax

Basic syntax:

CREATE ROUTINE LOAD <database>.<job_name> ON <table_name>
[COLUMNS TERMINATED BY "<separator>"]
[COLUMNS (<column_list>)]
[WHERE <predicate>]
[PARTITION (<partition_list>)]
PROPERTIES
(
    "desired_concurrent_number" = "3",
    "max_error_number" = "1000",
    "max_batch_interval" = "20",
    "max_batch_rows" = "200000",
    "format" = "json",
    ...
)
FROM KAFKA
(
    "kafka_broker_list" = "broker1:9092,broker2:9092",
    "kafka_topic" = "my_topic",
    "property.group.id" = "starrocks_consumer_group",
    "property.kafka_default_offsets" = "OFFSET_END"
);

CSV example:

CREATE ROUTINE LOAD mydb.kafka_csv_load ON user_behavior
COLUMNS TERMINATED BY ","
COLUMNS (user_id, item_id, behavior_type, ts_str, event_time = str_to_date(ts_str, '%Y-%m-%d %H:%i:%s'))
WHERE behavior_type IN ('buy', 'cart')
PROPERTIES
(
    "desired_concurrent_number" = "5",
    "max_error_number" = "1000",
    "max_batch_interval" = "15",
    "strict_mode" = "true"
)
FROM KAFKA
(
    "kafka_broker_list" = "kafka1:9092,kafka2:9092,kafka3:9092",
    "kafka_topic" = "user_behavior_topic",
    "property.kafka_default_offsets" = "OFFSET_END"
);

JSON example:

CREATE ROUTINE LOAD mydb.kafka_json_load ON user_behavior
COLUMNS (user_id, item_id, behavior_type, event_time)
PROPERTIES
(
    "desired_concurrent_number" = "3",
    "format" = "json",
    "jsonpaths" = '["$.uid", "$.iid", "$.action", "$.ts"]',
    "max_error_number" = "500"
)
FROM KAFKA
(
    "kafka_broker_list" = "kafka1:9092",
    "kafka_topic" = "user_events",
    "property.kafka_default_offsets" = "OFFSET_END"
);

Avro example (v3.0.1+):

CREATE ROUTINE LOAD mydb.kafka_avro_load ON user_behavior
PROPERTIES
(
    "format" = "avro",
    "confluent.schema.registry.url" = "http://schema-registry:8081",
    "desired_concurrent_number" = "3"
)
FROM KAFKA
(
    "kafka_broker_list" = "kafka1:9092",
    "kafka_topic" = "user_events_avro"
);

---

Kafka Parameter Configuration

PROPERTIES Parameters

FROM KAFKA Parameters

Kafka SSL/SASL Authentication Example

FROM KAFKA
(
    "kafka_broker_list" = "kafka1:9093",
    "kafka_topic" = "secure_topic",
    "property.security.protocol" = "SASL_SSL",
    "property.sasl.mechanism" = "PLAIN",
    "property.sasl.jaas.config" =
        "org.apache.kafka.common.security.plain.PlainLoginModule required username='user' password='pass';"
)

---

Data Formats

CSV

• Default column separator is \t; modify via COLUMNS TERMINATED BY
• Each Kafka message is one CSV row
• Column mapping and transformation are the same as Stream Load

JSON

• Each Kafka message is a JSON object
• Use jsonpaths to map fields precisely
• Supports nested field extraction

Avro (v3.0.1+)

• Requires Schema Registry
• Auto-maps by field name
• Set confluent.schema.registry.url to the Schema Registry address
• Supports basic types and logical types (DATE, TIMESTAMP, etc.)

---

Concurrency and Performance Tuning

Computing Actual Concurrency

actual_concurrent = min(
    alive_be_number,
    kafka_partition_number,
    desired_concurrent_number,
    max_routine_load_task_concurrent_num    -- FE parameter, default 5
)

Important: Setting desired_concurrent_number higher than the Kafka partition count is pointless — extra tasks idle.

Troubleshooting Insufficient Consumption Rate

Inspect the per-task execution in the BE logs:

# Search in the BE log
grep "routine load task" be.INFO
# Pay attention to the left_bytes field:
# left_bytes < 0 -> the per-round data volume hit the cap; increase max_batch_size
# left_bytes >= 0 -> consumption time ran out; increase max_batch_interval or routine_load_task_consume_second

Tuning Steps

1. Raise concurrency: Increase desired_concurrent_number (not exceeding the Kafka partition count) 2. Enlarge batches: Increase max_batch_size and max_batch_rows 3. Increase consumption time: Increase max_batch_interval 4. FE parameters: Increase max_routine_load_task_concurrent_num (default 5) and max_routine_load_task_num_per_be (default 16) 5. Expand Kafka partitions: If the concurrency bottleneck is the Kafka partition count, expand Kafka partitions

Key FE Parameters

---

Management and Monitoring

Viewing Task State

-- List all Routine Load jobs
SHOW ROUTINE LOAD FROM mydb;

-- View details of a specific job
SHOW ROUTINE LOAD FOR mydb.kafka_csv_load;

-- View running tasks
SHOW ROUTINE LOAD TASK WHERE JobName = "kafka_csv_load";

Task states:

• NEED_SCHEDULE: waiting to be scheduled
• RUNNING: running normally
• PAUSED: paused due to errors (resumable)
• STOPPED: manually stopped (not resumable)
• CANCELLED: cancelled due to errors (not resumable)

Key monitoring fields:

• ReasonOfStateChanged: reason for the state change (check when PAUSED)
• ErrorLogUrls: URLs of error logs
• OtherMsg: information such as Kafka offset lag
• Statistics: rows and bytes consumed

Operation Commands

-- Pause
PAUSE ROUTINE LOAD FOR mydb.kafka_csv_load;

-- Resume
RESUME ROUTINE LOAD FOR mydb.kafka_csv_load;

-- Stop (not resumable)
STOP ROUTINE LOAD FOR mydb.kafka_csv_load;

-- Modify parameters
ALTER ROUTINE LOAD FOR mydb.kafka_csv_load
PROPERTIES ("desired_concurrent_number" = "5");

-- Modify Kafka offsets
ALTER ROUTINE LOAD FOR mydb.kafka_csv_load
FROM KAFKA ("kafka_offsets" = "0:12345,1:23456,2:34567");

---

Best Practices

Job Design

• One Routine Load job corresponds to one Kafka topic and one table
• Set desired_concurrent_number to the Kafka partition count (but not above the BE count)
• Set max_error_number based on business tolerance; in production, prefer > 0 to avoid frequent PAUSE

Latency Control

• max_batch_interval controls the maximum latency (default 10s)
• Lowering this value reduces latency but increases the frequency of small-batch writes
• Recommended range: 10~30s (balances latency and compaction pressure)

Error Handling

• After PAUSED, check ReasonOfStateChanged and ErrorLogUrls
• Common PAUSE reasons:
• Data format errors exceeding max_error_number
• Kafka offsets out of range (data was cleaned up)
• Target table was dropped
• After fixing, resume with RESUME ROUTINE LOAD

Offset Management

• StarRocks manages offsets on its own (does not rely on the Kafka consumer group)
• OFFSET_END: consume from the latest (recommended for new jobs)
• OFFSET_BEGINNING: consume from the beginning (when backfill is needed)
• Specify exact offsets: ALTER ROUTINE LOAD ... FROM KAFKA ("kafka_offsets" = "0:12345")

---

Common Issues

Stream Load Detailed Guide

Table of Contents

1. Overview 2. HTTP Syntax 3. Core Parameters 4. CSV Ingestion 5. JSON Ingestion 6. Data Transformation 7. Multi-Table Ingestion 8. Best Practices 9. Common Issues

---

Overview

Stream Load pushes a local file or data stream to StarRocks via HTTP PUT and synchronously returns the result. Suitable for batch ingestion of a single file < 10GB.

Key characteristics:

• Synchronous operation; success/failure is known immediately
• Supports CSV and JSON formats
• Supports data transformation during ingestion (column mapping, function computation, filtering)
• Supports UPSERT / DELETE on Primary Key tables
• Supports multi-table transactional ingestion (v3.x+)

---

HTTP Syntax

Basic syntax:

curl --location-trusted -u <user>:<password> \
    -H "label:<label>" \
    -H "column_separator:<sep>" \
    -T <file_path> \
    http://<fe_host>:<fe_http_port>/api/<database>/<table>/_stream_load

Request target:

• Send to the FE HTTP port (default 8030); the FE redirects to a BE
• You can also send directly to the BE HTTP port (default 8040), skipping the FE redirect
• --location-trusted makes curl follow the redirect while carrying the auth info

Sample response:

{
    "TxnId": 1003,
    "Label": "my_label_20240101",
    "Status": "Success",
    "Message": "OK",
    "NumberTotalRows": 1000000,
    "NumberFilteredRows": 0,
    "NumberUnselectedRows": 0,
    "LoadBytes": 52943670,
    "LoadTimeMs": 3245
}

Status values:

• Success: ingestion succeeded
• Publish Timeout: data was written but publish timed out; it will complete automatically and is treated as success
• Label Already Exists: a label with the same name already exists (idempotency safeguard)
• Fail: ingestion failed; check Message and ErrorURL

---

Core Parameters

Passed via HTTP Header:

---

CSV Ingestion

Basic CSV ingestion:

curl --location-trusted -u root: \
    -H "label:csv_load_20240101" \
    -H "column_separator:," \
    -H "skip_header:1" \
    -T /data/user_data.csv \
    http://fe_host:8030/api/mydb/user_table/_stream_load

Handling special separators:

# Hive default separator \x01
-H "column_separator:\x01"

# Multi-character separator
-H "column_separator:||"

# Tab separator (default)
-H "column_separator:\t"

Handling quoted CSV:

# CSV fields are wrapped in double quotes and contain commas
# "John","New York, NY","30"
-H "column_separator:," \
-H "enclose:\"" \
-H "escape:\\"

Handling NULL values:

• NULL is represented as \N in CSV (default)
• Custom NULL mapping can be done via the columns expression using if / nullif

Handling compressed files:

curl --location-trusted -u root: \
    -H "label:gz_load" \
    -H "compression:gz" \
    -T /data/user_data.csv.gz \
    http://fe_host:8030/api/mydb/user_table/_stream_load

---

JSON Ingestion

Basic JSON ingestion:

# Single-line JSON (one JSON object per line)
curl --location-trusted -u root: \
    -H "format:json" \
    -H "strip_outer_array:true" \
    -T /data/users.json \
    http://fe_host:8030/api/mydb/user_table/_stream_load

JSON data formats:

// Format 1: JSON array (requires strip_outer_array:true)
[
    {"id": 1, "name": "Alice", "age": 30},
    {"id": 2, "name": "Bob", "age": 25}
]

// Format 2: NDJSON (one JSON per line; strip_outer_array not required)
{"id": 1, "name": "Alice", "age": 30}
{"id": 2, "name": "Bob", "age": 25}

Precise mapping with jsonpaths:

# When JSON field names do not exactly match table column names
-H 'jsonpaths:["$.user_id", "$.user_name", "$.user_age"]' \
-H 'columns:id, name, age'

Extracting nested JSON:

# JSON: {"data": {"id": 1, "info": {"name": "Alice"}}}
-H 'jsonpaths:["$.data.id", "$.data.info.name"]' \
-H 'columns:id, name'

Performance caveats for JSON ingestion:

• JSON parsing is 2-5x slower than CSV
• Prefer CSV for large data volumes
• JSON files should not exceed 1-2 GB
• streaming_load_max_batch_size_mb (BE parameter) caps JSON file size

---

Data Transformation

Stream Load supports lightweight ETL during ingestion via the columns parameter.

Column renaming:

# File has 5 columns, table has 3 columns; skip columns 2 and 4
-H "columns:col1, tmp_col2, col3, tmp_col4, col5"
# Columns prefixed with tmp_ that don't exist in the table are automatically ignored

Column computation:

# File columns: date_str, amount_cents
# Table columns: dt (DATE), amount (DECIMAL)
-H "columns:date_str, amount_cents, dt=str_to_date(date_str,'%Y%m%d'), amount=amount_cents/100"

Row filtering:

# Only ingest rows where age > 18
-H "where:age > 18"

Extracting partition fields from file paths:

# File path contains partition info: /data/dt=20240101/region=us/data.csv
-H "columns:col1, col2, col3" \
-H "column_from_path:dt, region"

Common transformation functions:

• str_to_date(str, format) — string to date
• if(condition, true_val, false_val) — conditional expression
• nullif(expr1, expr2) — returns NULL if values are equal
• ifnull(expr, default) — NULL replacement
• cast(expr AS type) — type cast
• substr(str, pos, len) — substring
• concat(str1, str2) — string concatenation
• now() — current time

---

Multi-Table Ingestion

v3.x+ supports ingesting data into multiple tables in a single HTTP request with transactional atomicity.

curl --location-trusted -u root: \
    -H "label:multi_table_load" \
    -F "table1=@/data/table1.csv;columns:c1,c2,c3" \
    -F "table2=@/data/table2.csv;columns:c1,c2" \
    http://fe_host:8030/api/mydb/_stream_load_multi_table

---

Best Practices

File Size

Label Management

• Use labels for idempotency: a given label can only be ingested once
• Recommended label format includes date and batch number: daily_load_20240101_001
• Label retention is controlled by the FE parameter label_keep_max_second (default 3 days)

Timeout Settings

• Default is 600s (10 minutes); adjust with -H "timeout:3600"
• Estimation formula: timeout ≈ file size (MB) / expected throughput (MB/s) × 2
• FE parameter stream_load_default_timeout_second controls the default value
• FE parameter max_stream_load_timeout_second controls the upper bound

Error Handling

• Set max_filter_ratio to 0 for zero tolerance (recommended for production)
• To tolerate some dirty data, set it to 0.01 (1%) or lower
• The ErrorURL in the response shows the specific error rows
• Combine with strict_mode:true so rows that fail type conversion are filtered (instead of converted to NULL)

Performance Optimization

• Per-ingestion throughput is typically 50-200 MB/s (depends on BE configuration and network)
• Launching multiple Stream Loads in parallel can boost overall throughput
• CSV is 2-5x faster than JSON; prefer CSV for large data volumes
• Compressed files (gz/lz4) reduce network transfer time but add BE decompression overhead

---

Common Issues

StarRocks Cluster Health Diagnostics

Quickly determine cluster health and pinpoint issues using a series of read-only SQL queries. All commands run through srsql (see connect.md).

Prerequisites

Depends on the srsql command. If not yet configured, run sr-login as described in connect.md.

Step 0: Identify the architecture

Required. The diagnostic path differs between shared-nothing and shared-data. SHOW WAREHOUSES exists in v3.1+; in v2.5 only BE / local-compaction paths apply.

srsql --format table -e "SHOW WAREHOUSES"

• Has results & `default_warehouse` exists -> shared-data architecture (EMR Serverless default)
• Focus on CN (Compute Nodes) + Warehouse + Cloud Native Compaction (`be_cloud_native_compactions`, v3.1+)
• information_schema.be_tablets is still populated (BE-local tablet metadata + cache info), but data files live in object storage
• No warehouse table or error -> shared-nothing architecture
• Focus on BE + local tablets + local compaction (`be_compactions`)

Step 1: FE (Frontend) nodes

Applies to both architectures:

srsql --format table -e "SHOW FRONTENDS"

How to read it:

• Every FE has Alive=true
• Exactly 1 FE has Role=LEADER (the rest are FOLLOWER or OBSERVER)
• ErrMsg non-empty -> investigate based on the message
• ReplayedJournalId differs by > 10000 across FEs -> a Follower can't keep up with the Leader; possibly a slow disk or network jitter

Step 2a: BE (shared-nothing only)

srsql --format table -e "SHOW BACKENDS"

How to read it:

• All Alive=true
• TabletNum differs > 20% between nodes -> tablet distribution skew; consider a manual rebalance
• MaxDiskUsedPct > 85% -> disk warning (any single disk hitting the threshold)
• UsedPct > 90% (overall) -> disk critical
• MemUsedPct > 90% -> memory pressure
• CpuUsedPct sustained > 80% -> CPU saturation

Step 2b: CN (v3.1+; column set varies by RunMode)

srsql --format table -e "SHOW COMPUTE NODES"

How to read it:

• All Alive=true
• WarehouseName column is present only in shared-data mode (not in shared-nothing)
• MemUsedPct, CpuUsedPct sustained high -> consider scaling up CUs or adding CN count
• NumRunningQueries abnormally high -> possibly slow queries blocking the system

Step 3: Warehouse status (shared-data only; output columns vary by version)

srsql --format table -e "SHOW WAREHOUSES"

Output column set by version (verified against `ShowWarehousesStmt`):

• v3.2: Id, Warehouse, State, ClusterCount (4 columns) — no queue/cluster-cap info; fall back to SHOW PROC '/current_queries' for queue inspection.
• v3.3+: Id, Name, State, NodeCount, CurrentClusterCount, MaxClusterCount, StartedClusters, RunningSql, QueuedSql, CreatedOn, ResumedOn, UpdatedOn, Comment

How to read it (v3.3+):

• State=AVAILABLE
• QueuedSql > 0 sustained -> insufficient concurrency; consider scaling up CUs
• CurrentClusterCount < MaxClusterCount and QueuedSql > 0 -> auto-scale not triggered, or already at the cap

Step 4: Database and tablet health overview

Applies to both architectures. This is the most important step.

srsql --format table -e "SHOW PROC '/statistic'"

Returns per-DB: TableNum / PartitionNum / IndexNum / TabletNum / ReplicaNum / UnhealthyTabletNum / InconsistentTabletNum / CloningTabletNum / ErrorStateTabletNum.

How to read it (all = 0 is best):

• UnhealthyTabletNum > 0 -> unhealthy replicas; immediately check the FE scheduling queue
• InconsistentTabletNum > 0 -> replica data inconsistency
• ErrorStateTabletNum > 0 -> tablet corruption; open a ticket

Step 5: Tablet scheduling queue

Applies to both architectures:

srsql --format table -e "
SELECT STATE, PRIORITY, COUNT(*) AS cnt
FROM information_schema.fe_tablet_schedules
GROUP BY STATE, PRIORITY
ORDER BY cnt DESC;
"

How to read it:

• Empty result, or only a small number of FINISHED, is best
• PENDING piling up > 100 sustained -> scheduling issue or insufficient BE/CN resources
• RUNNING high for a long time -> replicas are being cloned
• Many CANCELLED -> tablet replicas may be corrupted

Step 6: Compaction health

Shared-nothing:

srsql --format table -e "
SELECT BE_ID, CANDIDATES_NUM, LATEST_COMPACTION_SCORE, CANDIDATE_MAX_SCORE,
       BASE_COMPACTION_CONCURRENCY, CUMULATIVE_COMPACTION_CONCURRENCY
FROM information_schema.be_compactions
ORDER BY CANDIDATE_MAX_SCORE DESC;
"

• CANDIDATE_MAX_SCORE > 100 -> compaction can't keep up with writes; severe version backlog
• CANDIDATES_NUM sustained high -> increase compaction concurrency

Shared-data (v3.1+):

# Columns: BE_ID, TXN_ID, TABLET_ID, VERSION, SKIPPED, RUNS, START_TIME, FINISH_TIME, PROGRESS, STATUS, PROFILE
srsql --format table -e "
SELECT BE_ID, STATUS, COUNT(*) AS cnt
FROM information_schema.be_cloud_native_compactions
GROUP BY BE_ID, STATUS
ORDER BY cnt DESC;
"

• Any STATUS='FAILED' records -> drill in with SELECT TABLET_ID, RUNS, PROFILE FROM ... WHERE STATUS='FAILED'
• PROGRESS stuck low + RUNS increasing -> compaction repeatedly retrying

Step 7: Resource group status

Applies to both architectures:

srsql --format table -e "SHOW RESOURCE GROUPS ALL"

How to read it:

• Default default_wg and default_mv_wg exist
• Check classifiers to see business resource groups (user-defined)

Step 8: Recent ingestion job status

Applies to both architectures (ingestion is a common source of issues):

srsql --format table -e "
SELECT STATE, COUNT(*) AS cnt
FROM information_schema.loads
WHERE CREATE_TIME > DATE_SUB(NOW(), INTERVAL 1 DAY)
GROUP BY STATE
ORDER BY cnt DESC;
"

Details of failed jobs in the last 24 hours:

srsql --format table -e "
SELECT ID, LABEL, DB_NAME, TABLE_NAME, TYPE, STATE,
       CREATE_TIME, LOAD_FINISH_TIME, ERROR_MSG
FROM information_schema.loads
WHERE STATE IN ('CANCELLED', 'FAILED')
  AND CREATE_TIME > DATE_SUB(NOW(), INTERVAL 1 DAY)
ORDER BY CREATE_TIME DESC
LIMIT 20;
"

Synthesis template

After diagnostics are complete, summarize for the user using this template:

StarRocks Cluster Diagnostic Report
===================================

Architecture: [shared-nothing | shared-data]
Overall status: [healthy | warning | critical]

Nodes:
  FE: {alive}/{total} healthy, Leader={ip}
  BE: {alive}/{total} healthy (shared-nothing)
  CN: {alive}/{total} healthy, warehouse={default_warehouse} state=AVAILABLE (shared-data)

Data:
  DB count={db_count}, table count={table_count}, tablet count={tablet_count}
  Unhealthy tablets: {unhealthy}
  Scheduling queue: PENDING={pending}, RUNNING={running}

Resources:
  Max disk usage: {max_disk}%
  Max memory usage: {max_mem}%
  Max CPU usage: {max_cpu}%

Last 24 hours:
  Ingestion jobs: {finished} succeeded / {failed} failed
  Highest compaction score: {max_score}

Key issues:
  - ... (list anomalies)

Recommendations:
  - ... (actionable suggestions)

Severity quick reference

Follow-up diagnostic commands

Once you've identified a class of issue, you can drill deeper:

Out of scope

• Don't modify any configuration (this topic is read-only)
• Don't restart nodes (this is a control-plane operation; handle it via the EMR Serverless console or the corresponding OpenAPI)
• Don't kill queries (requires write privilege; outside this skill's scope)
• Don't export log files (no filesystem access)

RAM Permissions

This skill requires no Alibaba Cloud RAM permissions.

All cluster access goes through the bundled srsql CLI, which speaks the MySQL wire protocol (port 9030) to the StarRocks FE using a StarRocks-internal account (FE user/password). The skill does not call any Alibaba Cloud OpenAPI, does not assume a RAM role, and does not read aliyun configure / AccessKey credentials.

Privileges are evaluated entirely inside StarRocks via SHOW GRANTS FOR CURRENT_USER(). See connect.md for the auth and security model.

StarRocks Table Creation Best Practices

Required Information Checklist

Before giving recommendations, collect the following information (listed by priority). When something is missing, ask proactively:

Table Creation Decision Flow

Step 1: Choose table model     → Based on update requirements (see schema/table-types.md)
    ↓
Step 2: Design partitioning    → Based on data volume and time dimension (see schema/partitioning.md)
    ↓
Step 3: Design bucketing       → Based on query pattern and data volume (see schema/bucketing.md)
    ↓
Step 4: Design sort key        → Based on high-frequency query filter columns (see schema/sort-key-and-indexes.md)
    ↓
Step 5: Choose indexes         → Based on non-prefix column query requirements (see schema/sort-key-and-indexes.md)
    ↓
Step 6: Set storage properties → Based on performance/space trade-offs (see schema/storage-properties.md)
        On shared-data (存算分离), see the "Shared-data properties" section there for the
        required checklist (storage_volume, datacache.*, persistent_index_type) and the
        properties that are silently stripped or rejected.

Detailed rules for each step are in the corresponding reference file; read as needed.

Table Model Quick Reference

Partition and Bucketing Quick Reference

Common Anti-Patterns

Proactively check for and avoid these in your table design:

Reference File Index

When you need the detailed rules for a step, read the corresponding file:

Output Template

When giving table creation recommendations, use the following structured format:

## Table Design

### 1. Table Model: {Duplicate Key / Aggregate / Primary Key / Unique Key}
**Rationale:** {Why this model fits the user's scenario}

### 2. Partitioning Strategy
**Design:** {PARTITION BY ...}
**Rationale:** {Data volume, retention period, pruning benefit}
**Estimated partition count:** {Approximately N partitions}

### 3. Bucketing Strategy
**Design:** {DISTRIBUTED BY HASH(col) BUCKETS N / DISTRIBUTED BY RANDOM}
**Bucket count:** {N} (about {X} GB per tablet)
**Rationale:** {Why this bucket key and count}

### 4. Sort Key and Indexes
**Sort key:** ORDER BY (col1, col2, ...)
**Additional indexes:** {Bitmap / Bloom Filter / None}
**Rationale:** {Which high-frequency query filter conditions this matches}

### 5. Storage Properties
- Compression: {LZ4 / ZSTD / ...}
- Replicas: {N}
- Other: {As needed}

### 6. Full CREATE TABLE Statement

​```sql
CREATE TABLE ...
​```

### 7. Caveats
- {Version requirements, migration notes, follow-up optimization suggestions, etc.}