risingwavelabs/risingwave · error · SinkError::Config

Primary key not defined for upsert starrocks sink (please…

Error message

Primary key not defined for upsert starrocks sink (please define in `primary_key` field)

What it means

Thrown from the sink `validate` hook when a StarRocks sink is declared as non-append-only (upsert mode) but `pk_indices` is empty, meaning no primary key columns were defined. Upsert requires a primary key to know which rows to overwrite in StarRocks.

Solutions

1. Add `'primary_key' = 'col1,col2'` to the sink WITH options
2. Declare the sink with an explicit `primary_key` clause matching the target StarRocks table's key columns
3. If data is truly append-only, create the sink in append-only mode so the PK requirement is skipped

Example fix

// before
CREATE SINK s FROM mv INTO starrocks WITH (
  'connector' = 'starrocks'
);
// after
CREATE SINK s FROM mv INTO starrocks WITH (
  'connector' = 'starrocks',
  'primary_key' = 'order_id'
);

// Ensure PK is declared for non-append-only sources
if !source_is_append_only && !opts.contains_key("primary_key") {
    throw "upsert StarRocks sink requires 'primary_key' option";
}

When it happens

Trigger: `CREATE SINK ... INTO starrocks` where the upstream materialized view is not append-only (contains updates/deletes) and no `primary_key` field is given in the WITH options or the sink definition.

Common situations: Sinking an MV with aggregations or updates into StarRocks upsert-style but forgetting the `primary_key` option; users assuming the primary key is inferred from the source table.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• Primary key not defined for upsert mongodb sink (please…
• Primary key not defined for upsert SQL Server sink (please…
• collection.name.field
• `commit_checkpoint_interval` must be greater than 0
• {e}

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

Appendix: source

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

                "MAP is not supported for Starrocks sink.".to_owned(),
            )),
            DataType::Vector(_) => Err(SinkError::Starrocks(
                "VECTOR is not supported for Starrocks sink.".to_owned(),
            )),
        }
    }
}

impl Sink for StarrocksSink {
    type LogSinker = DecoupleCheckpointLogSinkerOf<StarrocksSinkWriter>;

    const SINK_NAME: &'static str = STARROCKS_SINK;

    crate::impl_validate_sink_unknown_fields!();

    async fn validate(&self) -> Result<()> {
        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"

View on GitHub (pinned to 6469eb736d)