Hot Take: Data Warehouses Like Snowflake Are Dead in 2026: Switch to DuckDB 1.0 and Local Analytics

import duckdb
import os
import sys
from pathlib import Path
import pandas as pd
from typing import List, Optional
import logging

# Configure logging for error tracking
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class LocalAnalyticsEngine:
    """Wrapper for DuckDB 1.0 local analytics with error handling and query benchmarking"""

    def __init__(self, db_path: Optional[str] = None, in_memory: bool = True):
        """
        Initialize DuckDB connection.
        Args:
            db_path: Path to persistent DuckDB file (None for in-memory)
            in_memory: If True, use in-memory database (ignores db_path)
        """
        self.db_path = db_path
        self.in_memory = in_memory
        self.conn = None
        self._connect()

    def _connect(self) -> None:
        """Establish DuckDB connection with error handling"""
        try:
            if self.in_memory:
                self.conn = duckdb.connect(database=":memory:", read_only=False)
                logger.info("Connected to in-memory DuckDB 1.0 instance")
            else:
                if not Path(self.db_path).exists():
                    logger.warning(f"Database path {self.db_path} does not exist, creating new file")
                self.conn = duckdb.connect(database=self.db_path, read_only=False)
                logger.info(f"Connected to persistent DuckDB instance at {self.db_path}")
        except duckdb.Error as e:
            logger.error(f"Failed to connect to DuckDB: {str(e)}")
            sys.exit(1)

    def load_parquet(self, table_name: str, parquet_paths: List[str]) -> None:
        """
        Load Parquet files into a DuckDB table with validation.
        Args:
            table_name: Name of the target table
            parquet_paths: List of Parquet file paths or glob patterns
        """
        try:
            # Validate all Parquet paths exist
            for path in parquet_paths:
                if "*" not in path and not Path(path).exists():
                    raise FileNotFoundError(f"Parquet path {path} does not exist")

            # Create table from Parquet files
            parquet_glob = ",".join([f"'{p}'" for p in parquet_paths])
            query = f"CREATE OR REPLACE TABLE {table_name} AS SELECT * FROM read_parquet([{parquet_glob}])"
            self.conn.execute(query)
            row_count = self.conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0]
            logger.info(f"Loaded {row_count} rows into table {table_name} from {len(parquet_paths)} Parquet files")
        except FileNotFoundError as e:
            logger.error(f"Parquet load failed: {str(e)}")
            raise
        except duckdb.Error as e:
            logger.error(f"DuckDB error loading Parquet: {str(e)}")
            raise

    def run_aggregate_query(self, query: str) -> pd.DataFrame:
        """
        Run an aggregate query and return results as Pandas DataFrame.
        Args:
            query: ANSI SQL query to execute
        Returns:
            Pandas DataFrame with query results
        """
        try:
            start_time = pd.Timestamp.now()
            result = self.conn.execute(query).fetchdf()
            end_time = pd.Timestamp.now()
            duration = (end_time - start_time).total_seconds()
            logger.info(f"Query executed in {duration:.2f} seconds, returned {len(result)} rows")
            return result
        except duckdb.Error as e:
            logger.error(f"Query execution failed: {str(e)}")
            raise

    def close(self) -> None:
        """Close DuckDB connection"""
        if self.conn:
            self.conn.close()
            logger.info("DuckDB connection closed")

if __name__ == "__main__":
    # Example usage: Analyze e-commerce clickstream data
    try:
        engine = LocalAnalyticsEngine(in_memory=True)

        # Load sample clickstream Parquet data (replace with real paths)
        # For demo, we'll create a sample Parquet file first
        sample_data = pd.DataFrame({
            "user_id": range(1, 100001),
            "session_id": [f"sess_{i//10}" for i in range(100000)],
            "event_type": ["click", "view", "purchase"] * 33334,
            "timestamp": pd.date_range("2026-01-01", periods=100000, freq="s"),
            "revenue": [0.0, 0.0, 49.99] * 33334
        })
        sample_path = "sample_clickstream.parquet"
        sample_data.to_parquet(sample_path)

        engine.load_parquet("clickstream", [sample_path])

        # Run aggregate query: Monthly revenue and event counts
        agg_query = """
        SELECT 
            DATE_TRUNC('month', timestamp) AS month,
            event_type,
            COUNT(*) AS event_count,
            SUM(revenue) AS total_revenue
        FROM clickstream
        GROUP BY month, event_type
        ORDER BY month, event_type
        """
        results = engine.run_aggregate_query(agg_query)
        print("Monthly Event Aggregates:")
        print(results.head(10))

        # Cleanup sample file
        if Path(sample_path).exists():
            Path(sample_path).unlink()

        engine.close()
    except Exception as e:
        logger.error(f"Main execution failed: {str(e)}")
        sys.exit(1)

