What causes "missing FORMAT ... ENCODE ..." in risingwavelabs/risingwave?

Append a FORMAT ... ENCODE ... clause to the CREATE SINK statement, e.g. FORMAT APPEND ONLY ENCODE JSON.

risingwavelabs/risingwave · error · SinkError::Config

missing FORMAT ... ENCODE ...

Error message

missing FORMAT ... ENCODE ...

What it means

Like the Pulsar sink, the Redis sink constructor requires a FORMAT ... ENCODE declaration (param.format_desc) to select the row formatter. When SinkParam carries no format description, the constructor returns SinkError::Config with 'missing FORMAT ... ENCODE ...'. Without it the sink cannot serialize rows for Redis.

Source

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

#[async_trait]
impl TryFrom<SinkParam> for RedisSink {
    type Error = SinkError;

    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {
        let Some(pk_indices) = param.downstream_pk.clone() else {
            return Err(SinkError::Config(anyhow!(
                "Redis Sink Primary Key must be specified."
            )));
        };
        let config = RedisConfig::from_btreemap(param.properties.clone())?;
        Ok(Self {
            config,
            schema: param.schema(),
            pk_indices,
            format_desc: param
                .format_desc
                .ok_or_else(|| SinkError::Config(anyhow!("missing FORMAT ... ENCODE ...")))?,
            db_name: param.db_name,
            sink_from_name: param.sink_from_name,
        })
    }
}

impl Sink for RedisSink {
    type LogSinker = AsyncTruncateLogSinkerOf<RedisSinkWriter>;

    const SINK_NAME: &'static str = "redis";

    crate::impl_validate_sink_unknown_fields!();

    async fn new_log_sinker(&self, _writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
        Ok(RedisSinkWriter::new(
            self.config.clone(),
            self.schema.clone(),
            self.pk_indices.clone(),

View on GitHub (pinned to 6469eb736d)

Solutions

Append a FORMAT ... ENCODE ... clause to the CREATE SINK statement, e.g. FORMAT APPEND ONLY ENCODE JSON.
Verify the chosen format/encode combination is supported by the Redis sink in your RisingWave version.
When constructing SinkParam programmatically, set format_desc to a valid SinkFormatDesc before calling RedisSink::try_from.

Example fix

// before
CREATE SINK s FROM mv WITH (connector='redis', url='...', primary_key='id');
// after
CREATE SINK s FROM mv WITH (connector='redis', url='...', primary_key='id') FORMAT APPEND ONLY ENCODE JSON;

Defensive patterns

Strategy: validation

Validate before calling

// Ensure FORMAT ... ENCODE is present before creating a redis sink
let ddl_l = ddl.to_lowercase();
if !(ddl_l.contains("format") && ddl_l.contains("encode")) {
    return Err("redis sink requires FORMAT ... ENCODE ... clause".into());
}

Type guard

fn has_format_desc(param: &SinkParam) -> bool { param.format_desc.is_some() }

Try / catch

match RedisSink::try_from(param) {
    Err(SinkError::Config(e)) if e.to_string().contains("missing FORMAT") => {
        eprintln!("add FORMAT APPEND ONLY ENCODE JSON (or equivalent) to CREATE SINK");
    }
    other => { /* proceed */ }
}

Prevention

Always append FORMAT .. ENCODE .. to redis CREATE SINK DDL
Use FORMAT APPEND ONLY ENCODE JSON as the common default
Verify format support for the redis connector version
Set format_desc when building SinkParam in tests/tools

When it happens

Trigger: Building a RedisSink from a SinkParam whose format_desc is None — a CREATE SINK for connector='redis' without a FORMAT ... ENCODE ... clause, or programmatic SinkParam construction that skipped format_desc.

Common situations: Omitting FORMAT APPEND ONLY ENCODE JSON (or DEBEZIUM/etc.) from the DDL; copying an example that lacked the format line; tests constructing SinkParam directly without format_desc.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

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