{"record":{"id":"d1553cdaf8cd55b0","repo":"risingwavelabs/risingwave","slug":"the-key-encode-is-text-but-the-primary-key-column","errorCode":null,"errorMessage":"The key encode is TEXT, but the primary key column {} has type {}. The key encode TEXT requires the primary key column to be of type varchar, bool, small int, int, big int, serial or rw_int256.","messagePattern":"The key encode is TEXT, but the primary key column (.+?) has type (.+?)\\. The key encode TEXT requires the primary key column to be of type varchar, bool, small int, int, big int, serial or rw_int256\\.","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/formatter/mod.rs","lineNumber":270,"sourceCode":"            }\n        }\n    }\n}\n\nimpl EncoderBuild for TextEncoder {\n    async fn build(params: EncoderParams<'_>, pk_indices: Option<Vec<usize>>) -> Result<Self> {\n        let (pk_index, schema_ref) = ensure_only_one_pk(\"TEXT\", &params, &pk_indices)?;\n        match &schema_ref.data_type() {\n            DataType::Varchar\n            | DataType::Boolean\n            | DataType::Int16\n            | DataType::Int32\n            | DataType::Int64\n            | DataType::Int256\n            | DataType::Serial => {}\n            _ => {\n                // why we don't allow float as text for key encode: https://github.com/risingwavelabs/risingwave/pull/16377#discussion_r1591864960\n                return Err(SinkError::Config(anyhow!(\n                    \"The key encode is TEXT, but the primary key column {} has type {}. The key encode TEXT requires the primary key column to be of type varchar, bool, small int, int, big int, serial or rw_int256.\",\n                    schema_ref.name,\n                    schema_ref.data_type\n                )));\n            }\n        }\n\n        Ok(Self::new(params.schema, pk_index))\n    }\n}\n\nimpl EncoderBuild for AvroEncoder {\n    async fn build(b: EncoderParams<'_>, pk_indices: Option<Vec<usize>>) -> Result<Self> {\n        use crate::schema::{SchemaLoader, SchemaVersion};\n\n        let loader = SchemaLoader::from_format_options(b.topic, &b.format_desc.options)\n            .await\n            .map_err(|e| SinkError::Config(anyhow!(e)))?;","sourceCodeStart":252,"sourceCodeEnd":288,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/formatter/mod.rs#L252-L288","documentation":"When key encode is TEXT, the primary key column must be one of the textual/integral types (varchar, bool, smallint, int, bigint, serial, int256) because the encoder renders it as a UTF-8 string. Floats and other types are deliberately rejected (see PR #16377 discussion on ambiguity of float key rendering).","triggerScenarios":"KEY ENCODE TEXT configured while the single PK column resolves to an unsupported DataType (e.g. Float64, Decimal, Timestamp, Struct).","commonSituations":"Using a float or timestamp primary key with a Kafka text key; expecting decimal keys to round-trip losslessly.","solutions":["Cast the key column to varchar in the sink query","Restructure so the key is an integer or varchar column","Use a binary key encode (e.g. AVRO) if the key must preserve non-textual types"],"exampleFix":"// before: t.ts is TIMESTAMP, key_encode='text'\nCREATE SINK s FROM t WITH (key_encode = 'text');\n// after\nCREATE SINK s AS SELECT ts::varchar AS ts_key, * FROM t WITH (key_encode = 'text');","handlingStrategy":"validation","validationCode":"const TEXT_KEY_TYPES: &[DataType] = &[Varchar, Boolean, Int16, Int32, Int64, Serial, Int256];\nif key_encode == \"text\" && !TEXT_KEY_TYPES.contains(&pk_field.data_type) {\n    return Err(\"pk type not supported by text key encode\");\n}","typeGuard":"fn supports_text_key(dt: &DataType) -> bool {\n    matches!(dt, DataType::Varchar | DataType::Boolean | DataType::Int16 | DataType::Int32 | DataType::Int64 | DataType::Serial | DataType::Int256)\n}","tryCatchPattern":null,"preventionTips":["Keep text key encodes on varchar/int/bool keys","Cast float/decimal/timestamp keys to varchar explicitly"],"tags":["sink","type-mismatch","text","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"}