{"record":{"id":"0ee95d3c1bda3789","repo":"risingwavelabs/risingwave","slug":"stream-column-must-be-set-to-varchar","errorCode":null,"errorMessage":"`{STREAM_COLUMN}` must be set to `varchar`","messagePattern":"`(.+?)` must be set to `varchar`","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/redis.rs","lineNumber":466,"sourceCode":"                    tracing::error!(\"test:for bug\");\n                    risingwave_common::license::Feature::RedisSinkStream\n                        .check_available()\n                        .map_err(|e| anyhow::anyhow!(e))?;\n                    let stream = self.format_desc.options.get(STREAM);\n                    let stream_column = self.format_desc.options.get(STREAM_COLUMN);\n                    if (stream.is_none() && stream_column.is_none())\n                        || (stream.is_some() && stream_column.is_some())\n                    {\n                        return Err(SinkError::Config(anyhow!(\n                            \"Please specific either `{STREAM}` or `{STREAM_COLUMN}`. They are mutually exclusive options.\"\n                        )));\n                    }\n\n                    if let Some(stream_column) = stream_column\n                        && let Some(stream_column_type) = all_map.get(stream_column)\n                        && (stream_column_type != &DataType::Varchar)\n                    {\n                        return Err(SinkError::Config(anyhow!(\n                            \"`{STREAM_COLUMN}` must be set to `varchar`\"\n                        )));\n                    }\n\n                    let value_format =\n                        self.format_desc.options.get(VALUE_FORMAT).ok_or_else(|| {\n                            SinkError::Config(anyhow!(\"Cannot find `{VALUE_FORMAT}`\"))\n                        })?;\n                    let key_format = self.format_desc.options.get(KEY_FORMAT).ok_or_else(|| {\n                        SinkError::Config(anyhow!(\n                            \"Cannot find '{KEY_FORMAT}', please set it or use JSON\"\n                        ))\n                    })?;\n                    TemplateStringEncoder::check_string_format(key_format, &pk_map)?;\n                    TemplateStringEncoder::check_string_format(value_format, &all_map)?;\n                }\n                _ => {\n                    return Err(SinkError::Config(anyhow!(","sourceCodeStart":448,"sourceCodeEnd":484,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/redis.rs#L448-L484","documentation":"Mirror of the channel check for streams: when `value type = 'stream'` and `stream column` is set, `validate` looks up that column in the schema and requires it to be `varchar`. Any other type causes this config error.","triggerScenarios":"Create a sink with `value type='stream'` and `stream column='<col>'` where `<col>` is int, timestamp, struct, or any non-varchar type.","commonSituations":"Users select a numeric or timestamp column as the per-row stream name; schema drift after adding a stream column option; copying pubsub `channel` configs across sinks whose column types differ.","solutions":["Cast the column to varchar in the source query and point `stream column` at the casted column.","Choose an existing varchar column for `stream column`.","Use the fixed `stream` option instead if the stream name doesn't vary per row."],"exampleFix":"// before\nWITH (connector='redis', value type='stream', stream column='region_id')  -- int\n// after\nCREATE SINK s FROM (SELECT region_id::varchar AS region, * FROM t) WITH (connector='redis', value type='stream', stream column='region');","handlingStrategy":"validation","validationCode":"const col = schema.find(c => c.name === opts['stream column']);\nif (col && col.type !== 'varchar') throw new Error('stream column must be varchar');","typeGuard":"const isVarchar = (c) => c.type === 'varchar';","tryCatchPattern":"try { await createSink(cfg); } catch (e) { if (String(e).includes('STREAM_COLUMN` must be set to `varchar`')) recastColumn(cfg['stream column']); else throw e; }","preventionTips":["Cast the stream column to varchar upstream in the sink query","Re-validate configs after any schema change","Prefer a fixed stream option when per-row naming is not required"],"tags":["redis","sink","config-validation"],"backgroundTag":"invalid-config-value","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"}