ClickHouse 24.3 vs. Snowflake 2026: OLAP Query Speed for 2026 Real-Time Dashboards

import os
import sys
import time
import logging
from clickhouse_driver import Client
from typing import List, Dict, Optional

# Configure logging for benchmark reproducibility
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

# Benchmark configuration (matches Snowflake 2026 test env)
CLICKHOUSE_HOST = "ch-benchmark-node-01.internal"
CLICKHOUSE_PORT = 9000
CLICKHOUSE_USER = "benchmark_user"
CLICKHOUSE_PASSWORD = os.getenv("CH_PASSWORD", "")
TPCDS_SCALE_FACTOR = 10000  # 10TB dataset
BENCHMARK_ITERATIONS = 10
QUERIES_TO_RUN = [1, 3, 7, 19, 42]  # TPC-DS query IDs for dashboard-relevant workloads

def init_clickhouse_client() -> Optional[Client]:
    """Initialize ClickHouse client with error handling and connection retry."""
    retry_count = 0
    max_retries = 3
    while retry_count < max_retries:
        try:
            client = Client(
                host=CLICKHOUSE_HOST,
                port=CLICKHOUSE_PORT,
                user=CLICKHOUSE_USER,
                password=CLICKHOUSE_PASSWORD,
                connect_timeout=10
            )
            # Test connection with simple query
            client.execute("SELECT 1")
            logger.info(f"Connected to ClickHouse {client.execute('SELECT version()')[0][0]}")
            return client
        except Exception as e:
            retry_count += 1
            logger.warning(f"Connection attempt {retry_count} failed: {str(e)}")
            time.sleep(2 ** retry_count)
    logger.error("Failed to connect to ClickHouse after max retries")
    return None

def load_tpcds_data(client: Client) -> bool:
    """Load 10TB TPC-DS data into ClickHouse with schema validation."""
    try:
        # Create TPC-DS schema (simplified for example, full schema in https://github.com/ClickHouse/ClickHouse/tree/master/tests/tpcds)
        client.execute("""
            CREATE DATABASE IF NOT EXISTS tpcds_10tb
            ENGINE = Atomic
        """)
        # Check if data already loaded
        row_count = client.execute("SELECT count() FROM tpcds_10tb.store_sales LIMIT 1")[0][0]
        if row_count > 0:
            logger.info(f"TPC-DS data already loaded: {row_count} store_sales rows")
            return True
        # Load data from S3 (matches Snowflake 2026 test dataset)
        logger.info("Loading 10TB TPC-DS data from S3...")
        client.execute("""
            CREATE TABLE tpcds_10tb.store_sales
            (
                ss_sold_date_sk UInt32,
                ss_sold_time_sk UInt32,
                ss_item_sk UInt32,
                ss_customer_sk UInt32,
                ss_cdemo_sk UInt32,
                ss_hdemo_sk UInt32,
                ss_addr_sk UInt32,
                ss_store_sk UInt32,
                ss_promo_sk UInt32,
                ss_ticket_number UInt32,
                ss_quantity UInt16,
                ss_wholesale_cost Decimal(7,2),
                ss_list_price Decimal(7,2),
                ss_sales_price Decimal(7,2),
                ss_ext_discount_amt Decimal(7,2),
                ss_ext_sales_price Decimal(7,2),
                ss_ext_wholesale_cost Decimal(7,2),
                ss_ext_list_price Decimal(7,2),
                ss_ext_tax Decimal(7,2),
                ss_coupon_amt Decimal(7,2),
                ss_net_paid Decimal(7,2),
                ss_net_paid_inc_tax Decimal(7,2),
                ss_net_profit Decimal(7,2)
            )
            ENGINE = MergeTree
            ORDER BY (ss_sold_date_sk, ss_item_sk)
            PARTITION BY toYYYYMM(toDate(ss_sold_date_sk))
        """)
        # Load data from public S3 bucket (reproducible benchmark dataset)
        client.execute("""
            INSERT INTO tpcds_10tb.store_sales
            SELECT * FROM s3(
                'https://tpcds-benchmark-datasets.s3.amazonaws.com/10tb/store_sales.parquet',
                'Parquet',
                'ss_sold_date_sk UInt32, ss_sold_time_sk UInt32, ss_item_sk UInt32, ss_customer_sk UInt32, ss_cdemo_sk UInt32, ss_hdemo_sk UInt32, ss_addr_sk UInt32, ss_store_sk UInt32, ss_promo_sk UInt32, ss_ticket_number UInt32, ss_quantity UInt16, ss_wholesale_cost Decimal(7,2), ss_list_price Decimal(7,2), ss_sales_price Decimal(7,2), ss_ext_discount_amt Decimal(7,2), ss_ext_sales_price Decimal(7,2), ss_ext_wholesale_cost Decimal(7,2), ss_ext_list_price Decimal(7,2), ss_ext_tax Decimal(7,2), ss_coupon_amt Decimal(7,2), ss_net_paid Decimal(7,2), ss_net_paid_inc_tax Decimal(7,2), ss_net_profit Decimal(7,2)'
            )
        """)
        logger.info("TPC-DS data load complete")
        return True
    except Exception as e:
        logger.error(f"Failed to load TPC-DS data: {str(e)}")
        return False

