The Complete Production SOP: How to Deploy and Run ClickHouse AggregatingMergeTree Materialized Views in Autonomous AI Agents

-- Raw Ingestion Table
CREATE TABLE IF NOT EXISTS default.pageviews_raw (
    event_time DateTime,
    tenant_id UInt32,
    user_id UInt64,
    bytes_sent UInt64
) ENGINE = MergeTree()
ORDER BY (tenant_id, event_time);

-- Rollup Storage Engine
CREATE TABLE IF NOT EXISTS default.pageviews_hourly_mv (
    tenant_id UInt32,
    hour_bucket DateTime,
    unique_users AggregateFunction(uniq, UInt64),
    total_bytes AggregateFunction(sum, UInt64)
) ENGINE = AggregatingMergeTree()
ORDER BY (tenant_id, hour_bucket);

-- Ingestion Pipeline Connector
CREATE MATERIALIZED VIEW IF NOT EXISTS default.pageviews_hourly_view
TO default.pageviews_hourly_mv AS
SELECT
    tenant_id,
    toStartOfHour(event_time) AS hour_bucket,
    uniqState(user_id) AS unique_users,
    sumState(bytes_sent) AS total_bytes
FROM default.pageviews_raw
GROUP BY tenant_id, hour_bucket;

import os
import clickhouse_connect

client = clickhouse_connect.get_client(
    host=os.getenv('CLICKHOUSE_HOST', 'localhost'),
    port=int(os.getenv('CLICKHOUSE_PORT', 8123)),
    username=os.getenv('CLICKHOUSE_USER', 'default'),
    password=os.getenv('CLICKHOUSE_PASSWORD', ''),
    database=os.getenv('CLICKHOUSE_DB', 'default')
)
res = client.command('SELECT version()')
print(f"Connected to ClickHouse version: {res}")

-- Correct Read Path: Merging intermediate aggregate states
SELECT
    tenant_id,
    hour_bucket,
    uniqMerge(unique_users) AS estimated_active_users,
    sumMerge(total_bytes) AS aggregate_bytes
FROM default.pageviews_hourly_mv
WHERE tenant_id = {tenant:UInt32}
  AND hour_bucket >= now() - INTERVAL 7 DAY
GROUP BY tenant_id, hour_bucket
ORDER BY hour_bucket DESC;

import unittest
import clickhouse_connect

class TestAggregatingMergeTree(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.client = clickhouse_connect.get_client(host='localhost', port=8123)
        with open('migrations/001_initial_aggregating_mergetree.sql') as f:
            statements = [s.strip() for s in f.read().split(';') if s.strip()]
            for s in statements:
                cls.client.command(s)

    def test_pipeline_aggregation(self):
        self.client.command("""
            INSERT INTO default.pageviews_raw VALUES 
            (now(), 1, 1001, 512),
            (now(), 1, 1001, 1024),
            (now(), 1, 1002, 2048)
        """)
        result = self.client.query("""
            SELECT 
                tenant_id, 
                uniqMerge(unique_users), 
                sumMerge(total_bytes) 
            FROM default.pageviews_hourly_mv 
            WHERE tenant_id = 1 
            GROUP BY tenant_id
        """).result_rows
        self.assertEqual(len(result), 1)
        self.assertEqual(result[0][1], 2)      # 2 unique users (1001, 1002)
        self.assertEqual(result[0][2], 3584)   # 512 + 1024 + 2048 bytes

if __name__ == '__main__':
    unittest.main()