{"record":{"id":"f4e9caddc76c8e1e","repo":"risingwavelabs/risingwave","slug":"key-encode-expects-only-one-primary-key-but-go","errorCode":null,"errorMessage":"KEY ENCODE {} expects only one primary key, but got {}","messagePattern":"KEY ENCODE (.+?) expects only one primary key, but got (.+?)","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/formatter/mod.rs","lineNumber":199,"sourceCode":"            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        ))\n    })?;\n\n    Ok((pk_indices[0], schema_ref))\n}\n\nimpl EncoderBuild for BytesEncoder {","sourceCodeStart":181,"sourceCodeEnd":217,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/formatter/mod.rs#L181-L217","documentation":"Key encoders in this formatter support encoding only a single-column primary key. The builder throws when pk_indices contains more than one index, because composite primary keys cannot be represented by the selected key encode format.","triggerScenarios":"build() called with pk_indices.len() > 1 — e.g. a sink with KEY ENCODE BYTES/TEXT on a relation with a composite PRIMARY KEY (a, b).","commonSituations":"Sinking a table declared with a multi-column primary key; joining MVs and inheriting multiple key columns; expecting the encoder to concatenate key columns (it will not).","solutions":["Restructure so the key is a single column (e.g. concat columns into one varchar key)","Use a key encode format that supports composite keys, or add a surrogate single-column PK","Select only one key column in the sink definition"],"exampleFix":"// before\nCREATE TABLE t (a INT, b INT, PRIMARY KEY (a, b));\nCREATE SINK s FROM t WITH (key_encode = 'text');\n// after\nCREATE SINK s AS SELECT concat(a::varchar, ':', b::varchar) AS k, * FROM t WITH (key_encode = 'text');","handlingStrategy":"validation","validationCode":"if let Some(pk) = &sink_pk_indices, pk.len() != 1 {\n    return Err(format!(\"key encode requires exactly 1 pk column, got {}\", pk.len()));\n}","typeGuard":"fn is_single_column_pk(pk: &Option<Vec<usize>>) -> bool { matches!(pk, Some(v) if v.len() == 1) }","tryCatchPattern":null,"preventionTips":["Avoid composite primary keys when the sink uses single-column key encodes","Build a surrogate single-column key column in the sink query"],"tags":["sink","primary-key","composite-key"],"backgroundTag":"invalid-argument-value","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}