if __name__ == "__main__":
    client = init_clickhouse_client()
    if not client:
        sys.exit(1)
    if not load_tpcds_data(client):
        sys.exit(1)
    logger.info("Benchmark setup complete for ClickHouse 24.3")

import os
import sys
import time
import logging
from snowflake.connector import connect, Error
from typing import Optional

# Configure logging to match ClickHouse benchmark config
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

# Snowflake 2026 test environment config (matches ClickHouse node specs)
SNOWFLAKE_ACCOUNT = "xy12345.us-east-1"
SNOWFLAKE_USER = "benchmark_user"
SNOWFLAKE_PASSWORD = os.getenv("SNOWFLAKE_PASSWORD", "")
SNOWFLAKE_WAREHOUSE = "BENCHMARK_WH_2026"
SNOWFLAKE_DATABASE = "TPCDS_10TB"
SNOWFLAKE_SCHEMA = "PUBLIC"
TPCDS_SCALE_FACTOR = 10000  # 10TB dataset
BENCHMARK_ITERATIONS = 10
QUERIES_TO_RUN = [1, 3, 7, 19, 42]  # Matching TPC-DS query IDs

def init_snowflake_connection() -> Optional[object]:
    """Initialize Snowflake connection with retry logic and version check."""
    retry_count = 0
    max_retries = 3
    while retry_count < max_retries:
        try:
            conn = connect(
                account=SNOWFLAKE_ACCOUNT,
                user=SNOWFLAKE_USER,
                password=SNOWFLAKE_PASSWORD,
                warehouse=SNOWFLAKE_WAREHOUSE,
                database=SNOWFLAKE_DATABASE,
                schema=SNOWFLAKE_SCHEMA,
                login_timeout=10
            )
            # Verify Snowflake version matches 2026.0.1 preview
            cursor = conn.cursor()
            cursor.execute("SELECT CURRENT_VERSION()")
            version = cursor.fetchone()[0]
            logger.info(f"Connected to Snowflake version: {version}")
            if not version.startswith("2026.0.1"):
                logger.warning(f"Expected Snowflake 2026.0.1, got {version}")
            return conn
        except Error as e:
            retry_count += 1
            logger.warning(f"Snowflake connection attempt {retry_count} failed: {str(e)}")
            time.sleep(2 ** retry_count)
        except Exception as e:
            logger.error(f"Unexpected error connecting to Snowflake: {str(e)}")
            return None
    logger.error("Failed to connect to Snowflake after max retries")
    return None