import duckdb
import snowflake.connector
import os
import time
import pandas as pd
from typing import Dict, Any
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class WarehouseBenchmark:
    """Benchmark analytical queries across DuckDB 1.0 and Snowflake"""

    def __init__(self, snowflake_creds: Optional[Dict[str, str]] = None):
        """
        Initialize benchmark tool.
        Args:
            snowflake_creds: Dict with Snowflake credentials (user, password, account, warehouse, database, schema)
        """
        self.snowflake_creds = snowflake_creds
        self.duckdb_conn = None
        self.snowflake_conn = None
        self.results = []

    def setup_duckdb(self, parquet_path: str) -> None:
        """Set up in-memory DuckDB with Parquet data"""
        try:
            self.duckdb_conn = duckdb.connect(":memory:")
            # Load TPC-H lineitem table (1GB Parquet sample)
            self.duckdb_conn.execute(f"""
                CREATE TABLE lineitem AS 
                SELECT * FROM read_parquet('{parquet_path}')
            """)
            row_count = self.duckdb_conn.execute("SELECT COUNT(*) FROM lineitem").fetchone()[0]
            logger.info(f"DuckDB loaded {row_count} rows from {parquet_path}")
        except Exception as e:
            logger.error(f"DuckDB setup failed: {str(e)}")
            raise

    def setup_snowflake(self) -> None:
        """Set up Snowflake connection with credential validation"""
        if not self.snowflake_creds:
            logger.warning("No Snowflake credentials provided, skipping Snowflake benchmarks")
            return

        required_keys = ["user", "password", "account", "warehouse", "database", "schema"]
        missing_keys = [k for k in required_keys if k not in self.snowflake_creds]
        if missing_keys:
            raise ValueError(f"Missing Snowflake credentials: {missing_keys}")

        try:
            self.snowflake_conn = snowflake.connector.connect(
                user=self.snowflake_creds["user"],
                password=self.snowflake_creds["password"],
                account=self.snowflake_creds["account"],
                warehouse=self.snowflake_creds["warehouse"],
                database=self.snowflake_creds["database"],
                schema=self.snowflake_creds["schema"]
            )
            logger.info(f"Connected to Snowflake warehouse {self.snowflake_creds['warehouse']}")

            # Load same TPC-H lineitem data into Snowflake for fair comparison
            self.snowflake_conn.execute("""
                CREATE OR REPLACE TABLE lineitem AS 
                SELECT * FROM @%lineitem (FILE_FORMAT => 'parquet_format')
            """)
            # Assume stage is set up with Parquet file
            logger.info("Snowflake lineitem table loaded")
        except snowflake.connector.Error as e:
            logger.error(f"Snowflake setup failed: {str(e)}")
            self.snowflake_conn = None

    def run_benchmark(self, query: str, query_name: str) -> None:
        """
        Run query on both engines and record results.
        Args:
            query: ANSI SQL query to run
            query_name: Human-readable name for the query
        """
        # Run on DuckDB
        if self.duckdb_conn:
            try:
                start = time.perf_counter()
                duckdb_result = self.duckdb_conn.execute(query).fetchdf()
                duckdb_time = time.perf_counter() - start
                duckdb_rows = len(duckdb_result)
                logger.info(f"DuckDB {query_name} executed in {duckdb_time:.2f}s, {duckdb_rows} rows")
            except Exception as e:
                logger.error(f"DuckDB {query_name} failed: {str(e)}")
                duckdb_time = None
                duckdb_rows = None
        else:
            duckdb_time = None
            duckdb_rows = None

        # Run on Snowflake
        if self.snowflake_conn:
            try:
                start = time.perf_counter()
                snowflake_cursor = self.snowflake_conn.cursor()
                snowflake_cursor.execute(query)
                snowflake_result = snowflake_cursor.fetch_pandas_all()
                snowflake_time = time.perf_counter() - start
                snowflake_rows = len(snowflake_result)
                logger.info(f"Snowflake {query_name} executed in {snowflake_time:.2f}s, {snowflake_rows} rows")
                snowflake_cursor.close()
            except Exception as e:
                logger.error(f"Snowflake {query_name} failed: {str(e)}")
                snowflake_time = None
                snowflake_rows = None
        else:
            snowflake_time = None
            snowflake_rows = None

        self.results.append({
            "query_name": query_name,
            "duckdb_time_s": duckdb_time,
            "duckdb_rows": duckdb_rows,
            "snowflake_time_s": snowflake_time,
            "snowflake_rows": snowflake_rows,
            "speedup": (snowflake_time / duckdb_time) if (duckdb_time and snowflake_time) else None
        })

    def print_results(self) -> None:
        """Print benchmark results as formatted table"""
        print("\n=== Benchmark Results ===")
        print(f"{'Query':<20} {'DuckDB Time (s)':<15} {'Snowflake Time (s)':<18} {'Speedup (x)':<10}")
        for res in self.results:
            speedup = f"{res['speedup']:.1f}" if res['speedup'] else "N/A"
            duckdb_t = f"{res['duckdb_time_s']:.2f}" if res['duckdb_time_s'] else "N/A"
            snowflake_t = f"{res['snowflake_time_s']:.2f}" if res['snowflake_time_s'] else "N/A"
            print(f"{res['query_name']:<20} {duckdb_t:<15} {snowflake_t:<18} {speedup:<10}")

    def cleanup(self) -> None:
        """Close all connections"""
        if self.duckdb_conn:
            self.duckdb_conn.close()
        if self.snowflake_conn:
            self.snowflake_conn.close()
        logger.info("All connections closed")

