-- 0 Query topic information imported to ClickHouse in ckafka
-- 1 Create a Kafka engine table in ClickHouse
CREATE TABLE kafka_table_qiang_test_1 ON CLUSTER default_cluster (
    key String,
    value String
) ENGINE = Kafka
SETTINGS
    kafka_broker_list = '10.0.x.x:9092',
    kafka_topic_list = 'qiang-test-mysql',
    kafka_group_name = 'cg_qiang-test-mysql_1',
    kafka_format = 'JSONEachRow',  -- Select an appropriate format based on your data format, such as JSON, Avro, or CSV.
    kafka_num_consumers = 1,  -- Adjust as needed.
    kafka_max_block_size = 65536,
    kafka_skip_broken_messages = 0; -- The default value is 0.

-- 2 Create a target table for storing data.
-- The Kafka engine table itself only consumes data and does not persist it. Therefore, you need to create a table for actually storing data.
CREATE TABLE real_table_qiang_test_1 ON CLUSTER default_cluster (
    key String,
    value String
) ENGINE = MergeTree()
ORDER BY key;

-- 3 Create a Materialized View to import data from the Kafka table into the actual table.
CREATE MATERIALIZED VIEW view_to_real_table_qiang_test_1 ON CLUSTER default_cluster  TO real_table_qiang_test_1 AS
SELECT
    key,
    value
FROM kafka_table_qiang_test_1;

-- 4 Query data in ClickHouse when you start sending data to kafka.
SELECT * FROM real_table_qiang_test_1;
select count(1) from INFORMATION_SCHEMA.real_table_qiang_test_1 limit 5; 


-- 5 To pause the materialized view and stop data collection
DROP VIEW IF EXISTS view_to_real_table_qiang_test_1 ON CLUSTER default_cluster;

SELECT name FROM system.tables WHERE name LIKE '%view_to_real_table_qiang_test_1%'; -- Verify that the deletion was successful.