risingwavelabs/risingwave · error · SinkError::Config

Redis Sink Primary Key must be specified.

Error message

Redis Sink Primary Key must be specified.

What it means

RedisSink::try_from requires the downstream primary key because Redis uses it to compute the key under which each row is written. If SinkParam.downstream_pk is None, construction fails with SinkError::Config 'Redis Sink Primary Key must be specified.' The library throws it so the sink is never created without deterministic keying semantics.

Source

Thrown at src/connector/src/sink/redis.rs:287

    sink_from_name: String,
}

impl EnforceSecret for RedisSink {
    fn enforce_secret<'a>(prop_iter: impl Iterator<Item = &'a str>) -> ConnectorResult<()> {
        for prop in prop_iter {
            RedisConfig::enforce_one(prop)?;
        }
        Ok(())
    }
}

#[async_trait]
impl TryFrom<SinkParam> for RedisSink {
    type Error = SinkError;

    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {
        let Some(pk_indices) = param.downstream_pk.clone() else {
            return Err(SinkError::Config(anyhow!(
                "Redis Sink Primary Key must be specified."
            )));
        };
        let config = RedisConfig::from_btreemap(param.properties.clone())?;
        Ok(Self {
            config,
            schema: param.schema(),
            pk_indices,
            format_desc: param
                .format_desc
                .ok_or_else(|| SinkError::Config(anyhow!("missing FORMAT ... ENCODE ...")))?,
            db_name: param.db_name,
            sink_from_name: param.sink_from_name,
        })
    }
}

impl Sink for RedisSink {

View on GitHub (pinned to 6469eb736d)

Solutions

1. Add `primary_key = '<column(s)>'` to the CREATE SINK WITH options.
2. Ensure the primary key columns exist in the sink's output schema.
3. Re-create the sink — sink options cannot be altered after creation.
4. If the data truly has no key, pick a unique column (or composite) to serve as the Redis key.

Example fix

// before
CREATE SINK s FROM mv WITH (connector='redis', url='...') FORMAT APPEND ONLY ENCODE JSON;
// after
CREATE SINK s FROM mv WITH (connector='redis', url='...', primary_key='user_id') FORMAT APPEND ONLY ENCODE JSON;

// Require a primary_key option for redis sinks before DDL submission
fn requires_pk(props: &BTreeMap<String, String>) -> Result<(), String> {
    match props.get("primary_key") {
        Some(pk) if !pk.trim().is_empty() => Ok(()),
        _ => Err("redis sink requires primary_key in WITH options".into()),
    }
}

fn pk_available(param: &SinkParam) -> bool {
    param.downstream_pk.as_ref().map_or(false, |pk| !pk.is_empty())
}

match RedisSink::try_from(param) {
    Err(SinkError::Config(e)) if e.to_string().contains("Primary Key") => {
        eprintln!("re-create the sink with primary_key in WITH options");
    }
    other => { /* proceed */ }
}

When it happens

Trigger: Converting a SinkParam into RedisSink where param.downstream_pk is None — i.e. the CREATE SINK statement omitted `primary_key` in WITH options (or the source has no PK and none was declared).

Common situations: Sink created from an append-only source without explicit primary_key option; forgetting primary_key while intending upsert-style writes; DDL templates that omit primary_key.

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 {:?} pulsar sink (please define
• Primary key not defined for upsert doris sink (please define
• DynamoDB table {} primary key {:?} must match RisingWave pri
• no value find in sink schema, index is {:?}
• please set the separator in the with option, when there are

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