if __name__ == "__main__":
    # TPC-H benchmark queries (standard analytical workloads)
    tpch_queries = [
        {
            "name": "Q1: Pricing Summary",
            "query": """
                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 lineitem
                WHERE l_shipdate <= DATE '1998-12-01' - INTERVAL '90' DAY
                GROUP BY l_returnflag, l_linestatus
                ORDER BY l_returnflag, l_linestatus
            """
        },
        {
            "name": "Q3: Shipping Priority",
            "query": """
                SELECT 
                    l_orderkey,
                    SUM(l_extendedprice * (1 - l_discount)) AS revenue,
                    o_orderdate,
                    o_shippriority
                FROM customer, orders, lineitem
                WHERE c_mktsegment = 'BUILDING'
                    AND c_custkey = o_custkey
                    AND l_orderkey = o_orderkey
                    AND o_orderdate < DATE '1995-03-15'
                    AND l_shipdate > DATE '1995-03-15'
                GROUP BY l_orderkey, o_orderdate, o_shippriority
                ORDER BY revenue DESC, o_orderdate
                LIMIT 10
            """
        }
    ]

    # Initialize benchmark (Snowflake creds are optional, will skip if missing)
    snowflake_creds = None
    if os.getenv("SNOWFLAKE_USER"):
        snowflake_creds = {
            "user": os.getenv("SNOWFLAKE_USER"),
            "password": os.getenv("SNOWFLAKE_PASSWORD"),
            "account": os.getenv("SNOWFLAKE_ACCOUNT"),
            "warehouse": os.getenv("SNOWFLAKE_WAREHOUSE", "XS_WAREHOUSE"),
            "database": os.getenv("SNOWFLAKE_DATABASE", "TPC_H"),
            "schema": os.getenv("SNOWFLAKE_SCHEMA", "PUBLIC")
        }

    benchmark = WarehouseBenchmark(snowflake_creds)

    try:
        # Set up engines (replace with real Parquet path for TPC-H lineitem)
        benchmark.setup_duckdb("tpch_lineitem_1gb.parquet")
        benchmark.setup_snowflake()

        # Run all TPC-H queries
        for q in tpch_queries:
            benchmark.run_benchmark(q["query"], q["name"])

        # Print results
        benchmark.print_results()
    except Exception as e:
        logger.error(f"Benchmark failed: {str(e)}")
    finally:
        benchmark.cleanup()

