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

Code before:
            Ok(val) => val,
            Err(err) => {
                yield Err(err);
                return;
            }
        }
    };
}

pub enum SinkFormatterImpl {
    // append-only
    AppendOnlyJson(AppendOnlyFormatter<JsonEncoder, JsonEncoder>),
    AppendOnlyTextJson(AppendOnlyFormatter<TextEncoder, JsonEncoder>),
    AppendOnlyBytesJson(AppendOnlyFormatter<BytesEncoder, JsonEncoder>),
    AppendOnlyAvro(AppendOnlyFormatter<AvroEncoder, AvroEncoder>),
    AppendOnlyTextAvro(AppendOnlyFormatter<TextEncoder, AvroEncoder>),
    AppendOnlyBytesAvro(AppendOnlyFormatter<BytesEncoder, AvroEncoder>),
    AppendOnlyProto(AppendOnlyFormatter<JsonEncoder, ProtoEncoder>),
    AppendOnlyTextProto(AppendOnlyFormatter<TextEncoder, ProtoEncoder>),
    AppendOnlyBytesProto(AppendOnlyFormatter<BytesEncoder, ProtoEncoder>),
    AppendOnlyTemplate(AppendOnlyFormatter<TemplateEncoder, TemplateEncoder>),
    AppendOnlyTextTemplate(AppendOnlyFormatter<TextEncoder, TemplateEncoder>),
    AppendOnlyBytesTemplate(AppendOnlyFormatter<BytesEncoder, TemplateEncoder>),
    AppendOnlyBytes(AppendOnlyFormatter<JsonEncoder, BytesEncoder>),
    AppendOnlyTextBytes(AppendOnlyFormatter<TextEncoder, BytesEncoder>),
    AppendOnlyBytesBytes(AppendOnlyFormatter<BytesEncoder, BytesEncoder>),
    // upsert
    UpsertJson(UpsertFormatter<JsonEncoder, JsonEncoder>),
    UpsertTextJson(UpsertFormatter<TextEncoder, JsonEncoder>),
    UpsertBytesJson(UpsertFormatter<BytesEncoder, JsonEncoder>),
    UpsertAvro(UpsertFormatter<AvroEncoder, AvroEncoder>),
    UpsertTextAvro(UpsertFormatter<TextEncoder, AvroEncoder>),
    UpsertBytesAvro(UpsertFormatter<BytesEncoder, AvroEncoder>),
    // `UpsertFormatter<ProtoEncoder, ProtoEncoder>` is intentionally left out
    // to avoid using `ProtoEncoder` as key:
    // <https://docs.confluent.io/platform/7.7/control-center/topics/schema.html#c3-schemas-best-practices-key-value-pairs>
    UpsertTextProto(UpsertFormatter<TextEncoder, ProtoEncoder>),
    UpsertBytesProto(UpsertFormatter<BytesEncoder, ProtoEncoder>),
    UpsertTemplate(UpsertFormatter<TemplateEncoder, TemplateEncoder>),
    UpsertTextTemplate(UpsertFormatter<TextEncoder, TemplateEncoder>),
    UpsertBytesTemplate(UpsertFormatter<BytesEncoder, TemplateEncoder>),
    // debezium
    DebeziumJson(DebeziumJsonFormatter),
}

#[derive(Debug, Clone)]
pub struct EncoderParams<'a> {
    format_desc: &'a SinkFormatDesc,
    schema: Schema,
    db_name: String,
    sink_from_name: String,

Review comment (suggestion):
```suggestion
    AppendOnlyBytes(AppendOnlyFormatter<BytesEncoder, BytesEncoder>),
```
Similarly, it makes no sense to allow key encode of text or json either.

Code after:
            Ok(val) => val,
            Err(err) => {
                yield Err(err);
                return;
            }
        }
    };
}

pub enum SinkFormatterImpl {
    // append-only
    AppendOnlyJson(AppendOnlyFormatter<JsonEncoder, JsonEncoder>),
    AppendOnlyTextJson(AppendOnlyFormatter<TextEncoder, JsonEncoder>),
    AppendOnlyBytesJson(AppendOnlyFormatter<BytesEncoder, JsonEncoder>),
    AppendOnlyAvro(AppendOnlyFormatter<AvroEncoder, AvroEncoder>),
    AppendOnlyTextAvro(AppendOnlyFormatter<TextEncoder, AvroEncoder>),
    AppendOnlyBytesAvro(AppendOnlyFormatter<BytesEncoder, AvroEncoder>),
    AppendOnlyProto(AppendOnlyFormatter<JsonEncoder, ProtoEncoder>),
    AppendOnlyTextProto(AppendOnlyFormatter<TextEncoder, ProtoEncoder>),
    AppendOnlyBytesProto(AppendOnlyFormatter<BytesEncoder, ProtoEncoder>),
    AppendOnlyTemplate(AppendOnlyFormatter<TemplateEncoder, TemplateEncoder>),
    AppendOnlyTextTemplate(AppendOnlyFormatter<TextEncoder, TemplateEncoder>),
    AppendOnlyBytesTemplate(AppendOnlyFormatter<BytesEncoder, TemplateEncoder>),
    AppendOnlyBytes(AppendOnlyFormatter<BytesEncoder, BytesEncoder>),
    // upsert
    UpsertJson(UpsertFormatter<JsonEncoder, JsonEncoder>),
    UpsertTextJson(UpsertFormatter<TextEncoder, JsonEncoder>),
    UpsertBytesJson(UpsertFormatter<BytesEncoder, JsonEncoder>),
    UpsertAvro(UpsertFormatter<AvroEncoder, AvroEncoder>),
    UpsertTextAvro(UpsertFormatter<TextEncoder, AvroEncoder>),
    UpsertBytesAvro(UpsertFormatter<BytesEncoder, AvroEncoder>),
    // `UpsertFormatter<ProtoEncoder, ProtoEncoder>` is intentionally left out
    // to avoid using `ProtoEncoder` as key:
    // <https://docs.confluent.io/platform/7.7/control-center/topics/schema.html#c3-schemas-best-practices-key-value-pairs>
    UpsertTextProto(UpsertFormatter<TextEncoder, ProtoEncoder>),
    UpsertBytesProto(UpsertFormatter<BytesEncoder, ProtoEncoder>),
    UpsertTemplate(UpsertFormatter<TemplateEncoder, TemplateEncoder>),
    UpsertTextTemplate(UpsertFormatter<TextEncoder, TemplateEncoder>),
    UpsertBytesTemplate(UpsertFormatter<BytesEncoder, TemplateEncoder>),
    // debezium
    DebeziumJson(DebeziumJsonFormatter),
}

#[derive(Debug, Clone)]
pub struct EncoderParams<'a> {
    format_desc: &'a SinkFormatDesc,
    schema: Schema,
    db_name: String,
    sink_from_name: String,
    topic: &'a str,
}