risingwavelabs/risingwave · error · SinkError

missing FORMAT ... ENCODE ...

Error message

missing FORMAT ... ENCODE ...

What it means

When building an OpenDAL-based file sink, the `FORMAT ... ENCODE ...` declaration of the sink is required to know how rows are serialized. If `param.format_desc` is None and the engine is not Snowflake (which has a plain-JSON fallback), the connector rejects creation with this config error.

Source

Thrown at src/connector/src/sink/file_sink/opendal_sink.rs:209

impl<S: OpendalSinkBackend> TryFrom<SinkParam> for FileSink<S> {
    type Error = SinkError;

    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {
        let schema = param.schema();
        let config = S::from_btreemap(param.properties)?;
        let unknown_fields = crate::sink::UnknownFields::unknown_fields(&config);
        let path = S::get_path(config.clone());
        let op = S::new_operator(config.clone())?;
        let batching_strategy = S::get_batching_strategy(config);
        let engine_type = S::get_engine_type();
        let format_desc = match param.format_desc {
            Some(desc) => desc,
            None => {
                if let EngineType::Snowflake = engine_type {
                    SinkFormatDesc::plain_json_for_snowflake_only()
                } else {
                    return Err(SinkError::Config(anyhow!("missing FORMAT ... ENCODE ...")));
                }
            }
        };
        Ok(Self {
            op,
            path,
            schema,
            is_append_only: param.sink_type.is_append_only(),
            batching_strategy,
            format_desc,
            engine_type,
            unknown_fields,
            _marker: PhantomData,
        })
    }
}

pub struct OpenDalSinkWriter {

View on GitHub (pinned to 6469eb736d)

Solutions

1. Add an explicit FORMAT/ENCODE clause to the CREATE SINK statement, e.g. `FORMAT PLAIN ENCODE JSON` or `FORMAT DEBEZIUM_JSON ENCODE JSON`
2. If targeting Snowflake engine specifically, ensure the engine type is detected correctly (format is defaulted there)
3. Check that the SQL was not truncated or that a wrapper is dropping the FORMAT option

Example fix

// before
CREATE SINK s FROM mv WITH (connector='file_s3', path='s3://bucket/out/');
// after
CREATE SINK s FROM mv WITH (connector='file_s3', path='s3://bucket/out/') FORMAT PLAIN ENCODE JSON;

// Ensure the sink SQL contains a FORMAT ... ENCODE clause
const required = /FORMAT\s+(PLAIN|DEBEZIUM|MAXWELL|CANAL|DEBEZIUM_JSON|UPSERT)\s+ENCODE\s+(JSON|AVRO|PROTOBUF)/i;
if (!required.test(createSinkSql)) throw new Error('CREATE SINK needs a FORMAT ... ENCODE clause');

When it happens

Trigger: Creating a file sink (e.g. S3/opendal based) without a `FORMAT ... ENCODE ...` clause on a non-Snowflake engine — the property simply was never parsed into format_desc.

Common situations: Old SQL habits from other systems where FORMAT is optional; copy-pasted CREATE SINK statements stripped of the FORMAT line; using a generic sink wrapper that does not forward format options.

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

• missing FORMAT ... ENCODE ...
• missing FORMAT ... ENCODE ...
• Turbopuffer sink requires either namespace or namespace_colu
• Turbopuffer sink requires distance_metric when sink schema c
• Must specify '{}' or '{}'

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