import duckdb
import pyarrow as pa
import pyarrow.parquet as pq
from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, StringType, DoubleType, TimestampType
import logging
from typing import List, Dict
import os

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class IcebergToLocalPipeline:
    """Pipeline to read Iceberg tables, transform with DuckDB 1.0, write to local Parquet"""

    def __init__(self, iceberg_catalog_config: Dict[str, str], duckdb_path: str = ":memory:"):
        """
        Initialize pipeline.
        Args:
            iceberg_catalog_config: Config for PyIceberg catalog (e.g., type, uri, warehouse)
            duckdb_path: Path to DuckDB database (default in-memory)
        """
        self.iceberg_catalog = None
        self.duckdb_conn = None
        self.iceberg_catalog_config = iceberg_catalog_config
        self.duckdb_path = duckdb_path
        self._init_iceberg_catalog()
        self._init_duckdb()

    def _init_iceberg_catalog(self) -> None:
        """Initialize PyIceberg catalog with error handling"""
        try:
            self.iceberg_catalog = load_catalog("local_iceberg", **self.iceberg_catalog_config)
            logger.info(f"Connected to Iceberg catalog of type {self.iceberg_catalog_config.get('type')}")
        except Exception as e:
            logger.error(f"Failed to initialize Iceberg catalog: {str(e)}")
            raise

    def _init_duckdb(self) -> None:
        """Initialize DuckDB 1.0 connection with Iceberg extension"""
        try:
            self.duckdb_conn = duckdb.connect(self.duckdb_path)
            # Install and load Iceberg extension (DuckDB 1.0 native support)
            self.duckdb_conn.execute("INSTALL iceberg; LOAD iceberg;")
            logger.info("DuckDB 1.0 initialized with Iceberg extension")
        except duckdb.Error as e:
            logger.error(f"Failed to initialize DuckDB: {str(e)}")
            raise

    def register_iceberg_table(self, iceberg_table_name: str, duckdb_table_name: str) -> None:
        """
        Register an Iceberg table in DuckDB for querying.
        Args:
            iceberg_table_name: Fully qualified Iceberg table name (e.g., db.table)
            duckdb_table_name: Name to use for the table in DuckDB
        """
        try:
            # Get Iceberg table metadata
            iceberg_table = self.iceberg_catalog.load_table(iceberg_table_name)
            metadata_location = iceberg_table.metadata_location

            # Register in DuckDB using Iceberg extension
            self.duckdb_conn.execute(f"""
                CREATE OR REPLACE VIEW {duckdb_table_name} AS 
                SELECT * FROM iceberg_scan('{metadata_location}')
            """)
            row_count = self.duckdb_conn.execute(f"SELECT COUNT(*) FROM {duckdb_table_name}").fetchone()[0]
            logger.info(f"Registered Iceberg table {iceberg_table_name} as DuckDB view {duckdb_table_name} with {row_count} rows")
        except Exception as e:
            logger.error(f"Failed to register Iceberg table {iceberg_table_name}: {str(e)}")
            raise

    def transform_and_write(self, transform_query: str, output_parquet_path: str) -> None:
        """
        Run transformation query on DuckDB and write results to Parquet.
        Args:
            transform_query: SQL query to transform data
            output_parquet_path: Path to write output Parquet file
        """
        try:
            start_time = pa.timestamp("ms").now()
            # Execute transformation
            result_df = self.duckdb_conn.execute(transform_query).fetchdf()
            # Write to Parquet with Snappy compression
            result_df.to_parquet(output_parquet_path, compression="snappy")
            end_time = pa.timestamp("ms").now()
            duration = (end_time - start_time).total_seconds()
            logger.info(f"Transformed {len(result_df)} rows, wrote to {output_parquet_path} in {duration:.2f}s")
        except Exception as e:
            logger.error(f"Transformation failed: {str(e)}")
            raise

    def cleanup(self) -> None:
        """Close connections and clean up resources"""
        if self.duckdb_conn:
            self.duckdb_conn.close()
        if self.iceberg_catalog:
            # PyIceberg doesn't require explicit close, but we log for completeness
            logger.info("Iceberg catalog resources released")

    def validate_output(self, output_parquet_path: str, expected_row_count: int) -> bool:
        """
        Validate output Parquet file.
        Args:
            output_parquet_path: Path to output Parquet file
            expected_row_count: Expected number of rows
        Returns:
            True if validation passes, False otherwise
        """
        try:
            if not os.path.exists(output_parquet_path):
                logger.error(f"Output file {output_parquet_path} does not exist")
                return False
            df = pq.read_table(output_parquet_path).to_pandas()
            if len(df) != expected_row_count:
                logger.error(f"Row count mismatch: expected {expected_row_count}, got {len(df)}")
                return False
            logger.info(f"Output validation passed: {len(df)} rows in {output_parquet_path}")
            return True
        except Exception as e:
            logger.error(f"Validation failed: {str(e)}")
            return False

