{"record":{"id":"e11492afe18662ce","repo":"risingwavelabs/risingwave","slug":"channel-column-must-be-set-to-varchar","errorCode":null,"errorMessage":"`{CHANNEL_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":436,"sourceCode":"                        )));\n                    }\n                }\n                Some(REDIS_VALUE_TYPE_PUBSUB) => {\n                    let channel = self.format_desc.options.get(CHANNEL);\n                    let channel_column = self.format_desc.options.get(CHANNEL_COLUMN);\n                    if (channel.is_none() && channel_column.is_none())\n                        || (channel.is_some() && channel_column.is_some())\n                    {\n                        return Err(SinkError::Config(anyhow!(\n                            \"`{CHANNEL}` and `{CHANNEL_COLUMN}` only one can be set\"\n                        )));\n                    }\n\n                    if let Some(channel_column) = channel_column\n                        && let Some(channel_column_type) = all_map.get(channel_column)\n                        && (channel_column_type != &DataType::Varchar)\n                    {\n                        return Err(SinkError::Config(anyhow!(\n                            \"`{CHANNEL_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                    TemplateStringEncoder::check_string_format(value_format, &all_map)?;\n                }\n                Some(REDIS_VALUE_TYPE_STREAM) => {\n                    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())","sourceCodeStart":418,"sourceCodeEnd":454,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/redis.rs#L418-L454","documentation":"The Redis sink's `validate` step checks that, when a `channel` option is configured for PUBSUB value type, the column named by `channel` exists in the sink schema and has type `varchar`. If the column exists but is any other data type, validation fails with this config error before the sink is created.","triggerScenarios":"Create a sink with `value type = 'pubsub'` and `channel = '<col>'` where `<col>` exists in the table but is int, boolean, struct, etc. rather than varchar.","commonSituations":"Users pick an integer or timestamp column as the channel name column, or rename/refactor columns so the channel column's type changed; often seen when reusing an existing table for a redis pubsub sink.","solutions":["Alter or recreate the sink so the `channel` column is of type varchar (e.g. cast via a source view: `channel::varchar as channel`).","Point the `channel` option at an existing varchar column in the schema.","Remove the `channel` option if a fixed static channel is intended instead of a per-row channel column."],"exampleFix":"// before\nCREATE SINK s FROM t WITH (connector='redis', value type='pubsub', channel='ch_id');  -- ch_id is int\n// after\nCREATE SINK s FROM (SELECT ch_id::varchar AS channel, * FROM t) WITH (connector='redis', value type='pubsub', channel='channel');","handlingStrategy":"validation","validationCode":"// Ensure the channel column exists and is varchar before creating the sink\nif (!schema.some(c => c.name === opts.channel && c.type === 'varchar')) {\n  throw new Error(`channel column '${opts.channel}' must exist and be varchar`);\n}","typeGuard":"const isVarchar = (c) => c.type === 'varchar';","tryCatchPattern":"try { await createSink(cfg); } catch (e) { if (String(e).includes('must be set to `varchar`')) fixColumnType(cfg.channel); else throw e; }","preventionTips":["Cast channel columns to varchar in the sink's source query","Keep a schema check step before every sink creation","Document required column types next to sink configs"],"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"}