risingwavelabs/risingwave · error · SinkError::Config

If you want to use upsert, please set the keysType of…

Error message

If you want to use upsert, please set the keysType of starrocks to PRIMARY_KEY

What it means

Thrown during sink validation when the sink is upsert (non-append-only) but the target StarRocks table's `table_model` (from `information_schema.tables_config`) is not `PRIMARY_KEYS`. Upsert semantics require the StarRocks table to use the PRIMARY_KEY keys type so incoming rows replace existing rows.

Solutions

Alter/recreate the StarRocks table with `PRIMARY KEY (...) DISTRIBUTED BY ...` so its model is PRIMARY_KEYS
Change the RisingWave sink to append-only if you do not need upsert semantics
Point the sink at a different StarRocks table that uses PRIMARY_KEYS

Example fix

// before (StarRocks DDL)
CREATE TABLE t (id INT, v INT) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id);
// after
CREATE TABLE t (id INT, v INT) PRIMARY KEY(id) DISTRIBUTED BY HASH(id);

Defensive patterns

Strategy: validation

Validate before calling

// Before creating the sink, check the target model:
SELECT table_model FROM information_schema.tables_config
WHERE table_name = 't' AND table_schema = 'db';
-- must return PRIMARY_KEYS for upsert sinks

Prevention

Create StarRocks tables with PRIMARY KEY(...) when planning upsert sinks
Re-check the table model after any StarRocks-side schema changes

When it happens

Trigger: Sink validated against an existing StarRocks table whose keysType is DUPLICATE_KEYS, AGG_KEYS, or UNIQUE_KEYS while the RW sink is created with a primary key / non-append-only data.

Common situations: Pointing a RisingWave upsert sink at a pre-existing StarRocks detail or aggregate table; the StarRocks table was created without `PRIMARY KEY(...)` / `PRIMARY KEYS(...)` in its DDL.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/f73359a9da801c24. Report an issue: GitHub.

Appendix: source

Thrown at src/connector/src/sink/starrocks.rs:393

        if !self.is_append_only && self.pk_indices.is_empty() {
            return Err(SinkError::Config(anyhow!(
                "Primary key not defined for upsert starrocks sink (please define in `primary_key` field)"
            )));
        }
        // check reachability
        let mut client = StarrocksSchemaClient::new(
            self.config.common.host.clone(),
            self.config.common.mysql_port.clone(),
            self.config.common.table.clone(),
            self.config.common.database.clone(),
            self.config.common.user.clone(),
            self.config.common.password.clone(),
        )
        .await?;
        let (read_model, pks) = client.get_pk_from_starrocks().await?;

        if !self.is_append_only && read_model.ne("PRIMARY_KEYS") {
            return Err(SinkError::Config(anyhow!(
                "If you want to use upsert, please set the keysType of starrocks to PRIMARY_KEY"
            )));
        }

        for (index, filed) in self.schema.fields().iter().enumerate() {
            if self.pk_indices.contains(&index) && !pks.contains(&filed.name) {
                return Err(SinkError::Starrocks(format!(
                    "Can't find pk {:?} in starrocks",
                    filed.name
                )));
            }
        }

        let starrocks_columns_desc = client.get_columns_from_starrocks().await?;

        self.check_column_name_and_type(starrocks_columns_desc)?;
        Ok(())
    }

View on GitHub (pinned to 6469eb736d)