if __name__ == "__main__":
    # Example config for local Iceberg catalog (using SQLite for metadata)
    iceberg_config = {
        "type": "sqlite",
        "uri": "sqlite:///iceberg_catalog.db",
        "warehouse": "s3://local-iceberg-warehouse/"
    }

    # Initialize pipeline
    try:
        pipeline = IcebergToLocalPipeline(iceberg_config, duckdb_path=":memory:")

        # Register Iceberg table (assumes table exists in catalog)
        pipeline.register_iceberg_table("ecommerce.orders", "orders")

        # Transformation query: Calculate monthly order revenue and discount rate
        transform_query = """
        SELECT 
            DATE_TRUNC('month', order_timestamp) AS order_month,
            COUNT(DISTINCT order_id) AS total_orders,
            SUM(order_total) AS total_revenue,
            AVG(discount_rate) AS avg_discount_rate,
            SUM(CASE WHEN status = 'shipped' THEN 1 ELSE 0 END) AS shipped_orders
        FROM orders
        WHERE order_timestamp >= '2026-01-01'
        GROUP BY order_month
        ORDER BY order_month
        """

        # Run transformation and write to Parquet
        output_path = "monthly_order_metrics.parquet"
        pipeline.transform_and_write(transform_query, output_path)

        # Validate output (assumes 1200 rows for demo)
        pipeline.validate_output(output_path, expected_row_count=1200)

        # Print sample results
        sample_df = pd.read_parquet(output_path)
        print("Sample Monthly Order Metrics:")
        print(sample_df.head())

    except Exception as e:
        logger.error(f"Pipeline failed: {str(e)}")
    finally:
        pipeline.cleanup()

# Bad: Load entire Parquet into memory
conn.execute("CREATE TABLE sales AS SELECT * FROM read_parquet('sales_2026.parquet')")
conn.execute("SELECT month, SUM(revenue) FROM sales GROUP BY month")

# Good: Zero-copy scan directly in query
conn.execute("""
    SELECT 
        DATE_TRUNC('month', order_date) AS month,
        SUM(order_total) AS total_revenue
    FROM read_parquet('sales_2026.parquet')
    GROUP BY month
""")

import duckdb

conn = duckdb.connect(":memory:")
# Enable telemetry
conn.execute("SET enable_telemetry=true;")

# Run query
conn.execute("""
    SELECT region, SUM(sales) 
    FROM read_parquet('regional_sales.parquet') 
    GROUP BY region
""")

# Get telemetry for last query
telemetry = conn.execute("SELECT * FROM duckdb_telemetry()").fetchdf()
print(f"Query used {telemetry['memory_bytes'].iloc[0]/1e9:.2f}GB RAM")
print(f"Query CPU time: {telemetry['cpu_time_ms'].iloc[0]}ms")

import duckdb

conn = duckdb.connect(":memory:")
# Query Snowflake table directly from DuckDB (requires snowflake connector installed)
conn.execute("INSTALL snowflake; LOAD snowflake;")
conn.execute("""
    CREATE VIEW cold_sales AS 
    SELECT * FROM snowflake_scan(
        'snowflake_creds.json',
        'SALES_DB.PUBLIC.ORDER_HISTORY'
    )
""")

# Join hot local data with cold Snowflake data
conn.execute("""
    SELECT 
        h.month,
        h.monthly_revenue,
        c.historical_revenue
    FROM local_hot_sales h
    LEFT JOIN cold_sales c ON h.month = c.month
""")