risingwavelabs/risingwave · error · SinkError::Config

` ` must be set to `varchar

Error message

`{CHANNEL_COLUMN}` must be set to `varchar`

What it means

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.

Solutions

1. Alter or recreate the sink so the `channel` column is of type varchar (e.g. cast via a source view: `channel::varchar as channel`).
2. Point the `channel` option at an existing varchar column in the schema.
3. Remove the `channel` option if a fixed static channel is intended instead of a per-row channel column.

Example fix

// before
CREATE SINK s FROM t WITH (connector='redis', value type='pubsub', channel='ch_id');  -- ch_id is int
// after
CREATE SINK s FROM (SELECT ch_id::varchar AS channel, * FROM t) WITH (connector='redis', value type='pubsub', channel='channel');

// Ensure the channel column exists and is varchar before creating the sink
if (!schema.some(c => c.name === opts.channel && c.type === 'varchar')) {
  throw new Error(`channel column '${opts.channel}' must exist and be varchar`);
}

const isVarchar = (c) => c.type === 'varchar';

try { await createSink(cfg); } catch (e) { if (String(e).includes('must be set to `varchar`')) fixColumnType(cfg.channel); else throw e; }

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "Invalid value" and "allowed values are" config errors: what your library rejected and how to fix it — this error's family across 41 libraries.

Related errors

• Please specific either
• ` ` must be set to `varchar
• auth.method=key_pair_file must not set `password`
• auth.method=key_pair_file must not set `private_key_pem`
• auth.method=key_pair_file requires `private_key_file`

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/e11492afe18662ce. Report an issue: GitHub.

Appendix: source

Thrown at src/connector/src/sink/redis.rs:436

                        )));
                    }
                }
                Some(REDIS_VALUE_TYPE_PUBSUB) => {
                    let channel = self.format_desc.options.get(CHANNEL);
                    let channel_column = self.format_desc.options.get(CHANNEL_COLUMN);
                    if (channel.is_none() && channel_column.is_none())
                        || (channel.is_some() && channel_column.is_some())
                    {
                        return Err(SinkError::Config(anyhow!(
                            "`{CHANNEL}` and `{CHANNEL_COLUMN}` only one can be set"
                        )));
                    }

                    if let Some(channel_column) = channel_column
                        && let Some(channel_column_type) = all_map.get(channel_column)
                        && (channel_column_type != &DataType::Varchar)
                    {
                        return Err(SinkError::Config(anyhow!(
                            "`{CHANNEL_COLUMN}` must be set to `varchar`"
                        )));
                    }

                    let value_format =
                        self.format_desc.options.get(VALUE_FORMAT).ok_or_else(|| {
                            SinkError::Config(anyhow!("Cannot find `{VALUE_FORMAT}`"))
                        })?;
                    TemplateStringEncoder::check_string_format(value_format, &all_map)?;
                }
                Some(REDIS_VALUE_TYPE_STREAM) => {
                    tracing::error!("test:for bug");
                    risingwave_common::license::Feature::RedisSinkStream
                        .check_available()
                        .map_err(|e| anyhow::anyhow!(e))?;
                    let stream = self.format_desc.options.get(STREAM);
                    let stream_column = self.format_desc.options.get(STREAM_COLUMN);
                    if (stream.is_none() && stream_column.is_none())

View on GitHub (pinned to 6469eb736d)