MERGE:
0 rows in set. Elapsed: 14.899 sec. Processed 635.55 million rows, 335.40 GB (42.66 million rows/s., 22.51 GB/s.)

Table1:
0 rows in set. Elapsed: 5.730 sec. Processed 417.29 million rows, 14.05 GB (72.83 million rows/s., 2.45 GB/s.)
Table2:
0 rows in set. Elapsed: 2.943 sec. Processed 218.27 million rows, 6.27 GB (74.17 million rows/s., 2.13 GB/s.)

create database test_db;

-- Таблица, куда пишем данные полученные в очереди
CREATE TABLE IF NOT EXISTS test_db.test_table (
    company_id           UInt32,
    product_id           UInt32,
    event_date           Date DEFAULT toDate(now())
) ENGINE=MergeTree(
     event_date,
     (company_id),
     8192
);

-- Очередь
CREATE TABLE IF NOT EXISTS test_db.queue (
   company_id           UInt32,
   product_id           UInt32,
   event_date           Date DEFAULT toDate(now())
) ENGINE=Kafka() SETTINGS
    kafka_broker_list = 'kafka:9092',
    kafka_topic_list = 'topic8',
    kafka_group_name = 'clickhouse',
    kafka_format = 'Protobuf',
    kafka_schema = 'test.proto:TestPayload';

-- Консьюмер
CREATE MATERIALIZED VIEW consumer TO test_db.test_table
AS (SELECT company_id, product_id, event_date FROM test_db.queue);

--
SELECT * FROM test_db.test_table;

/*Потобаф:
syntax = "proto3";

package promowidgets;

message TestPayload {
  uint32 company_id = 1;
  uint32 product_id = 2;
  uint32 event_date = 3;
}
*/