risingwavelabs/risingwave · error · SinkError::Config

missing FORMAT ... ENCODE ...

Error message

missing FORMAT ... ENCODE ...

What it means

SinkFromParam::try_from expects the sink definition to carry a format descriptor produced by the `FORMAT ... ENCODE ...` clause. Google Pub/Sub sinks need it to know how to encode the payload. When param.format_desc is None (no format clause given), construction fails with this configuration error.

Source

Thrown at src/connector/src/sink/google_pubsub.rs:189

            &self.format_desc,
            self.db_name.clone(),
            self.sink_from_name.clone(),
        )
        .await?
        .into_log_sinker(PUBSUB_SEND_FUTURE_BUFFER_MAX_SIZE))
    }
}

impl TryFrom<SinkParam> for GooglePubSubSink {
    type Error = SinkError;

    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {
        let schema = param.schema();
        let pk_indices = param.downstream_pk_or_empty();
        let config = GooglePubSubConfig::from_btreemap(param.properties)?;
        let format_desc = param
            .format_desc
            .ok_or_else(|| SinkError::Config(anyhow!("missing FORMAT ... ENCODE ...")))?;
        Ok(Self {
            config,
            is_append_only: param.sink_type.is_append_only(),
            schema,
            pk_indices,
            format_desc,
            db_name: param.db_name,
            sink_from_name: param.sink_from_name,
        })
    }
}

struct GooglePubSubPayloadWriter<'w> {
    publisher: &'w mut Publisher,
    message_vec: Vec<PubsubMessage>,
    add_future: DeliveryFutureManagerAddFuture<'w, GooglePubSubSinkDeliveryFuture>,
}

View on GitHub (pinned to 6469eb736d)

Solutions

1. Add the format clause: `FORMAT APPEND ONLY ENCODE JSON` (Pub/Sub only supports append-only)
2. Ensure the sink DDL includes both FORMAT and ENCODE keywords
3. If constructing SinkParam programmatically, populate format_desc before calling try_from

Example fix

// before
CREATE SINK s FROM mv WITH (connector='google_pubsub');
// after
CREATE SINK s FROM mv WITH (connector='google_pubsub', FORMAT APPEND ONLY ENCODE JSON);

fn ensure_format_clause(ddl: &str) -> Result<(), String> {
    if !ddl.to_uppercase().contains("FORMAT APPEND ONLY ENCODE") {
        return Err("google_pubsub sinks require FORMAT APPEND ONLY ENCODE ...".into());
    }
    Ok(())
}

When it happens

Trigger: `CREATE SINK ... WITH (connector='google_pubsub', ...)` omitted the `FORMAT APPEND ONLY ENCODE ...` clause, so SinkParam.format_desc is None when the sink is built.

Common situations: Users coming from older RisingWave syntax that only specified `connector` and `type` without FORMAT/ENCODE; copy-pasted sink DDL missing the format clause.

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

• sink type unsupported: {}
• sink format unsupported: {}
• sink encode unsupported: {}
• unsupported {} as sink key encode
• serde_json deserialization error of ClickHouseConfig from pr

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