{"record":{"id":"42623417dab4f923","repo":"risingwavelabs/risingwave","slug":"turbopuffer-schema-option-references-unknown-attri","errorCode":null,"errorMessage":"Turbopuffer schema option references unknown attribute column '{}'","messagePattern":"Turbopuffer schema option references unknown attribute column '(.+?)'","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/turbopuffer.rs","lineNumber":767,"sourceCode":"        .collect::<HashSet<_>>();\n    let columns: HashSet<String> = match value {\n        Some(value) if value.trim() == \"*\" => {\n            return Ok(attribute_names\n                .into_iter()\n                .map(str::to_owned)\n                .collect::<HashSet<_>>());\n        }\n        Some(value) => value\n            .split(',')\n            .map(str::trim)\n            .filter(|name| !name.is_empty())\n            .map(str::to_owned)\n            .collect(),\n        None => return Ok(HashSet::new()),\n    };\n    for column in &columns {\n        if !attribute_names.contains(column.as_str()) {\n            return Err(SinkError::Config(anyhow!(\n                \"Turbopuffer schema option references unknown attribute column '{}'\",\n                column\n            )));\n        }\n    }\n    Ok(columns)\n}\n\nfn build_turbopuffer_schema(\n    schema: &Schema,\n    attribute_indices: &[usize],\n    full_text_search_columns: &HashSet<String>,\n    filterable_columns: &HashSet<String>,\n) -> Result<Value> {\n    let mut result = Map::new();\n    for index in attribute_indices {\n        let field = &schema[*index];\n        let mut config = Map::new();","sourceCodeStart":749,"sourceCodeEnd":785,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/turbopuffer.rs#L749-L785","documentation":"parse_column_selection parses options like turbopuffer.filterable or turbopuffer.full_text_search, which take a comma-separated list of column names, and validates each name against the sink's attribute columns. If a listed column does not exist among the sink's attributes, the sink config is rejected at creation time with this error.","triggerScenarios":"Creating a Turbopuffer sink where turbopuffer.filterable='a,b' (or turbopuffer.full_text_search) mentions a column name that is not one of the table's attribute columns — typically a typo, a column dropped/renamed in the source, or the primary key/id column which is not an attribute.","commonSituations":"Renaming columns in the upstream table without updating the sink WITH options; typos in the comma-separated list; assuming the primary key column can be marked filterable (it is handled separately as the document id).","solutions":["Check the exact column names in the sink's schema and correct the option list.","Remove stale/renamed columns from turbopuffer.filterable / turbopuffer.full_text_search.","Do not list the primary key column in these options; it is mapped to the Turbopuffer document id automatically."],"exampleFix":"// before\nWITH (connector='turbopuffer', turbopuffer.filterable='catgory,ts')\n// after (typo fixed)\nWITH (connector='turbopuffer', turbopuffer.filterable='category,ts')","handlingStrategy":"validation","validationCode":"-- confirm every listed column exists before CREATE SINK\n-- SELECT name FROM rw_catalog.rw_columns WHERE relation_id = <sink_source_id>;\nfn cols_exist(listed: &[&str], schema_cols: &[&str]) -> Vec<String> {\n    listed.iter().filter(|c| !schema_cols.contains(c)).map(|c| c.to_string()).collect()\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Copy column names directly from the table schema, not from memory.","Update sink WITH options whenever upstream columns are renamed or dropped.","Don't list the primary key in filterable/full_text_search — it's the doc id."],"tags":["turbopuffer","config","schema","sink"],"backgroundTag":"resource-not-found","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}