{"record":{"id":"c3e03f7d0813b763","repo":"risingwavelabs/risingwave","slug":"encoder-requires-primary-key-columns-to-be-speci","errorCode":null,"errorMessage":"{}Encoder requires primary key columns to be specified","messagePattern":"(.+?)Encoder requires primary key columns to be specified","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/formatter/mod.rs","lineNumber":193,"sourceCode":"        let (descriptor, sid) =\n            crate::schema::protobuf::fetch_descriptor(&b.format_desc.options, b.topic, None)\n                .await\n                .map_err(|e| SinkError::Config(anyhow!(e)))?;\n        let header = match sid {\n            None => ProtoHeader::None,\n            Some(sid) => ProtoHeader::ConfluentSchemaRegistry(sid),\n        };\n        ProtoEncoder::new(b.schema, None, descriptor, header)\n    }\n}\n\nfn ensure_only_one_pk<'a>(\n    data_type_name: &'a str,\n    params: &'a EncoderParams<'_>,\n    pk_indices: &'a Option<Vec<usize>>,\n) -> Result<(usize, &'a Field)> {\n    let Some(pk_indices) = pk_indices else {\n        return Err(SinkError::Config(anyhow!(\n            \"{}Encoder requires primary key columns to be specified\",\n            data_type_name\n        )));\n    };\n    if pk_indices.len() != 1 {\n        return Err(SinkError::Config(anyhow!(\n            \"KEY ENCODE {} expects only one primary key, but got {}\",\n            data_type_name,\n            pk_indices.len(),\n        )));\n    }\n\n    let schema_ref = params.schema.fields().get(pk_indices[0]).ok_or_else(|| {\n        SinkError::Config(anyhow!(\n            \"The primary key column index {} is out of bounds in schema {:?}\",\n            pk_indices[0],\n            params.schema\n        ))","sourceCodeStart":175,"sourceCodeEnd":211,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/formatter/mod.rs#L175-L211","documentation":"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.","triggerScenarios":"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.","commonSituations":"Creating a sink from an unkeyed source or MV; forgetting the PRIMARY KEY clause; upstream refactor removed pk column info before build.","solutions":["Define a PRIMARY KEY on the table/MV the sink reads from","Add a key column via CREATE SINK ... AS SELECT so a PK index exists","If the encoder is meant as a value encoder only, do not configure KEY ENCODE (pass None key encoding)"],"exampleFix":"// before: sink on unkeyed relation\nCREATE SINK s FROM mv_no_pk WITH (key_encode = 'bytes');\n// after\nCREATE TABLE t (id INT PRIMARY KEY, ...);\nCREATE SINK s FROM t WITH (key_encode = 'bytes');","handlingStrategy":"validation","validationCode":"if sink_pk_indices.is_none() {\n    return Err(\"key encode requires a relation with a PRIMARY KEY\");\n}","typeGuard":"fn has_single_pk(pk: &Option<Vec<usize>>) -> bool { matches!(pk, Some(v) if !v.is_empty()) }","tryCatchPattern":null,"preventionTips":["Always declare PRIMARY KEY on tables/MVs used as sink sources","Check the relation's key before choosing a key encode format"],"tags":["sink","primary-key","configuration"],"backgroundTag":"missing-required-argument","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"}