risingwavelabs/risingwave · error · SinkError

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

Error message

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

What it means

When the Doris sink runs in upsert mode, the target Doris table must use the UNIQUE_KEYS keysType so Doris can merge rows by key. `validate()` queries the Doris FE schema and fails if `keysType` is anything else (e.g. DUPLICATE_KEYS or AGGREGATE_KEYS).

Source

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

            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 {
    pub config: DorisConfig,
    #[expect(dead_code)]
    schema: Schema,
    #[expect(dead_code)]
    pk_indices: Vec<usize>,
    inserter_inner_builder: InserterInnerBuilder,
    is_append_only: bool,
    client: Option<DorisClient>,
    row_encoder: JsonEncoder,

View on GitHub (pinned to 6469eb736d)

Solutions

Recreate the Doris table with `DUPLICATE KEY` replaced by `UNIQUE KEY(...)` in the DDL.
Verify with `SHOW CREATE TABLE` that keysType is UNIQUE_KEYS before creating the sink.
Alternatively keep the table as-is and define the sink as `type='append-only'`.

Example fix

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

Defensive patterns

Strategy: validation

Validate before calling

// run before creating the sink
// SHOW CREATE TABLE doris_table;  -> must contain UNIQUE KEY(...)

Try / catch

match sink.validate().await {
    Err(SinkError::Config(e)) if e.to_string().contains("UNIQUE_KEYS") => {
        eprintln!("recreate the Doris table with UNIQUE KEY");
    }
    other => other?,
}

Prevention

Create Doris tables with UNIQUE KEY when they will receive upserts.
Run SHOW CREATE TABLE to verify keysType before wiring a sink.
Keep one table per write mode; don't reuse append-only tables for upsert.

When it happens

Trigger: Creating a Doris upsert sink against a Doris table whose `keysType` property is not `UNIQUE_KEYS`, detected during sink validation via get_schema_from_doris.

Common situations: Pointing the sink at a pre-existing Doris table created with default DUPLICATE_KEYS; table created for an earlier append-only sink then reused for upsert.

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/91f9d904192a7d7f. Report an issue: GitHub.