CREATE TABLE queue (
         timestamp UInt64,
         level String,
         message String
 ) ENGINE = Kafka
 SETTINGS
                 kafka_broker_list = 'localhost:9092',
                 kafka_topic_list = 'topic',
                 kafka_group_name = 'group',
                 kafka_format = 'JSONEachRow',
                 kafka_num_consumers = 1,
                 kafka_max_block_size = 65536,
                 kafka_skip_broken_messages = 0,
                 kafka_auto_offset_reset = 'latest';

CREATE TABLE daily on cluster default_cluster
(
    day Date,
    level String,
    total UInt64
)
engine = SummingMergeTree()
order by int_id;

create table daily on cluster default_cluster
(
    day Date,
    level String,
    total UInt64
)
engine = ReplicatedSummingMergeTree('/clickhouse/tables/test/test/{shard}', '{replica}')
order by int_id;`

create table daily_dis on cluster default_cluster
AS test.test
engine = Distributed('default_cluster', 'default', 'daily', rand());

CREATE MATERIALIZED VIEW consumer TO daily
AS SELECT toDate(toDateTime(timestamp)) AS day, level, count() as total
FROM queue GROUP BY day, level;

SELECT level, sum(total) FROM daily GROUP BY level;