risingwavelabs/risingwave · error · SinkError::Config

primary key must be specified for upsert iceberg sink

Error message

primary key must be specified for upsert iceberg sink

What it means

An upsert-type iceberg sink must know the downstream primary key to perform key-based upserts. If no primary key is available (param.downstream_pk is None or empty) and force_append_only is not set, sink construction fails.

Source

Thrown at src/connector/src/sink/iceberg/mod.rs:177

    pub fn new(config: IcebergConfig, param: SinkParam) -> Result<Self> {
        if let Some(order_key) = &config.order_key {
            validate_order_key_columns(
                order_key,
                param.columns.iter().map(|column| column.name.as_str()),
            )
            .context("invalid order_key")
            .map_err(SinkError::Config)?;
        }

        let upsert_primary_key_column_names =
            if config.r#type == SINK_TYPE_UPSERT && !config.force_append_only {
                let pk_indices = param
                    .downstream_pk
                    .as_ref()
                    .filter(|pk| !pk.is_empty())
                    .ok_or_else(|| {
                        SinkError::Config(anyhow!(
                            "primary key must be specified for upsert iceberg sink"
                        ))
                    })?;
                Some(
                    pk_indices
                        .iter()
                        .map(|&idx| {
                            param
                                .columns
                                .get(idx)
                                .map(|column| column.name.clone())
                                .ok_or_else(|| {
                                    SinkError::Config(anyhow!(
                                        "primary key column index {} out of range in sink schema",
                                        idx
                                    ))
                                })
                        })

View on GitHub (pinned to 6469eb736d)

Solutions

1. Add a `primary_key` option to the sink listing key column(s)
2. Or define a primary key on the source materialized view
3. Or use type='append-only' / set force_append_only=true if upsert semantics are not needed

Example fix

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

if sink_type == "upsert" && !force_append_only && (primary_key.is_none() || primary_key.as_deref() == Some("")) {
    return Err("upsert iceberg sink requires primary_key");
}

When it happens

Trigger: CREATE SINK with type='upsert' (or an upsert-capable default) from a source without a primary key, without specifying a `primary_key` option and without force_append_only=true.

Common situations: Sink on an append-only MV without PK; forgetting the primary_key option; upgrading from append-only behavior where upsert is now the inferred type.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

• VARIANT column `{}` cannot be used as the primary key of an
• Primary key not defined for upsert clickhouse sink (please d
• `primary-key` must not be empty in {}
• Must set `primary-key` in {}
• primary key column index {} out of range in sink schema

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