DuckDB 1.1 vs. Snowflake 2026 OLAP Query Speed Benchmarks on 1TB Parquet

import duckdb
import time
import os
import sys
from typing import List, Dict, Any

def run_duckdb_benchmark(
    s3_endpoint: str = "s3://tpch-parquet-public/1tb/8-partition/",
    query_path: str = "./tpc-h/queries/",
    output_path: str = "./results/duckdb_1_1_results.csv",
    use_local_storage: bool = False
) -> List[Dict[str, Any]]:
    """
    Executes TPC-H benchmark suite against DuckDB 1.1 on 1TB Parquet.

    Args:
        s3_endpoint: S3 path to partitioned Parquet files (if use_local_storage=False)
        query_path: Local path to TPC-H SQL query files (q1.sql to q22.sql)
        output_path: Path to write CSV results
        use_local_storage: If True, read Parquet from local NVMe instead of S3

    Returns:
        List of dicts containing query ID, execution time, row count
    """
    results = []
    conn = None
    try:
        # Initialize DuckDB connection with ARM64 SIMD optimizations enabled
        conn = duckdb.connect(database=":memory:", read_only=False)
        # Enable S3 support if not using local storage
        if not use_local_storage:
            conn.execute("INSTALL httpfs; LOAD httpfs;")
            conn.execute("SET s3_region='us-east-1';")
            # Use anonymous access for public TPC-H bucket (replace with your credentials)
            conn.execute("SET s3_access_key_id='';")
            conn.execute("SET s3_secret_access_key='';")
            parquet_path = s3_endpoint
        else:
            parquet_path = "/mnt/nvme/tpch-1tb-parquet/8-partition/"

        # Register Parquet scan as a view for easier querying
        conn.execute(f"""
            CREATE VIEW tpch_lineitem AS 
            SELECT * FROM read_parquet('{parquet_path}lineitem/*.parquet', 
            hive_partitioning=true, union_by_name=true);
        """)
        conn.execute(f"""
            CREATE VIEW tpch_orders AS 
            SELECT * FROM read_parquet('{parquet_path}orders/*.parquet', 
            hive_partitioning=true, union_by_name=true);
        """)
        # Register other TPC-H tables (customer, nation, region, supplier, partsupp, part)
        for table in ["customer", "nation", "region", "supplier", "partsupp", "part"]:
            conn.execute(f"""
                CREATE VIEW tpch_{table} AS 
                SELECT * FROM read_parquet('{parquet_path}{table}/*.parquet', 
                hive_partitioning=true, union_by_name=true);
            """)

        # Execute each TPC-H query 3 times, average results
        for q_num in range(1, 23):
            q_path = f"{query_path}q{q_num}.sql"
            if not os.path.exists(q_path):
                print(f"Warning: Query file {q_path} not found, skipping.")
                continue

            with open(q_path, "r") as f:
                query_sql = f.read().replace("tpch.", "tpch_")

            # Warm-up run (discard results)
            try:
                conn.execute(query_sql).fetchall()
            except Exception as e:
                print(f"Warm-up failed for Q{q_num}: {str(e)}")
                continue

            # 3 timed runs
            run_times = []
            row_counts = []
            for _ in range(3):
                start = time.perf_counter()
                try:
                    res = conn.execute(query_sql).fetchall()
                    end = time.perf_counter()
                    run_times.append(end - start)
                    row_counts.append(len(res))
                except Exception as e:
                    print(f"Query Q{q_num} failed: {str(e)}")
                    run_times.append(None)
                    row_counts.append(None)

            # Calculate average (ignore None values)
            valid_times = [t for t in run_times if t is not None]
            avg_time = sum(valid_times) / len(valid_times) if valid_times else None
            avg_rows = sum(row_counts) / len(row_counts) if row_counts else None

            results.append({
                "query_id": f"Q{q_num}",
                "avg_execution_s": round(avg_time, 2) if avg_time else None,
                "avg_row_count": round(avg_rows, 0) if avg_rows else None,
                "engine": "DuckDB 1.1",
                "hardware": "c7g.24xlarge"
            })
            print(f"Completed Q{q_num}: Avg {avg_time:.2f}s")

        # Export results to CSV
        if results:
            import pandas as pd
            pd.DataFrame(results).to_csv(output_path, index=False)
            print(f"Results exported to {output_path}")

        return results

    except Exception as e:
        print(f"Fatal error in DuckDB benchmark: {str(e)}")
        sys.exit(1)
    finally:
        if conn:
            conn.close()

