{"record":{"id":"6ebde24530c1a92f","repo":"risingwavelabs/risingwave","slug":"redis-sink-primary-key-must-be-specified","errorCode":null,"errorMessage":"Redis Sink Primary Key must be specified.","messagePattern":"Redis Sink Primary Key must be specified\\.","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/redis.rs","lineNumber":287,"sourceCode":"    sink_from_name: String,\n}\n\nimpl EnforceSecret for RedisSink {\n    fn enforce_secret<'a>(prop_iter: impl Iterator<Item = &'a str>) -> ConnectorResult<()> {\n        for prop in prop_iter {\n            RedisConfig::enforce_one(prop)?;\n        }\n        Ok(())\n    }\n}\n\n#[async_trait]\nimpl TryFrom<SinkParam> for RedisSink {\n    type Error = SinkError;\n\n    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {\n        let Some(pk_indices) = param.downstream_pk.clone() else {\n            return Err(SinkError::Config(anyhow!(\n                \"Redis Sink Primary Key must be specified.\"\n            )));\n        };\n        let config = RedisConfig::from_btreemap(param.properties.clone())?;\n        Ok(Self {\n            config,\n            schema: param.schema(),\n            pk_indices,\n            format_desc: param\n                .format_desc\n                .ok_or_else(|| SinkError::Config(anyhow!(\"missing FORMAT ... ENCODE ...\")))?,\n            db_name: param.db_name,\n            sink_from_name: param.sink_from_name,\n        })\n    }\n}\n\nimpl Sink for RedisSink {","sourceCodeStart":269,"sourceCodeEnd":305,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/redis.rs#L269-L305","documentation":"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.","triggerScenarios":"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).","commonSituations":"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.","solutions":["Add `primary_key = '<column(s)>'` to the CREATE SINK WITH options.","Ensure the primary key columns exist in the sink's output schema.","Re-create the sink — sink options cannot be altered after creation.","If the data truly has no key, pick a unique column (or composite) to serve as the Redis key."],"exampleFix":"// before\nCREATE SINK s FROM mv WITH (connector='redis', url='...') FORMAT APPEND ONLY ENCODE JSON;\n// after\nCREATE SINK s FROM mv WITH (connector='redis', url='...', primary_key='user_id') FORMAT APPEND ONLY ENCODE JSON;","handlingStrategy":"validation","validationCode":"// Require a primary_key option for redis sinks before DDL submission\nfn requires_pk(props: &BTreeMap<String, String>) -> Result<(), String> {\n    match props.get(\"primary_key\") {\n        Some(pk) if !pk.trim().is_empty() => Ok(()),\n        _ => Err(\"redis sink requires primary_key in WITH options\".into()),\n    }\n}","typeGuard":"fn pk_available(param: &SinkParam) -> bool {\n    param.downstream_pk.as_ref().map_or(false, |pk| !pk.is_empty())\n}","tryCatchPattern":"match RedisSink::try_from(param) {\n    Err(SinkError::Config(e)) if e.to_string().contains(\"Primary Key\") => {\n        eprintln!(\"re-create the sink with primary_key in WITH options\");\n    }\n    other => { /* proceed */ }\n}","preventionTips":["Always set primary_key for redis sinks","Use the MV's primary key columns for primary_key","Re-create the sink after changing key options","Pick a unique column when the source has no PK"],"tags":["rust","redis","sink","primary-key","validation"],"backgroundTag":"missing-required-config-field","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}