risingwavelabs/risingwave · error · ConnectorError

Unsupported combination of format {:?} and encode {:?}

Error message

Unsupported combination of format {:?} and encode {:?}

What it means

RisingWave validates that the (format, encode) pair given in a source's WITH/FORMAT options is one of the supported combinations (e.g. Plain+Bytes, Upsert+Protobuf). If the pairing is not in the translation table in extract_source_struct, creation is rejected with this message naming both values. It prevents sources being built with encoders that their format cannot consume.

Source

Thrown at src/connector/src/source/base.rs:536

        }
        (PbFormatType::Native, PbEncodeType::Native) => {
            (SourceFormat::Native, SourceEncode::Native)
        }
        (PbFormatType::None, PbEncodeType::None) => (SourceFormat::None, SourceEncode::None),
        (PbFormatType::Debezium, PbEncodeType::Avro) => {
            (SourceFormat::Debezium, SourceEncode::Avro)
        }
        (PbFormatType::Upsert, PbEncodeType::Json) => (SourceFormat::Upsert, SourceEncode::Json),
        (PbFormatType::Upsert, PbEncodeType::Avro) => (SourceFormat::Upsert, SourceEncode::Avro),
        (PbFormatType::DebeziumMongo, PbEncodeType::Json) => {
            (SourceFormat::DebeziumMongo, SourceEncode::Json)
        }
        (PbFormatType::Plain, PbEncodeType::Bytes) => (SourceFormat::Plain, SourceEncode::Bytes),
        (PbFormatType::Upsert, PbEncodeType::Protobuf) => {
            (SourceFormat::Upsert, SourceEncode::Protobuf)
        }
        (format, encode) => {
            bail!(
                "Unsupported combination of format {:?} and encode {:?}",
                format,
                encode
            );
        }
    };
    Ok(SourceStruct::new(format, encode))
}

/// Stream of [`SourceMessage`]. Messages flow through the stream in the unit of a batch.
pub type BoxSourceMessageStream =
    BoxStream<'static, crate::error::ConnectorResult<Vec<SourceMessage>>>;
/// Stream of source message events.
pub type BoxSourceMessageEventStream =
    BoxStream<'static, crate::error::ConnectorResult<SourceMessageEvent>>;
/// Stream of [`StreamChunk`]s parsed from the messages from the external source.
pub type BoxSourceChunkStream = BoxStream<'static, crate::error::ConnectorResult<StreamChunk>>;
/// Stream of source reader events.

View on GitHub (pinned to 6469eb736d)

Solutions

  1. Pick a supported combination, e.g. FORMAT PLAIN ENCODE BYTES or FORMAT UPSERT ENCODE PROTOBUF
  2. Check the match in extract_source_struct (src/connector/src/source/base.rs) for the exact list of valid pairs and match your options to one
  3. If a new valid pair is genuinely needed, add the (format, encode) arm to extract_source_struct and rebuild

Example fix

// before
CREATE SOURCE s (...) WITH (connector='kafka') FORMAT PLAIN ENCODE PROTOBUF;
// after
CREATE SOURCE s (...) WITH (connector='kafka') FORMAT PLAIN ENCODE BYTES;
Defensive patterns

Strategy: validation

Validate before calling

const VALID_PAIRS: &[(PbFormatType, PbEncodeType)] = &[
    (PbFormatType::Plain, PbEncodeType::Bytes),
    (PbFormatType::Upsert, PbEncodeType::Protobuf),
];
assert!(VALID_PAIRS.contains(&(fmt, enc)), "unsupported format/encode: {:?}/{:?}", fmt, enc);

Prevention

When it happens

Trigger: Calling extract_source_struct (via source CREATE/ALTER paths `new` or `handle_alter_source_column`) with a PbFormatType/PbEncodeType pair absent from the match arms, e.g. FORMAT PLAIN ENCODE PROTOBUF or FORMAT DEBEZIUM JSON when unsupported, or a type+encode pairing added to protobuf enums without updating the match.

Common situations: Typo or unsupported FORMAT ... ENCODE ... combination in CREATE SOURCE/CREATE TABLE ... WITH connector; enabling a new encode type in the protobuf definitions but forgetting to add the match arm; altering a source's format column pair to an invalid pairing.

Related errors

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