{"record":{"id":"7ca1b20096a48cee","repo":"risingwavelabs/risingwave","slug":"unexpected-payload-column-type-expected-varchar-o","errorCode":null,"errorMessage":"unexpected payload column type, expected varchar or jsonb","messagePattern":"unexpected payload column type, expected varchar or jsonb","errorType":"error_code","errorClass":"SinkError::Http","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/http.rs","lineNumber":369,"sourceCode":"            },\n        }\n    }\n\n    fn strip_payload_for_log(&self, row: &impl Row) -> String {\n        match row.datum_at(self.payload_index) {\n            Some(ScalarRefImpl::Utf8(s)) => strip_text_payload(s),\n            Some(ScalarRefImpl::Jsonb(j)) => strip_jsonb_payload(j),\n            Some(_) => \"<unexpected payload type>\".to_owned(),\n            None => \"NULL\".to_owned(),\n        }\n    }\n\n    fn extract_payload(&self, row: &impl Row) -> Result<Option<String>> {\n        Ok(match row.datum_at(self.payload_index) {\n            Some(ScalarRefImpl::Utf8(s)) => Some(s.to_owned()),\n            Some(ScalarRefImpl::Jsonb(j)) => Some(j.to_string()),\n            Some(_) => {\n                return Err(SinkError::Http(anyhow!(\n                    \"unexpected payload column type, expected varchar or jsonb\"\n                )));\n            }\n            None => None, // skip NULL rows\n        })\n    }\n}\n\nfn strip_text_payload(payload: &str) -> String {\n    const EDGE_CHAR_COUNT: usize = 100;\n    let char_count = payload.chars().count();\n    if char_count <= EDGE_CHAR_COUNT * 2 {\n        return payload.to_owned();\n    }\n\n    let prefix: String = payload.chars().take(EDGE_CHAR_COUNT).collect();\n    let suffix: String = payload.chars().skip(char_count - EDGE_CHAR_COUNT).collect();\n    format!(\"{prefix}...{suffix}\")","sourceCodeStart":351,"sourceCodeEnd":387,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/http.rs#L351-L387","documentation":"At write time, `extract_payload` accepts only Utf8 (VARCHAR) and Jsonb scalars for the payload column. Any other scalar type in the payload column datum causes `write_chunk` to fail with this SinkError::Http, since the payload cannot be turned into a request body.","triggerScenarios":"A row's payload column datum is a scalar other than Utf8/Jsonb during `write_chunk` -> `extract_payload` — typically after the underlying column type changed post sink creation, or schema drift in a source table.","commonSituations":"Column type altered after sink creation (e.g. VARCHAR -> INT); upstream query changed so the sink now receives numeric/boolean/struct payloads; index/column misalignment in internal sink plumbing.","solutions":["Recreate the sink to re-validate the payload column type","Cast the payload to VARCHAR or JSONB in the sink query","Restore the payload column's original VARCHAR/JSONB type"],"exampleFix":"// before\nSELECT payload FROM t; -- payload became INT after ALTER\n// after\nSELECT payload::JSONB AS payload FROM t;","handlingStrategy":"try-catch","validationCode":null,"typeGuard":"fn is_payload_compatible(d: &ScalarRefImpl) -> bool {\n    matches!(d, ScalarRefImpl::Utf8(_) | ScalarRefImpl::Jsonb(_))\n}","tryCatchPattern":"match res {\n    Err(e) if e.to_string().contains(\"unexpected payload column type\") => {\n        // fix payload column type or recreate sink\n    }\n    other => other?,\n}","preventionTips":["Keep payload column VARCHAR/JSONB for the sink lifetime","Cast payload in the sink query: payload::JSONB","Recreate sinks after upstream schema/type changes"],"tags":["sink","http","runtime-error","type-mismatch"],"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"}