def load_tpcds_data(conn: object) -> bool:
    """Load 10TB TPC-DS data into Snowflake, matching ClickHouse dataset."""
    try:
        cursor = conn.cursor()
        # Check if data already exists
        cursor.execute("SELECT COUNT(*) FROM store_sales")
        row_count = cursor.fetchone()[0]
        if row_count > 0:
            logger.info(f"TPC-DS data already loaded: {row_count} store_sales rows")
            return True
        # Create TPC-DS schema (matching ClickHouse column types)
        logger.info("Creating TPC-DS schema in Snowflake 2026...")
        cursor.execute("""
            CREATE DATABASE IF NOT EXISTS TPCDS_10TB
        """)
        cursor.execute("""
            CREATE SCHEMA IF NOT EXISTS TPCDS_10TB.PUBLIC
        """)
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS store_sales (
                ss_sold_date_sk NUMBER(38,0),
                ss_sold_time_sk NUMBER(38,0),
                ss_item_sk NUMBER(38,0),
                ss_customer_sk NUMBER(38,0),
                ss_cdemo_sk NUMBER(38,0),
                ss_hdemo_sk NUMBER(38,0),
                ss_addr_sk NUMBER(38,0),
                ss_store_sk NUMBER(38,0),
                ss_promo_sk NUMBER(38,0),
                ss_ticket_number NUMBER(38,0),
                ss_quantity NUMBER(38,0),
                ss_wholesale_cost DECIMAL(7,2),
                ss_list_price DECIMAL(7,2),
                ss_sales_price DECIMAL(7,2),
                ss_ext_discount_amt DECIMAL(7,2),
                ss_ext_sales_price DECIMAL(7,2),
                ss_ext_wholesale_cost DECIMAL(7,2),
                ss_ext_list_price DECIMAL(7,2),
                ss_ext_tax DECIMAL(7,2),
                ss_coupon_amt DECIMAL(7,2),
                ss_net_paid DECIMAL(7,2),
                ss_net_paid_inc_tax DECIMAL(7,2),
                ss_net_profit DECIMAL(7,2)
            )
        """)
        # Load data from same S3 bucket as ClickHouse benchmark
        logger.info("Loading 10TB TPC-DS data from S3 into Snowflake...")
        cursor.execute("""
            COPY INTO store_sales
            FROM 's3://tpcds-benchmark-datasets/10tb/store_sales.parquet'
            FILE_FORMAT = (TYPE = PARQUET)
            MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE
        """)
        logger.info("TPC-DS data load complete")
        return True
    except Error as e:
        logger.error(f"Failed to load TPC-DS data into Snowflake: {str(e)}")
        return False
    except Exception as e:
        logger.error(f"Unexpected error loading Snowflake data: {str(e)}")
        return False

if __name__ == "__main__":
    conn = init_snowflake_connection()
    if not conn:
        sys.exit(1)
    if not load_tpcds_data(conn):
        sys.exit(1)
    logger.info("Benchmark setup complete for Snowflake 2026")
    conn.close()

import os
import sys
import time
import csv
import logging
from typing import List, Dict, Tuple
from clickhouse_driver import Client
from snowflake.connector import connect, Error

# Shared benchmark config
BENCHMARK_ITERATIONS = 10
OUTPUT_CSV = "olap_benchmark_results_2026.csv"
TPCDS_QUERIES = {
    1: """
        SELECT
            ss_store_sk,
            SUM(ss_sales_price) AS total_sales,
            COUNT(DISTINCT ss_customer_sk) AS unique_customers
        FROM tpcds_10tb.store_sales
        WHERE ss_sold_date_sk BETWEEN 2451180 AND 2451200  # Q1 2026 date range
        GROUP BY ss_store_sk
        ORDER BY total_sales DESC
        LIMIT 100
    """,
    3: """
        SELECT
            ss_item_sk,
            AVG(ss_sales_price) AS avg_price,
            SUM(ss_quantity) AS total_quantity
        FROM tpcds_10tb.store_sales
        WHERE ss_sold_date_sk >= 2451000  # Last 6 months of 2025
        GROUP BY ss_item_sk
        HAVING SUM(ss_quantity) > 1000
        ORDER BY total_quantity DESC
        LIMIT 50
    """,
    7: """
        SELECT
            toDate(ss_sold_date_sk) AS sale_date,
            SUM(ss_ext_sales_price) AS daily_revenue
        FROM tpcds_10tb.store_sales
        WHERE ss_sold_date_sk BETWEEN 2451180 AND 2451240  # Q1 2026
        GROUP BY sale_date
        ORDER BY sale_date
    """,
    19: """
        SELECT
            ss_promo_sk,
            COUNT(*) AS promo_uses,
            SUM(ss_ext_sales_price) AS promo_revenue
        FROM tpcds_10tb.store_sales
        WHERE ss_promo_sk IS NOT NULL
        GROUP BY ss_promo_sk
        ORDER BY promo_revenue DESC
        LIMIT 20
    """,
    42: """
        SELECT
            ss_customer_sk,
            SUM(ss_net_paid_inc_tax) AS total_spent,
            COUNT(DISTINCT ss_ticket_number) AS visit_count
        FROM tpcds_10tb.store_sales
        WHERE ss_sold_date_sk >= 2450900  # Last 9 months
        GROUP BY ss_customer_sk
        HAVING total_spent > 1000
        ORDER BY total_spent DESC
        LIMIT 100
    """
}

def run_clickhouse_benchmark(client: Client) -> List[Dict]:
    """Run TPC-DS queries on ClickHouse 24.3 and collect latency metrics."""
    results = []
    for query_id, query in TPCDS_QUERIES.items():
        latencies = []
        errors = 0
        for i in range(BENCHMARK_ITERATIONS):
            try:
                start = time.perf_counter()
                client.execute(query)
                end = time.perf_counter()
                latency_ms = (end - start) * 1000
                latencies.append(latency_ms)
                logging.info(f"ClickHouse Query {query_id} Iter {i+1}: {latency_ms:.2f}ms")
            except Exception as e:
                errors += 1
                logging.error(f"ClickHouse Query {query_id} failed: {str(e)}")
        if latencies:
            avg_latency = sum(latencies) / len(latencies)
            p99_latency = sorted(latencies)[int(len(latencies)*0.99)]
            results.append({
                "engine": "ClickHouse 24.3",
                "query_id": query_id,
                "avg_latency_ms": avg_latency,
                "p99_latency_ms": p99_latency,
                "error_rate": errors / BENCHMARK_ITERATIONS
            })
    return results

def run_snowflake_benchmark(conn: object) -> List[Dict]:
    """Run TPC-DS queries on Snowflake 2026 and collect latency metrics."""
    results = []
    cursor = conn.cursor()
    for query_id, query in TPCDS_QUERIES.items():
        latencies = []
        errors = 0
        for i in range(BENCHMARK_ITERATIONS):
            try:
                start = time.perf_counter()
                cursor.execute(query)
                cursor.fetchall()  # Ensure full result set is retrieved
                end = time.perf_counter()
                latency_ms = (end - start) * 1000
                latencies.append(latency_ms)
                logging.info(f"Snowflake Query {query_id} Iter {i+1}: {latency_ms:.2f}ms")
            except Error as e:
                errors += 1
                logging.error(f"Snowflake Query {query_id} failed: {str(e)}")
        if latencies:
            avg_latency = sum(latencies) / len(latencies)
            p99_latency = sorted(latencies)[int(len(latencies)*0.99)]
            results.append({
                "engine": "Snowflake 2026",
                "query_id": query_id,
                "avg_latency_ms": avg_latency,
                "p99_latency_ms": p99_latency,
                "error_rate": errors / BENCHMARK_ITERATIONS
            })
    return results

def export_results(clickhouse_results: List[Dict], snowflake_results: List[Dict]):
    """Export benchmark results to CSV for analysis."""
    with open(OUTPUT_CSV, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=["engine", "query_id", "avg_latency_ms", "p99_latency_ms", "error_rate"])
        writer.writeheader()
        for row in clickhouse_results + snowflake_results:
            writer.writerow(row)
    logging.info(f"Results exported to {OUTPUT_CSV}")

if __name__ == "__main__":
    # Initialize clients (config from previous scripts)
    ch_client = Client(host="ch-benchmark-node-01.internal", port=9000, user="benchmark_user", password=os.getenv("CH_PASSWORD"))
    sf_conn = connect(account="xy12345.us-east-1", user="benchmark_user", password=os.getenv("SNOWFLAKE_PASSWORD"), warehouse="BENCHMARK_WH_2026", database="TPCDS_10TB")

    logging.info("Starting ClickHouse 24.3 benchmark...")
    ch_results = run_clickhouse_benchmark(ch_client)

    logging.info("Starting Snowflake 2026 benchmark...")
    sf_results = run_snowflake_benchmark(sf_conn)

    export_results(ch_results, sf_results)
    sf_conn.close()
    logging.info("Benchmark complete")

-- ClickHouse 24.3 materialized view for daily store sales
CREATE MATERIALIZED VIEW tpcds_10tb.daily_store_sales_mv
ENGINE = MergeTree
ORDER BY (sale_date, ss_store_sk)
AS
SELECT
    toDate(ss_sold_date_sk) AS sale_date,
    ss_store_sk,
    SUM(ss_ext_sales_price) AS total_daily_sales,
    COUNT(DISTINCT ss_customer_sk) AS unique_daily_customers,
    SUM(ss_quantity) AS total_daily_quantity
FROM tpcds_10tb.store_sales
GROUP BY sale_date, ss_store_sk

-- Snowflake 2026: Enable Search Optimization on ss_item_sk
ALTER TABLE store_sales
ADD SEARCH OPTIMIZATION ON COLUMN ss_item_sk;

-- Verify SOS is enabled
SHOW SEARCH OPTIMIZATION ON store_sales;

# ClickHouse 24.3 connection pool config (Python)
from clickhouse_driver import Client
from dbutils.pooled_db import PooledDB

pool = PooledDB(
    creator=Client,
    host="ch-benchmark-node-01.internal",
    port=9000,
    user="benchmark_user",
    password=os.getenv("CH_PASSWORD"),
    maxconnections=20,
    mincached=5,
    maxcached=10,
    setsession=["SET max_threads=10"]  # Tune to avoid CPU contention
)

# Get connection from pool
client = pool.connection()