ClickHouse Kafka引擎表如何在物化视图层面配置字段默认值？

问题场景 #

ClickHouse官方文档明确Kafka表引擎不支持字段默认值，需在物化视图层面处理。按文档创建结构后，发送的JSON消息消费后，目标表未生成预期默认值：

数据表结构 #

create or replace table test.kafka_test1_data_table (
    ts DateTime,
    dt Date DEFAULT toDate(ts),
    url String,
    path String DEFAULT path(url),
    urlPath String DEFAULT cutQueryStringAndFragment(url),
    int_value UInt32 DEFAULT 1
)
ENGINE=MergeTree
ORDER BY dt;

Kafka队列表 #

create or replace table test.kafka_test1_kafka_queue (
    ts DateTime,
    dt Date,
    url String,
    path String,
    urlPath String,
    int_value UInt32
)
ENGINE=Kafka
SETTINGS
    kafka_broker_list='broker:port',
    kafka_topic_list='test_kafka_topc_default_values',
    kafka_group_name='clickhouse-consumer-1',
    kafka_format='JSONEachRow',
    kafka_num_consumers=3;

物化视图 #

CREATE MATERIALIZED VIEW test.kafka_test1_mv TO test.kafka_test1_data_table (
    ts DateTime,
    dt Date DEFAULT toDate(ts),
    url String,
    path String DEFAULT path(url),
    urlPath String DEFAULT cutQueryStringAndFragment(url),
    int_value UInt32 DEFAULT 1
) AS SELECT * FROM test.kafka_test1_kafka_queue;

发送的JSON消息 #

{
    "ts": "2024-07-15 11:37:00",
    "url": "https://host.com/path/to/resource.ext?query=123#fragment",
    "int_value": 0
}

实际查询结果 #

SELECT *
FROM test.kafka_test1_data_table

Query id: 546d75e6-1afc-4ac7-aa89-7b303795dd78

┌──────────────────ts─┬─────────dt─┬─url─────────────────────────────────────────────────────────┬─path─┬─urlPath─┬─int_value─┐
│ 2024-07-15 11:37:00 │ 1970-01-01 │ https://host.com/path/to/resource.ext?query=123#fragment    │      │         │         0 │
└─────────────────────┴────────────┴─────────────────────────────────────────────────────────────┴──────┴─────────┴───────────┘

问题根源 #

Kafka表中未被消息填充的字段会被自动设为对应数据类型的默认值（如Date类型为1970-01-01，String类型为空字符串）。物化视图使用SELECT *时，会直接将这些默认值写入目标表，完全绕过了目标表的默认值计算逻辑。

正确实现方式 #

方法一：仅选择消息中存在的字段，依赖目标表默认值计算 #

修改物化视图，只查询Kafka消息中实际存在的字段，目标表会自动对缺失字段应用定义好的默认值：

-- 删除原有物化视图
DROP MATERIALIZED VIEW IF EXISTS test.kafka_test1_mv;

-- 创建新物化视图
CREATE MATERIALIZED VIEW test.kafka_test1_mv TO test.kafka_test1_data_table
AS SELECT ts, url, int_value FROM test.kafka_test1_kafka_queue;

同时可简化Kafka队列表，去掉不需要从消息读取的字段，避免混淆：

DROP TABLE IF EXISTS test.kafka_test1_kafka_queue;

create or replace table test.kafka_test1_kafka_queue (
    ts DateTime,
    url String,
    int_value UInt32
)
ENGINE=Kafka
SETTINGS
    kafka_broker_list='broker:port',
    kafka_topic_list='test_kafka_topc_default_values',
    kafka_group_name='clickhouse-consumer-1',
    kafka_format='JSONEachRow',
    kafka_num_consumers=3;

方法二：在物化视图中显式计算默认值 #

如果需要更灵活的控制，可在物化视图的SELECT语句中直接计算默认值，再写入目标表：

DROP MATERIALIZED VIEW IF EXISTS test.kafka_test1_mv;

CREATE MATERIALIZED VIEW test.kafka_test1_mv TO test.kafka_test1_data_table
AS SELECT
    ts,
    toDate(ts) AS dt,
    url,
    path(url) AS path,
    cutQueryStringAndFragment(url) AS urlPath,
    -- 若消息未传入int_value则用默认值1，否则用消息中的值
    coalesce(int_value, 1) AS int_value
FROM test.kafka_test1_kafka_queue;

验证结果 #

使用上述任一方法后，重新发送消息，查询目标表将得到预期结果：

┌──────────────────ts─┬─────────dt─┬─url─────────────────────────────────────────────────────────┬─path───────────────────┬─urlPath───────────────────────┬─int_value─┐
│ 2024-07-15 11:37:00 │ 2024-07-15 │ https://host.com/path/to/resource.ext?query=123#fragment    │ /path/to/resource.ext   │ https://host.com/path/to/resource.ext │         0 │
└─────────────────────┴────────────┴─────────────────────────────────────────────────────────────┴─────────────────────────┴───────────────────────────────┴───────────┘

内容的提问来源于stack exchange，提问作者Artur Eshenbrener