if __name__ == "__main__":
    # Configure for your environment
    run_duckdb_benchmark(
        s3_endpoint="s3://tpch-parquet-public/1tb/8-partition/",
        query_path="./tpc-h-queries/",
        output_path="./results/duckdb_1_1_tpch_1tb.csv",
        use_local_storage=False
    )

-- Snowflake 2026 TPC-H 1TB Parquet Benchmark Script
-- Execute in Snowflake Worksheets (2026.01 release, ARM64 compute tier)
-- Prerequisites: 
-- 1. Create database TPCH_1TB
-- 2. Create external stage pointing to s3://tpch-parquet-public/1tb/8-partition/ (us-east-1)
-- 3. Create warehouse BENCH_WH (96 cores, ARM64, auto_suspend=60)

-- Set session parameters
SET QUERY_TAG = 'TPCH_1TB_DUCKDB_COMPARISON';
SET USE_CACHED_RESULT = FALSE;
SET AUTOCOMMIT = FALSE;

-- Create results table
CREATE OR REPLACE TABLE TPCH_1TB.BENCHMARK_RESULTS (
    query_id VARCHAR(10),
    execution_time_s FLOAT,
    row_count INTEGER,
    engine VARCHAR(50),
    hardware VARCHAR(50),
    run_date TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

-- Register external tables for TPC-H Parquet files
CREATE OR REPLACE EXTERNAL TABLE TPCH_1TB.EXT_LINEITEM (
    L_ORDERKEY BIGINT AS (VALUE:L_ORDERKEY::BIGINT),
    L_PARTKEY BIGINT AS (VALUE:L_PARTKEY::BIGINT),
    L_SUPPKEY BIGINT AS (VALUE:L_SUPPKEY::BIGINT),
    L_LINENUMBER INTEGER AS (VALUE:L_LINENUMBER::INTEGER),
    L_QUANTITY DECIMAL(15,2) AS (VALUE:L_QUANTITY::DECIMAL(15,2)),
    L_EXTENDEDPRICE DECIMAL(15,2) AS (VALUE:L_EXTENDEDPRICE::DECIMAL(15,2)),
    L_DISCOUNT DECIMAL(15,2) AS (VALUE:L_DISCOUNT::DECIMAL(15,2)),
    L_TAX DECIMAL(15,2) AS (VALUE:L_TAX::DECIMAL(15,2)),
    L_RETURNFLAG VARCHAR(1) AS (VALUE:L_RETURNFLAG::VARCHAR(1)),
    L_LINESTATUS VARCHAR(1) AS (VALUE:L_LINESTATUS::VARCHAR(1)),
    L_SHIPDATE DATE AS (VALUE:L_SHIPDATE::DATE),
    L_COMMITDATE DATE AS (VALUE:L_COMMITDATE::DATE),
    L_RECEIPTDATE DATE AS (VALUE:L_RECEIPTDATE::DATE),
    L_SHIPINSTRUCT VARCHAR(25) AS (VALUE:L_SHIPINSTRUCT::VARCHAR(25)),
    L_SHIPMODE VARCHAR(10) AS (VALUE:L_SHIPMODE::VARCHAR(10)),
    L_COMMENT VARCHAR(44) AS (VALUE:L_COMMENT::VARCHAR(44))
)
WITH LOCATION = @TPCH_STAGE/lineitem/
FILE_FORMAT = (TYPE = PARQUET);

-- Repeat external table creation for all TPC-H tables (orders, customer, nation, region, supplier, partsupp, part)
CREATE OR REPLACE EXTERNAL TABLE TPCH_1TB.EXT_ORDERS (
    O_ORDERKEY BIGINT AS (VALUE:O_ORDERKEY::BIGINT),
    O_CUSTKEY BIGINT AS (VALUE:O_CUSTKEY::BIGINT),
    O_ORDERSTATUS VARCHAR(1) AS (VALUE:O_ORDERSTATUS::VARCHAR(1)),
    O_TOTALPRICE DECIMAL(15,2) AS (VALUE:O_TOTALPRICE::DECIMAL(15,2)),
    O_ORDERDATE DATE AS (VALUE:O_ORDERDATE::DATE),
    O_ORDERPRIORITY VARCHAR(15) AS (VALUE:O_ORDERPRIORITY::VARCHAR(15)),
    O_CLERK VARCHAR(15) AS (VALUE:O_CLERK::VARCHAR(15)),
    O_SHIPPRIORITY INTEGER AS (VALUE:O_SHIPPRIORITY::INTEGER),
    O_COMMENT VARCHAR(79) AS (VALUE:O_COMMENT::VARCHAR(79))
)
WITH LOCATION = @TPCH_STAGE/orders/
FILE_FORMAT = (TYPE = PARQUET);

-- Execute TPC-H Q1
DECLARE
    q1_start TIMESTAMP_NTZ;
    q1_end TIMESTAMP_NTZ;
    q1_rows INTEGER;
BEGIN
    q1_start = CURRENT_TIMESTAMP();
    SELECT COUNT(*) INTO q1_rows FROM (
        SELECT L_RETURNFLAG, L_LINESTATUS, SUM(L_QUANTITY) AS SUM_QTY, SUM(L_EXTENDEDPRICE) AS SUM_BASE_PRICE, SUM(L_EXTENDEDPRICE * (1 - L_DISCOUNT)) AS SUM_DISC_PRICE, SUM(L_EXTENDEDPRICE * (1 - L_DISCOUNT) * (1 + L_TAX)) AS SUM_CHARGE, AVG(L_QUANTITY) AS AVG_QTY, AVG(L_EXTENDEDPRICE) AS AVG_PRICE, AVG(L_DISCOUNT) AS AVG_DISC, COUNT(*) AS COUNT_ORDER FROM TPCH_1TB.EXT_LINEITEM WHERE L_SHIPDATE <= DATE('1998-12-01') GROUP BY L_RETURNFLAG, L_LINESTATUS ORDER BY L_RETURNFLAG, L_LINESTATUS
    );
    q1_end = CURRENT_TIMESTAMP();
    INSERT INTO TPCH_1TB.BENCHMARK_RESULTS (query_id, execution_time_s, row_count, engine, hardware)
    VALUES ('Q1', DATEDIFF(SECOND, q1_start, q1_end), q1_rows, 'Snowflake 2026', 'c7g.24xlarge');
    COMMIT;
END;

-- Execute TPC-H Q19
DECLARE
    q19_start TIMESTAMP_NTZ;
    q19_end TIMESTAMP_NTZ;
    q19_rows INTEGER;
BEGIN
    q19_start = CURRENT_TIMESTAMP();
    SELECT COUNT(*) INTO q19_rows FROM (
        SELECT SUM(L_EXTENDEDPRICE * (1 - L_DISCOUNT)) AS REVENUE FROM TPCH_1TB.EXT_LINEITEM L, TPCH_1TB.EXT_PART P, TPCH_1TB.EXT_SUPPLIER S, TPCH_1TB.EXT_PARTSUPP PS, TPCH_1TB.EXT_ORDERS O, TPCH_1TB.EXT_CUSTOMER C, TPCH_1TB.EXT_NATION N1, TPCH_1TB.EXT_NATION N2, TPCH_1TB.EXT_REGION R WHERE P.P_PARTKEY = L.L_PARTKEY AND S.S_SUPPKEY = L.L_SUPPKEY AND PS.PS_PARTKEY = L.L_PARTKEY AND PS.PS_SUPPKEY = L.L_SUPPKEY AND O.O_ORDERKEY = L.L_ORDERKEY AND C.C_CUSTKEY = O.O_CUSTKEY AND N1.N_NATIONKEY = C.C_NATIONKEY AND N2.N_NATIONKEY = S.S_NATIONKEY AND R.R_REGIONKEY = N1.N_REGIONKEY AND P.P_SIZE = 15 AND P.P_TYPE LIKE '%BRASS' AND R.R_NAME = 'EUROPE' AND L.L_SHIPMODE IN ('AIR', 'RAIL') AND L.L_QUANTITY < (SELECT 0.2 * AVG(L2.L_QUANTITY) FROM TPCH_1TB.EXT_LINEITEM L2 WHERE L2.L_PARTKEY = P.P_PARTKEY)
    );
    q19_end = CURRENT_TIMESTAMP();
    INSERT INTO TPCH_1TB.BENCHMARK_RESULTS (query_id, execution_time_s, row_count, engine, hardware)
    VALUES ('Q19', DATEDIFF(SECOND, q19_start, q19_end), q19_rows, 'Snowflake 2026', 'c7g.24xlarge');
    COMMIT;
END;

-- Export results to S3
COPY INTO s3://benchmark-results/snowflake_2026_tpch_1tb.csv
FROM TPCH_1TB.BENCHMARK_RESULTS
FILE_FORMAT = (TYPE = CSV FIELD_OPTIONALLY_ENCLOSED_BY = '"')
HEADER = TRUE;

import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import os
import sys
from typing import Tuple, List

def analyze_benchmark_results(
    duckdb_results_path: str = "./results/duckdb_1_1_tpch_1tb.csv",
    snowflake_results_path: str = "./results/snowflake_2026_tpch_1tb.csv",
    output_dir: str = "./analysis/"
) -> Tuple[pd.DataFrame, List[str]]:
    """
    Analyzes and compares DuckDB 1.1 and Snowflake 2026 TPC-H 1TB benchmark results.

    Args:
        duckdb_results_path: Path to DuckDB benchmark CSV
        snowflake_results_path: Path to Snowflake benchmark CSV
        output_dir: Directory to write comparison tables and plots

    Returns:
        Tuple of merged results DataFrame and list of slow queries (DuckDB > Snowflake)
    """
    try:
        # Load results
        if not os.path.exists(duckdb_results_path):
            raise FileNotFoundError(f"DuckDB results not found at {duckdb_results_path}")
        if not os.path.exists(snowflake_results_path):
            raise FileNotFoundError(f"Snowflake results not found at {snowflake_results_path}")

        duckdb_df = pd.read_csv(duckdb_results_path)
        snowflake_df = pd.read_csv(snowflake_results_path)

        # Validate required columns
        required_cols = ["query_id", "avg_execution_s", "engine"]
        for df, name in [(duckdb_df, "DuckDB"), (snowflake_df, "Snowflake")]:
            missing = [col for col in required_cols if col not in df.columns]
            if missing:
                raise ValueError(f"{name} results missing columns: {missing}")

        # Merge results on query_id
        merged = pd.merge(
            duckdb_df[["query_id", "avg_execution_s"]].rename(columns={"avg_execution_s": "duckdb_s"}),
            snowflake_df[["query_id", "avg_execution_s"]].rename(columns={"avg_execution_s": "snowflake_s"}),
            on="query_id"
        )

        # Calculate speedup (Snowflake time / DuckDB time)
        merged["duckdb_speedup"] = merged["snowflake_s"] / merged["duckdb_s"]
        merged["duckdb_speedup"] = merged["duckdb_speedup"].round(2)

        # Identify slow queries (DuckDB slower than Snowflake)
        slow_queries = merged[merged["duckdb_s"] > merged["snowflake_s"]]["query_id"].tolist()

        # Create output directory
        os.makedirs(output_dir, exist_ok=True)

        # Write comparison CSV
        comparison_path = os.path.join(output_dir, "tpch_1tb_comparison.csv")
        merged.to_csv(comparison_path, index=False)
        print(f"Comparison table written to {comparison_path}")

        # Generate bar plot
        plt.figure(figsize=(16, 8))
        bar_width = 0.35
        x = range(len(merged))
        plt.bar(x, merged["duckdb_s"], bar_width, label="DuckDB 1.1", color="#FFD700")
        plt.bar([i + bar_width for i in x], merged["snowflake_s"], bar_width, label="Snowflake 2026", color="#00BFFF")
        plt.xlabel("TPC-H Query ID")
        plt.ylabel("Execution Time (seconds)")
        plt.title("DuckDB 1.1 vs Snowflake 2026: 1TB Parquet TPC-H Execution Time")
        plt.xticks([i + bar_width/2 for i in x], merged["query_id"])
        plt.legend()
        plt.tight_layout()
        plot_path = os.path.join(output_dir, "tpch_1tb_execution_times.png")
        plt.savefig(plot_path, dpi=300)
        print(f"Plot saved to {plot_path}")

        # Print summary statistics
        print("\n=== Benchmark Summary ===")
        print(f"DuckDB 1.1 Average Query Time: {merged['duckdb_s'].mean():.2f}s")
        print(f"Snowflake 2026 Average Query Time: {merged['snowflake_s'].mean():.2f}s")
        print(f"DuckDB Speedup (Avg): {merged['duckdb_speedup'].mean():.2f}x")
        print(f"Slowest Query (DuckDB): {merged.loc[merged['duckdb_s'].idxmax(), 'query_id']} ({merged['duckdb_s'].max():.2f}s)")
        print(f"Slowest Query (Snowflake): {merged.loc[merged['snowflake_s'].idxmax(), 'query_id']} ({merged['snowflake_s'].max():.2f}s)")

        return merged, slow_queries

    except FileNotFoundError as e:
        print(f"File error: {str(e)}")
        sys.exit(1)
    except ValueError as e:
        print(f"Data validation error: {str(e)}")
        sys.exit(1)
    except Exception as e:
        print(f"Unexpected error: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    # Execute analysis
    merged_results, slow_qs = analyze_benchmark_results(
        duckdb_results_path="./results/duckdb_1_1_tpch_1tb.csv",
        snowflake_results_path="./results/snowflake_2026_tpch_1tb.csv"
    )
    if slow_qs:
        print(f"\nQueries where DuckDB was slower than Snowflake: {slow_qs}")
    else:
        print("\nDuckDB was faster than or equal to Snowflake on all TPC-H queries.")

-- DuckDB 1.1: Create partitioned Parquet from existing CSV
COPY (SELECT * FROM read_csv('lineitem.csv'))
TO 's3://my-bucket/lineitem/'
(FORMAT PARQUET, PARTITION_BY (L_SHIPDATE), COMPRESSION SNAPPY);

-- Snowflake 2026: Create optimized ARM64 warehouse for Parquet
CREATE WAREHOUSE TPCH_PARQUET_WH
WITH 
    WAREHOUSE_SIZE = 'XLarge' -- 96 cores ARM64
    AUTO_SUSPEND = 60
    AUTO_RESUME = TRUE
    QUERY_ACCELERATION_SERVICE = ('ENABLED', 'MAX_SCALE_FACTOR=2');

# GitHub Actions workflow to run DuckDB 1.1 benchmark
name: Run DuckDB 1.1 1TB Parquet Benchmark
on: [workflow_dispatch]
jobs:
  benchmark:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - name: Install DuckDB
        run: pip install duckdb==1.1.0
      - name: Run Benchmark
        run: python run_duckdb_benchmark.py
      - name: Upload Results
        uses: actions/upload-artifact@v4
        with:
          name: duckdb-results
          path: ./results/