risingwavelabs/risingwave · error · SinkError

Primary key not defined for upsert doris sink (please define

Error message

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

What it means

The Doris sink in upsert mode requires a primary key so updates/deletes can be applied row-by-row on the Doris side. `validate()` raises this error when the sink is not append-only and `pk_indices` is empty, meaning no `primary_key` was defined.

Source

Thrown at src/connector/src/sink/doris.rs:274

    const SINK_NAME: &'static str = DORIS_SINK;

    crate::impl_validate_sink_unknown_fields!();

    async fn new_log_sinker(&self, writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
        Ok(DorisSinkWriter::new(
            self.config.clone(),
            self.schema.clone(),
            self.pk_indices.clone(),
            self.is_append_only,
        )
        .await?
        .into_log_sinker(SinkWriterMetrics::new(&writer_param)))
    }

    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 doris sink (please define in `primary_key` field)"
            )));
        }
        // check reachability
        let client = self.config.common.build_get_client();
        let doris_schema = client.get_schema_from_doris().await?;

        if !self.is_append_only && doris_schema.keys_type.ne("UNIQUE_KEYS") {
            return Err(SinkError::Config(anyhow!(
                "If you want to use upsert, please set the keysType of doris to UNIQUE_KEYS"
            )));
        }
        self.check_column_name_and_type(doris_schema.properties)?;
        Ok(())
    }
}

pub struct DorisSinkWriter {

View on GitHub (pinned to 6469eb736d)

Solutions

Add a `primary_key` option to the sink WITH clause listing the key columns.
If the data is truly append-only, switch the sink `type` to `append-only`.
Ensure the upstream materialized view defines a primary key so pk_indices is non-empty.

Example fix

// before
CREATE SINK s FROM mv WITH ('connector'='doris', 'type'='upsert');
// after
CREATE SINK s FROM mv WITH ('connector'='doris', 'type'='upsert', 'primary_key'='id');

Defensive patterns

Strategy: validation

Validate before calling

if sink_type == "upsert" && with_options.get("primary_key").is_none() {
    return Err("upsert doris sink requires a `primary_key` option");
}

Try / catch

if let Err(SinkError::Config(e)) = sink.validate().await {
    if e.to_string().contains("Primary key not defined") {
        eprintln!("add `primary_key` to the sink WITH clause");
    }
}

Prevention

Always define `primary_key` when sink type is upsert.
Ensure upstream MVs/sources declare primary keys.
Use append-only type when no key semantics are needed.

When it happens

Trigger: Creating a Doris sink with `type='upsert'` (or on a non-append-only stream) while omitting the `primary_key` option in the WITH clause or the underlying source having no PK.

Common situations: Upserting from a source table without a primary key; forgetting `primary_key` in the sink definition; relying on implicit keys that RisingWave does not propagate to the sink.

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

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