{"record":{"id":"da0510d304aed16f","repo":"risingwavelabs/risingwave","slug":"turbopuffer-document-id-column-must-be-an-integer-da0510","errorCode":null,"errorMessage":"Turbopuffer document id column must be an integer or varchar","messagePattern":"Turbopuffer document id column must be an integer or varchar","errorType":"http","errorClass":"SinkError::Http","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/turbopuffer.rs","lineNumber":534,"sourceCode":"    // RisingWave UUID IDs can be represented with varchar.\n    fn id_for_row(&self, row: &impl Row) -> Result<DocumentId> {\n        let datum = row.datum_at(self.pk_index).ok_or_else(|| {\n            SinkError::Http(anyhow!(\"Turbopuffer document id column cannot be null\"))\n        })?;\n        match datum {\n            ScalarRefImpl::Int16(value) => Ok(document_id_from_i64(value as i64)),\n            ScalarRefImpl::Int32(value) => Ok(document_id_from_i64(value as i64)),\n            ScalarRefImpl::Int64(value) => Ok(document_id_from_i64(value)),\n            ScalarRefImpl::Serial(value) => Ok(document_id_from_i64(value.into_inner())),\n            ScalarRefImpl::Utf8(value) => {\n                if value.len() > 64 {\n                    return Err(SinkError::Http(anyhow!(\n                        \"Turbopuffer string document id exceeds 64 bytes\"\n                    )));\n                }\n                Ok(DocumentId::String(value.to_owned()))\n            }\n            _ => Err(SinkError::Http(anyhow!(\n                \"Turbopuffer document id column must be an integer or varchar\"\n            ))),\n        }\n    }\n\n    fn upsert_row(&self, row: &impl Row, id: DocumentId) -> Result<Map<String, Value>> {\n        let mut value = self.row_encoder.encode(row)?;\n        value.insert(\n            \"id\".to_owned(),\n            serde_json::to_value(id).expect(\"serialize document id\"),\n        );\n        Ok(value)\n    }\n\n    fn request_body(\n        &self,\n        upsert_rows: Vec<Map<String, Value>>,\n        deletes: Vec<DocumentId>,","sourceCodeStart":516,"sourceCodeEnd":552,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/turbopuffer.rs#L516-L552","documentation":"The Turbopuffer sink derives each document's id from the primary key column of the streamed row. Turbopuffer only accepts unsigned 64-bit integers, UUIDs, or strings up to 64 bytes, so RisingWave maps Int16/Int32/Int64/Serial/VARCHAR primary keys to document ids. When the primary key datum is any other type (e.g. Float, Decimal, Boolean, Struct), id_for_row throws this error. The caller (write_chunk) logs a warning and skips the row, so the data is silently dropped.","triggerScenarios":"Creating a Turbopuffer sink whose primary key column is not an integer type (SMALLINT/INT/BIGINT/SERIAL) or VARCHAR — e.g. a FLOAT, DECIMAL, BOOLEAN, or STRUCT primary key — and writing a row through write_chunk.","commonSituations":"Defining a materialized view or table with a decimal or float primary key for data that will be sunk to Turbopuffer; forgetting that Turbopuffer ids must be integers or short strings; legacy schemas with composite or non-scalar primary keys.","solutions":["Change the sink's source table/materialized view primary key to BIGINT or VARCHAR.","Cast the primary key to VARCHAR or BIGINT in an intermediate materialized view before sinking.","Use a different sink that supports arbitrary primary key types."],"exampleFix":"// before: PRIMARY KEY (price DECIMAL) sinking to turbopuffer\n// after\nCREATE MATERIALIZED VIEW mv_ids AS SELECT price::VARCHAR AS price_id, * FROM source;\n-- sink from mv_ids with PRIMARY KEY (price_id)","handlingStrategy":"validation","validationCode":"// ensure the sink PK column is BIGINT or VARCHAR before creating the sink\n-- SELECT data_type FROM rw_catalog.rw_columns WHERE name = '<pk_col>' AND relation_id = <table_id>;\nfn is_valid_tpuf_pk(t: &str) -> bool {\n    matches!(t, \"SMALLINT\" | \"INTEGER\" | \"BIGINT\" | \"SERIAL\" | \"VARCHAR\")\n}","typeGuard":"fn valid_doc_id(v: &ScalarRefImpl) -> bool {\n    matches!(v, ScalarRefImpl::Int16(_) | ScalarRefImpl::Int32(_) | ScalarRefImpl::Int64(_) | ScalarRefImpl::Serial(_) | ScalarRefImpl::Utf8(_))\n}","tryCatchPattern":null,"preventionTips":["Always use BIGINT or VARCHAR primary keys on tables destined for Turbopuffer.","Cast exotic PK types in an intermediate materialized view.","Remember the sink skips (drops) such rows with only a log warning — check logs after creating new sinks."],"tags":["turbopuffer","sink","type-mismatch","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-14T11:17:12.474Z"}