{"record":{"id":"0c7f1ab9e9e78b46","repo":"risingwavelabs/risingwave","slug":"turbopuffer-sink-requires-exactly-one-primary-key","errorCode":null,"errorMessage":"Turbopuffer sink requires exactly one primary_key column","messagePattern":"Turbopuffer sink requires exactly one primary_key column","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/turbopuffer.rs","lineNumber":154,"sourceCode":"impl EnforceSecret for TurbopufferSink {\n    fn enforce_secret<'a>(\n        prop_iter: impl Iterator<Item = &'a str>,\n    ) -> crate::error::ConnectorResult<()> {\n        for prop in prop_iter {\n            TurbopufferConfig::enforce_one(prop)?;\n        }\n        Ok(())\n    }\n}\n\nimpl TryFrom<SinkParam> for TurbopufferSink {\n    type Error = SinkError;\n\n    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {\n        let schema = param.schema();\n        let pk_indices = param.downstream_pk_or_empty();\n        let [pk_index] = pk_indices.as_slice() else {\n            return Err(SinkError::Config(anyhow!(\n                \"Turbopuffer sink requires exactly one primary_key column\"\n            )));\n        };\n        let pk_index = *pk_index;\n        match schema[pk_index].data_type() {\n            DataType::Int16\n            | DataType::Int32\n            | DataType::Int64\n            | DataType::Serial\n            | DataType::Varchar => {}\n            data_type => {\n                return Err(SinkError::Config(anyhow!(\n                    \"Turbopuffer document id column must be an integer or varchar, got {:?}\",\n                    data_type\n                )));\n            }\n        };\n        let config = TurbopufferConfig::from_btreemap(param.properties)?;","sourceCodeStart":136,"sourceCodeEnd":172,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/turbopuffer.rs#L136-L172","documentation":"The turbopuffer sink uses the primary key column as the document id, so it requires the sink schema to have exactly one primary key column. Schemas with zero or multiple PK columns are rejected at sink construction.","triggerScenarios":"CREATE SINK on a materialized view/table whose primary key has 0 or 2+ columns, with connector='turbopuffer'.","commonSituations":"Sink over an append-only MV without a PK; composite primary keys from a join; forgetting to define a PK before creating the sink.","solutions":["Create the sink from a source with exactly one primary key column","Reshape the query to produce a single-column unique key (e.g. concat/hash into one id column)","Choose a different sink connector if composite keys are required"],"exampleFix":"// before\n-- MV with composite PK (a, b)\nCREATE SINK tp FROM mv WITH (connector='turbopuffer', ...);\n// after\nCREATE MATERIALIZED VIEW mv2 AS SELECT md5(concat(a::text, b::text)) AS id, ... FROM mv;\nCREATE SINK tp FROM mv2 WITH (connector='turbopuffer', primary_key = 'id', ...);","handlingStrategy":"validation","validationCode":"const pks = schema.columns.filter(c => c.isPrimaryKey); if (pks.length !== 1) throw new Error('turbopuffer needs exactly one PK column')","typeGuard":"const hasSinglePk = (schema) => schema.primaryKeyColumns?.length === 1;","tryCatchPattern":"catch (SinkError::Config(e)) if e.contains('exactly one primary_key') { reshape schema upstream }","preventionTips":["Ensure upstream MV/table has exactly one PK column before turbopuffer sinks","Avoid composite keys; hash into a single id column","Check PK count in pre-creation schema validation"],"tags":["turbopuffer","schema","primary-key","sink"],"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-14T16:17:12.679Z"}