risingwavelabs/risingwave · error · SinkError

{}Encoder requires primary key columns to be specified

Error message

{}Encoder requires primary key columns to be specified

What it means

A sink formatter encoder that encodes the key (e.g. BytesEncoder) requires exactly one primary key column to derive the key from. The builder throws this when pk_indices is None, meaning no primary key was resolvable for the sink. Without a PK column there is no meaningful key to encode.

Source

Thrown at src/connector/src/sink/formatter/mod.rs:193

        let (descriptor, sid) =
            crate::schema::protobuf::fetch_descriptor(&b.format_desc.options, b.topic, None)
                .await
                .map_err(|e| SinkError::Config(anyhow!(e)))?;
        let header = match sid {
            None => ProtoHeader::None,
            Some(sid) => ProtoHeader::ConfluentSchemaRegistry(sid),
        };
        ProtoEncoder::new(b.schema, None, descriptor, header)
    }
}

fn ensure_only_one_pk<'a>(
    data_type_name: &'a str,
    params: &'a EncoderParams<'_>,
    pk_indices: &'a Option<Vec<usize>>,
) -> Result<(usize, &'a Field)> {
    let Some(pk_indices) = pk_indices else {
        return Err(SinkError::Config(anyhow!(
            "{}Encoder requires primary key columns to be specified",
            data_type_name
        )));
    };
    if pk_indices.len() != 1 {
        return Err(SinkError::Config(anyhow!(
            "KEY ENCODE {} expects only one primary key, but got {}",
            data_type_name,
            pk_indices.len(),
        )));
    }

    let schema_ref = params.schema.fields().get(pk_indices[0]).ok_or_else(|| {
        SinkError::Config(anyhow!(
            "The primary key column index {} is out of bounds in schema {:?}",
            pk_indices[0],
            params.schema
        ))

View on GitHub (pinned to 6469eb736d)

Solutions

1. Define a PRIMARY KEY on the table/MV the sink reads from
2. Add a key column via CREATE SINK ... AS SELECT so a PK index exists
3. If the encoder is meant as a value encoder only, do not configure KEY ENCODE (pass None key encoding)

Example fix

// before: sink on unkeyed relation
CREATE SINK s FROM mv_no_pk WITH (key_encode = 'bytes');
// after
CREATE TABLE t (id INT PRIMARY KEY, ...);
CREATE SINK s FROM t WITH (key_encode = 'bytes');

if sink_pk_indices.is_none() {
    return Err("key encode requires a relation with a PRIMARY KEY");
}

fn has_single_pk(pk: &Option<Vec<usize>>) -> bool { matches!(pk, Some(v) if !v.is_empty()) }

When it happens

Trigger: Calling EncoderBuild::build for a key encoder (key encode = BYTES or similar) with pk_indices = None — e.g. a CREATE SINK with KEY ENCODE BYTES on a materialized view/table that has no PRIMARY KEY defined.

Common situations: Creating a sink from an unkeyed source or MV; forgetting the PRIMARY KEY clause; upstream refactor removed pk column info before build.

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

• Primary key not defined for upsert bigquery sink (please def
• Unknown sink connector: {sink_name}
• Primary key not defined for upsert clickhouse sink (please d
• should have meta client
• Primary key not defined for upsert doris sink (please define

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