Repository: risingwavelabs/risingwave
File: src/connector/src/sink/formatter/mod.rs (Rust)

Code before:
                (F::Upsert, E::Avro, None) => Impl::UpsertAvro(build(p).await?),
                (F::Upsert, E::Protobuf, Some(E::Text)) => Impl::UpsertTextProto(build(p).await?),
                (F::Upsert, E::Protobuf, Some(E::Bytes)) => {
                    Impl::UpsertBytesProto(build(p).await?)
                }
                (F::Upsert, E::Template, Some(E::Text)) => {
                    Impl::UpsertTextTemplate(build(p).await?)
                }
                (F::Upsert, E::Template, Some(E::Bytes)) => {
                    Impl::UpsertBytesTemplate(build(p).await?)
                }
                (F::Upsert, E::Template, None) => Impl::UpsertTemplate(build(p).await?),
                (F::Debezium, E::Json, None) => Impl::DebeziumJson(build(p).await?),
                (F::AppendOnly | F::Upsert, E::Text, _) => {
                    return Err(SinkError::Config(anyhow!(
                        "ENCODE TEXT is only valid as key encode."
                    )));
                }
                (F::AppendOnly, E::Avro, _)
                | (F::Upsert, E::Protobuf, _)
                | (F::Upsert, E::Bytes, _)
                | (F::Debezium, E::Json, Some(_))
                | (F::Debezium, E::Avro | E::Protobuf | E::Template | E::Text | E::Bytes, _)
                | (_, E::Parquet, _)
                | (_, _, Some(E::Parquet))
                | (F::AppendOnly, E::Bytes, _)
                | (F::AppendOnly | F::Upsert, _, Some(E::Template) | Some(E::Json) | Some(E::Avro) | Some(E::Protobuf)) // reject other encode as key encode
                => {
                    return Err(SinkError::Config(anyhow!(
                        "sink format/encode/key_encode unsupported: {:?} {:?} {:?}",
                        format_desc.format,
                        format_desc.encode,
                        format_desc.key_encode
                    )));
                }
            },
        )
    }
}

/// Macro to dispatch formatting implementation for all supported sink formatter types.
/// Used when the message key can be either bytes or string.
///
/// Takes a formatter implementation ($impl), binds it to a name ($name),
/// and executes the provided code block ($body) with that binding.
#[macro_export]
macro_rules! dispatch_sink_formatter_impl {
    ($impl:expr, $name:ident, $body:expr) => {
        match $impl {
            SinkFormatterImpl::AppendOnlyJson($name) => $body,
            SinkFormatterImpl::AppendOnlyBytesJson($name) => $body,

Review comment (suggestion):
```suggestion
                | (F::AppendOnly, E::Bytes, Some(_))
```
given `None` is already covered above

Code after:
                (F::Upsert, E::Avro, None) => Impl::UpsertAvro(build(p).await?),
                (F::Upsert, E::Protobuf, Some(E::Text)) => Impl::UpsertTextProto(build(p).await?),
                (F::Upsert, E::Protobuf, Some(E::Bytes)) => {
                    Impl::UpsertBytesProto(build(p).await?)
                }
                (F::Upsert, E::Template, Some(E::Text)) => {
                    Impl::UpsertTextTemplate(build(p).await?)
                }
                (F::Upsert, E::Template, Some(E::Bytes)) => {
                    Impl::UpsertBytesTemplate(build(p).await?)
                }
                (F::Upsert, E::Template, None) => Impl::UpsertTemplate(build(p).await?),
                (F::Debezium, E::Json, None) => Impl::DebeziumJson(build(p).await?),
                (F::AppendOnly | F::Upsert, E::Text, _) => {
                    return Err(SinkError::Config(anyhow!(
                        "ENCODE TEXT is only valid as key encode."
                    )));
                }
                (F::AppendOnly, E::Avro, _)
                | (F::Upsert, E::Protobuf, _)
                | (F::Upsert, E::Bytes, _)
                | (F::Debezium, E::Json, Some(_))
                | (F::Debezium, E::Avro | E::Protobuf | E::Template | E::Text | E::Bytes, _)
                | (_, E::Parquet, _)
                | (_, _, Some(E::Parquet))
                | (F::AppendOnly, E::Bytes, Some(_))
                | (F::AppendOnly | F::Upsert, _, Some(E::Template) | Some(E::Json) | Some(E::Avro) | Some(E::Protobuf)) // reject other encode as key encode
                => {
                    return Err(SinkError::Config(anyhow!(
                        "sink format/encode/key_encode unsupported: {:?} {:?} {:?}",
                        format_desc.format,
                        format_desc.encode,
                        format_desc.key_encode
                    )));
                }
            },
        )
    }
}

/// Macro to dispatch formatting implementation for all supported sink formatter types.
/// Used when the message key can be either bytes or string.
///
/// Takes a formatter implementation ($impl), binds it to a name ($name),
/// and executes the provided code block ($body) with that binding.
#[macro_export]
macro_rules! dispatch_sink_formatter_impl {
    ($impl:expr, $name:ident, $body:expr) => {
        match $impl {
            SinkFormatterImpl::AppendOnlyJson($name) => $body,
            SinkFormatterImpl::AppendOnlyBytesJson($name) => $body,