{"record":{"id":"e3357ca8d436c45f","repo":"risingwavelabs/risingwave","slug":"turbopuffer-sink-does-not-support-column-type","errorCode":null,"errorMessage":"Turbopuffer sink does not support column type {}","messagePattern":"Turbopuffer sink does not support column type (.+?)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/turbopuffer.rs","lineNumber":871,"sourceCode":"        DataType::List(list_type) => match list_type.elem() {\n            DataType::Boolean => Ok(\"[]bool\".to_owned()),\n            DataType::Int16 | DataType::Int32 | DataType::Int64 | DataType::Serial => {\n                Ok(\"[]int\".to_owned())\n            }\n            DataType::Float32 | DataType::Float64 | DataType::Decimal => Ok(\"[]float\".to_owned()),\n            DataType::Varchar => Ok(\"[]string\".to_owned()),\n            DataType::Date | DataType::Timestamp | DataType::Timestamptz => {\n                Ok(\"[]datetime\".to_owned())\n            }\n            elem_type => Err(unsupported_type(&format!(\"list element {:?}\", elem_type))),\n        },\n        DataType::Vector(dimension) => Ok(format!(\"[{}]f32\", dimension)),\n        data_type => Err(unsupported_type(&format!(\"{:?}\", data_type))),\n    }\n}\n\nfn unsupported_type(data_type: &str) -> SinkError {\n    SinkError::Config(anyhow!(\n        \"Turbopuffer sink does not support column type {}\",\n        data_type\n    ))\n}\n\n#[cfg(test)]\nmod tests {\n    #[cfg(not(madsim))]\n    use std::collections::VecDeque;\n    #[cfg(not(madsim))]\n    use std::io::{Read, Write};\n    #[cfg(not(madsim))]\n    use std::net::TcpListener;\n    #[cfg(not(madsim))]\n    use std::sync::{Arc, Mutex, mpsc};\n    #[cfg(not(madsim))]\n    use std::thread;\n","sourceCodeStart":853,"sourceCodeEnd":889,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/turbopuffer.rs#L853-L889","documentation":"turbopuffer_type maps RisingWave data types to Turbopuffer schema types ([N]f32 vectors, string, int, boolean, etc.). Any column type outside the supported set reaches the catch-all arm, which produces this error naming the unsupported Rust DataType. The sink cannot be created until all columns map to a supported Turbopuffer type.","triggerScenarios":"Creating a Turbopuffer sink whose table contains a column of an unmapped type — e.g. DECIMAL, DATE/TIME/TIMESTAMP, JSONB, STRUCT, MAP, LIST of non-string elements, BYTEA — detected during schema construction in try_from.","commonSituations":"Sinking tables with decimal money columns or timestamp columns directly; complex nested structs from upstream sources; bytea blobs.","solutions":["Inspect the type name in the error and exclude that column from the sink (or from the schema list).","Cast unsupported columns in a materialized view: DECIMAL→DOUBLE PRECISION (if float mapping exists) or TIMESTAMP/DATE→VARCHAR.","Flatten STRUCT columns into scalars or serialize them to JSON strings before sinking.","Check turbopuffer_type in src/connector/src/sink/turbopuffer.rs for the currently supported type set."],"exampleFix":"// before: created_at TIMESTAMP column sinking to turbopuffer\n// after\nCREATE MVIEW mv AS SELECT *, created_at::VARCHAR AS created_at FROM src;\n-- sink from mv","handlingStrategy":"validation","validationCode":"-- check all sink columns map to supported Turbopuffer types\n-- SELECT name, data_type FROM rw_catalog.rw_columns WHERE relation_id = <table_id>;\nfn tpuf_supported(dt: &str) -> bool {\n    matches!(dt,\n        \"BOOLEAN\" | \"SMALLINT\" | \"INTEGER\" | \"BIGINT\" | \"REAL\" | \"DOUBLE PRECISION\" |\n        \"VARCHAR\" | \"VARCHAR[]\" | \"DATE\" ) || dt.starts_with(\"VECTOR\")\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Review turbopuffer_type's supported types in turbopuffer.rs before designing sink tables.","Pre-cast DECIMAL/TIMESTAMP/STRUCT columns in an intermediate materialized view.","Keep sink tables to scalar primitives and VECTOR columns."],"tags":["turbopuffer","type-mismatch","schema","sink"],"backgroundTag":"unsupported-operation","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"}