Why ClickHouseSinkConnector with Debezium PostgresConnector can not sink for JSONB to Nested field in Clickhouse?

[phongbui81062]

Hi I get this issue from Kafka connect sink:
when I flatten WAL from Kafka, fields mapping become before.* and after.*
The issue is if I have a Jsonb column in Postgres, they can not sink to ClickHouse because before or after can be null when insert or delete occur in Postgres and Nested type can not insert null into it?
For example:
I have this table in Postgres

create table test
(
    id           bigserial
        primary key,
    status       varchar(16) not null,
    items        jsonb       not null
);

this table in Clickhouse

create or replace table testing.`demo_db.public.test`(
    `before.id`           Nullable(Int64),
    `before.items` Nested(guid UUID,
        product_name String,
        product_id Nullable(String)),
    `after.id`           Nullable(Int64),
    `after.status`       Nullable(String),
    `after.items` Nested(guid UUID,
        product_name String,
        product_id Nullable(String)),
    ts_ms Int64,
    op    LowCardinality(String)
)
engine MergeTree
order by tuple();

and this is kafka connect sink clickhouse config:

{
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "topics": "demo_db.public.shipment_line_item",
    "ssl": "false",
    "security.protocol": "SASL_PLAINTEXT",
    "behavior.on.null.values": "ignore",
    "behavior.on.error": "ignore",
    "hostname": "clickhouse-server",
    "database": "testing",
    "port": "8123",
    "transforms": "flatten",
    "transforms.flatten.type": "org.apache.kafka.connect.transforms.Flatten$Value",
    "transforms.flatten.delimiter": ".",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081/",
    "value.converter.schemas.enable": "true",
    "exactlyOnce": "true",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "deadletterqueue",
    "errors.deadletterqueue.topic.replication.factor": "1",
    "errors.deadletterqueue.context.headers.enable": "true",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true",
    "tasks.max": "5",
    "user": "default",
    "username": "default",
    "schemas.enable": "true"
}

When I try to insert new data in Postgres, they can not sink to Clickhouse because Clickhouse cannot insert null into Nested field. Any suggestion for me?
Thanks

[mzitnik]

As a quick aid, can you flatten those fields?
For example before.items.product_name and before.items.product_id maybe using
If you are getting an error from Connect, please open an issue with detailed info so we can reproduce and fix it.

[phongbui81062]

Thanks for this suggestion, I'll try it first