feat(sink): support encode jsonb data as dynamic json type in sink

src/connector/src/sink/encoder/mod.rs

@@ -150,6 +150,33 @@ pub enum CustomJsonType {
  None,
 }
 
+/// How the jsonb type is encoded.
+///
+/// - `String`: encode jsonb as string. `[1, true, "foo"] -> "[1, true, \"foo\"]"`
+/// - `Dynamic`: encode jsonb as json type dynamically. `[1, true, "foo"] -> [1, true, "foo"]`
+/// - `Custom`: decided by [`CustomJsonType`].
+pub enum EncodeJsonbMode {
+ String,
+ Dynamic,
+ Custom,
+}
+
+impl EncodeJsonbMode {
+ pub const OPTION_KEY: &'static str = "encode.jsonb.mode";
+
+ pub fn from_options(options: &BTreeMap<String, String>) -> Result<Self> {
+ match options.get(Self::OPTION_KEY).map(std::ops::Deref::deref) {
+ Some("string") | None => Ok(Self::String),
+ Some("dynamic") => Ok(Self::Dynamic),
+ Some(v) => Err(super::SinkError::Config(anyhow::anyhow!(
+ "unrecognized {} value {}",
+ Self::OPTION_KEY,
+ v
+ ))),
+ }
+ }
+}
+
 #[derive(Debug)]
 struct FieldEncodeError {
  message: String,

src/connector/src/sink/formatter/debezium_json.rs

@@ -21,8 +21,8 @@ use tracing::warn;
 
 use super::{Result, SinkFormatter, StreamChunk};
 use crate::sink::encoder::{
- DateHandlingMode, JsonEncoder, RowEncoder, TimeHandlingMode, TimestampHandlingMode,
- TimestamptzHandlingMode,
+ DateHandlingMode, EncodeJsonbMode, JsonEncoder, RowEncoder, TimeHandlingMode,
+ TimestampHandlingMode, TimestamptzHandlingMode,
 };
 use crate::tri;
 
@@ -69,6 +69,7 @@ impl DebeziumJsonFormatter {
  TimestampHandlingMode::Milli,
  TimestamptzHandlingMode::UtcString,
  TimeHandlingMode::Milli,
+ EncodeJsonbMode::String,
  );
  let val_encoder = JsonEncoder::new(
  schema.clone(),
@@ -77,6 +78,7 @@ impl DebeziumJsonFormatter {
  TimestampHandlingMode::Milli,
  TimestamptzHandlingMode::UtcString,
  TimeHandlingMode::Milli,
+ EncodeJsonbMode::String,
  );
  Self {
  schema,
@@ -397,6 +399,7 @@ mod tests {
  TimestampHandlingMode::Milli,
  TimestamptzHandlingMode::UtcString,
  TimeHandlingMode::Milli,
+ EncodeJsonbMode::String,
  );
  let json_chunk = chunk_to_json(chunk, &encoder).unwrap();
  let schema_json = schema_to_json(&schema, "test_db", "test_table");

src/connector/src/sink/formatter/mod.rs

@@ -31,7 +31,8 @@ use super::catalog::{SinkEncode, SinkFormat, SinkFormatDesc};
 use super::encoder::template::TemplateEncoder;
 use super::encoder::text::TextEncoder;
 use super::encoder::{
- DateHandlingMode, KafkaConnectParams, TimeHandlingMode, TimestamptzHandlingMode,
+ DateHandlingMode, EncodeJsonbMode, KafkaConnectParams, TimeHandlingMode,
+ TimestamptzHandlingMode,
 };
 use super::redis::{KEY_FORMAT, VALUE_FORMAT};
 use crate::sink::encoder::{
@@ -113,13 +114,15 @@ pub trait EncoderBuild: Sized {
 impl EncoderBuild for JsonEncoder {
  async fn build(b: EncoderParams<'_>, pk_indices: Option<Vec<usize>>) -> Result<Self> {
  let timestamptz_mode = TimestamptzHandlingMode::from_options(&b.format_desc.options)?;
+ let encode_jsonb_mode = EncodeJsonbMode::from_options(&b.format_desc.options)?;
  let encoder = JsonEncoder::new(
  b.schema,
  pk_indices,
  DateHandlingMode::FromCe,
  TimestampHandlingMode::Milli,
  timestamptz_mode,
  TimeHandlingMode::Milli,
+ encode_jsonb_mode,
  );
  let encoder = if let Some(s) = b.format_desc.options.get("schemas.enable") {
  match s.to_lowercase().parse::<bool>() {

src/connector/src/sink/kafka.rs

@@ -590,7 +590,7 @@ mod test {
 
  use super::*;
  use crate::sink::encoder::{
- DateHandlingMode, JsonEncoder, TimeHandlingMode, TimestampHandlingMode,
+ DateHandlingMode, EncodeJsonbMode, JsonEncoder, TimeHandlingMode, TimestampHandlingMode,
  TimestamptzHandlingMode,
  };
  use crate::sink::formatter::AppendOnlyFormatter;
@@ -778,6 +778,7 @@ mod test {
  TimestampHandlingMode::Milli,
  TimestamptzHandlingMode::UtcString,
  TimeHandlingMode::Milli,
+ EncodeJsonbMode::String,
  ),
  )),
  )

src/connector/src/sink/mqtt.rs

@@ -31,8 +31,8 @@ use with_options::WithOptions;
 
 use super::catalog::{SinkEncode, SinkFormat, SinkFormatDesc};
 use super::encoder::{
- DateHandlingMode, JsonEncoder, ProtoEncoder, ProtoHeader, RowEncoder, SerTo, TimeHandlingMode,
- TimestampHandlingMode, TimestamptzHandlingMode,
+ DateHandlingMode, EncodeJsonbMode, JsonEncoder, ProtoEncoder, ProtoHeader, RowEncoder, SerTo,
+ TimeHandlingMode, TimestampHandlingMode, TimestamptzHandlingMode,
 };
 use super::writer::AsyncTruncateSinkWriterExt;
 use super::{DummySinkCommitCoordinator, SinkWriterParam};
@@ -221,7 +221,7 @@ impl MqttSinkWriter {
  }
 
  let timestamptz_mode = TimestamptzHandlingMode::from_options(&format_desc.options)?;
-
+ let encode_jsonb_mode = EncodeJsonbMode::from_options(&format_desc.options)?;
  let encoder = match format_desc.format {
  SinkFormat::AppendOnly => match format_desc.encode {
  SinkEncode::Json => RowEncoderWrapper::Json(JsonEncoder::new(
@@ -231,6 +231,7 @@ impl MqttSinkWriter {
  TimestampHandlingMode::Milli,
  timestamptz_mode,
  TimeHandlingMode::Milli,
+ encode_jsonb_mode,
  )),
  SinkEncode::Protobuf => {
  let (descriptor, sid) = crate::schema::protobuf::fetch_descriptor(

src/connector/src/sink/nats.rs

@@ -28,7 +28,9 @@ use tokio_retry::strategy::{jitter, ExponentialBackoff};
 use tokio_retry::Retry;
 use with_options::WithOptions;
 
-use super::encoder::{DateHandlingMode, TimeHandlingMode, TimestamptzHandlingMode};
+use super::encoder::{
+ DateHandlingMode, EncodeJsonbMode, TimeHandlingMode, TimestamptzHandlingMode,
+};
 use super::utils::chunk_to_json;
 use super::{DummySinkCommitCoordinator, SinkWriterParam};
 use crate::connector_common::NatsCommon;
@@ -151,6 +153,7 @@ impl NatsSinkWriter {
  TimestampHandlingMode::Milli,
  TimestamptzHandlingMode::UtcWithoutSuffix,
  TimeHandlingMode::Milli,
+ EncodeJsonbMode::String,
  ),
  })
  }

src/connector/src/sink/snowflake.rs

@@ -34,7 +34,8 @@ use uuid::Uuid;
 use with_options::WithOptions;
 
 use super::encoder::{
- JsonEncoder, RowEncoder, TimeHandlingMode, TimestampHandlingMode, TimestamptzHandlingMode,
+ EncodeJsonbMode, JsonEncoder, RowEncoder, TimeHandlingMode, TimestampHandlingMode,
+ TimestamptzHandlingMode,
 };
 use super::writer::LogSinkerOf;
 use super::{SinkError, SinkParam};
@@ -193,6 +194,7 @@ impl SnowflakeSinkWriter {
  TimestampHandlingMode::String,
  TimestamptzHandlingMode::UtcString,
  TimeHandlingMode::String,
+ EncodeJsonbMode::String,
  ),
  // initial value of `epoch` will be set to 0
  epoch: 0,