risingwavelabs/risingwave · error · SinkError

Cannot find '{VALUE_FORMAT}',please set it.

Error message

Cannot find '{VALUE_FORMAT}',please set it.

What it means

For `redis_value_type = 'pubsub'` sinks WITHOUT a primary key, the builder encodes each message with a template and requires the `value_format` option. If `value_format` is absent the sink build fails with this SinkError::Config.

Source

Thrown at src/connector/src/sink/formatter/mod.rs:379

                    let channel = b.format_desc.options.get(CHANNEL).cloned();
                    let channel_column = b.format_desc.options.get(CHANNEL_COLUMN).cloned();
                    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"
                        )));
                    }
                    TemplateEncoder::new_pubsub_stream_key(
                        b.schema,
                        pk_indices,
                        channel,
                        channel_column,
                    )
                }
                None => {
                    let template = b.format_desc.options.get(VALUE_FORMAT).ok_or_else(|| {
                        SinkError::Config(anyhow!("Cannot find '{VALUE_FORMAT}',please set it."))
                    })?;
                    Ok(TemplateEncoder::new_string(
                        b.schema,
                        pk_indices,
                        template.clone(),
                    ))
                }
            },
            REDIS_VALUE_TYPE_STREAM => match pk_indices {
                Some(_) => {
                    let stream = b.format_desc.options.get(STREAM).cloned();
                    let stream_column = b.format_desc.options.get(STREAM_COLUMN).cloned();
                    if (stream.is_none() && stream_column.is_none())
                        || (stream.is_some() && stream_column.is_some())
                    {
                        return Err(SinkError::Config(anyhow!(
                            "`{STREAM}` and `{STREAM_COLUMN}` only one can be set"
                        )));

View on GitHub (pinned to 6469eb736d)

Solutions

  1. Add `value_format = '<template>'` (e.g. a JSON template with column placeholders) to the WITH options.
  2. Ensure the channel options (`channel`/`channel_column`) are consistent with the table's primary key.
  3. Check the option spelling — it must be exactly `value_format`.

Example fix

// before
WITH (redis_value_type='pubsub', channel='events');
// after
WITH (redis_value_type='pubsub', channel='events', value_format='{"k": %s}');
Defensive patterns

Strategy: validation

Validate before calling

if value_type == "pubsub" && !has_pk && !with_options.contains_key("value_format") {
    return Err("PK-less Redis pubsub sink requires `value_format`".into());
}

Try / catch

match result {
    Err(e) if e.to_string().contains("`value_format`") => {
        eprintln!("Add `value_format = '<template>'` to WITH options: {}", e);
    }
    other => other?,
}

Prevention

When it happens

Trigger: Creating a Redis pubsub sink on a table without a primary key but omitting `value_format = '<template>'` from the WITH options.

Common situations: Users switch a pubsub sink from a primary-key table (which needed `channel`/`channel_column` only) to a no-pk table and do not add the now-required `value_format` template.

Understand the failure class

Background: "Must pass :limit option" / "Missing required option" — required option errors explained — this error's family across 41 libraries.

Related errors

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