{"record":{"id":"5578cafa203ee19f","repo":"risingwavelabs/risingwave","slug":"the-key-encode-is-bytes-but-the-primary-key-colum","errorCode":null,"errorMessage":"The key encode is BYTES, but the primary key column {} has type {}","messagePattern":"The key encode is BYTES, but the primary key column (.+?) has type (.+?)","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/formatter/mod.rs","lineNumber":226,"sourceCode":"            \"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 {\n    async fn build(params: EncoderParams<'_>, pk_indices: Option<Vec<usize>>) -> Result<Self> {\n        match pk_indices {\n            // This is being used as a key encoder\n            Some(_) => {\n                let (pk_index, schema_ref) = ensure_only_one_pk(\"BYTES\", &params, &pk_indices)?;\n                if let DataType::Bytea = schema_ref.data_type() {\n                    Ok(BytesEncoder::new(params.schema, pk_index))\n                } else {\n                    Err(SinkError::Config(anyhow!(\n                        \"The key encode is BYTES, but the primary key column {} has type {}\",\n                        schema_ref.name,\n                        schema_ref.data_type\n                    )))\n                }\n            }\n            // This is being used as a value encoder\n            None => {\n                // Ensure the schema has exactly one column and it's of type BYTEA\n                if params.schema.len() != 1 {\n                    return Err(SinkError::Config(anyhow!(\n                        \"ENCODE BYTES requires exactly one column, got {} columns\",\n                        params.schema.len()\n                    )));\n                }\n\n                let field = &params.schema.fields[0];\n                if let DataType::Bytea = field.data_type {","sourceCodeStart":208,"sourceCodeEnd":244,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/formatter/mod.rs#L208-L244","documentation":"When key encode is BYTES, the encoder maps the raw primary key bytes directly, so the key column must be of type BYTEA. The builder throws when the resolved PK column has any other type, since it refuses to reinterpret non-byte types as raw bytes.","triggerScenarios":"KEY ENCODE BYTES configured and ensure_only_one_pk succeeds, but schema_ref.data_type() is not DataType::Bytea (e.g. INT or VARCHAR primary key).","commonSituations":"Setting key_encode='bytes' out of habit on an INT-keyed table; migrating a Kafka sink config from another system where 'bytes' meant arbitrary serialization.","solutions":["Change the key column to BYTEA (cast in the sink query)","Use key_encode='text' for varchar/int/bool keys","Use a different key encode format that supports the column's type"],"exampleFix":"// before\nCREATE SINK s FROM t WITH (key_encode = 'bytes'); -- t.id is INT\n// after\nCREATE SINK s AS SELECT id::bytea AS id, * FROM t WITH (key_encode = 'bytes');","handlingStrategy":"validation","validationCode":"if key_encode == \"bytes\" && pk_field.data_type != DataType::Bytea {\n    return Err(\"key encode bytes requires a bytea pk column\");\n}","typeGuard":"fn is_bytes_key(dt: &DataType) -> bool { *dt == DataType::Bytea }","tryCatchPattern":null,"preventionTips":["Match key encode format to the pk column type","Prefer 'text' for varchar/int keys"],"tags":["sink","type-mismatch","bytes","primary-key"],"backgroundTag":"type-mismatch","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"}