Overview

What You'll Learn

What You'll Build

Prerequisites

What You'll Need

Understand Interactive Warehouses

Interactive Warehouses

Zero-copy interactive analytics

-- 1. Create an interactive warehouse
CREATE OR REPLACE INTERACTIVE WAREHOUSE analytics_iwh
  WAREHOUSE_SIZE = 'XSMALL';

-- 2. (Optional) Attach high-priority tables for proactive caching
ALTER WAREHOUSE analytics_iwh
  ADD TABLES (your_db.your_schema.critical_table_1, your_db.your_schema.critical_table_2);

-- 3. Query any standard table, no conversion needed
USE WAREHOUSE analytics_iwh;
SELECT * FROM your_db.your_schema.any_standard_table WHERE ...;

Note: Before zero-copy interactive analytics, the only way to query data at interactive latency was to convert it into an interactive table. Interactive tables still exist and remain supported, mainly for compatibility with earlier interactive analytics setups. For new work, Snowflake recommends querying your standard tables directly through zero-copy interactive analytics instead.

Use cases

Limitations

Setup

Data operations

Note: The companion notebook creates all of these objects automatically using the {{DB_NAME}} and {{STANDARD_WH_NAME}} variables defined in the "Set common variables" cell. The steps below show the equivalent manual SQL. If running outside the notebook, replace {{DB_NAME}} and {{STANDARD_WH_NAME}} with your own names (e.g. JSMITH_MY_DEMO_DB and JSMITH_STD_WH).

Optional: Create warehouse

CREATE OR REPLACE WAREHOUSE {{STANDARD_WH_NAME}} WITH WAREHOUSE_SIZE='X-SMALL';

Step 1: Create a Database and Schema

CREATE DATABASE IF NOT EXISTS {{DB_NAME}};
CREATE SCHEMA IF NOT EXISTS {{DB_NAME}}.BENCHMARK_FDN;

Step 2: Create a new stage

-- Define database and schema to use
USE SCHEMA {{DB_NAME}}.BENCHMARK_FDN;

-- Create a stage that includes the definition for the CSV file format
CREATE OR REPLACE STAGE my_csv_stage
  FILE_FORMAT = (
    TYPE = 'CSV'
    SKIP_HEADER = 1
    FIELD_OPTIONALLY_ENCLOSED_BY = '"'
  );

Step 3: Upload CSV to a stage

Step 4: Create the Table and Load Data

-- Use your database and schema
USE SCHEMA {{DB_NAME}}.BENCHMARK_FDN;

-- Create the table with the correct data types
CREATE OR REPLACE TABLE HITS2_CSV (
    EventDate DATE,
    CounterID INT,
    ClientIP STRING,
    SearchEngineID INT,
    SearchPhrase STRING,
    ResolutionWidth INT,
    Title STRING,
    IsRefresh INT,
    DontCountHits INT
);

-- Copy the data from your stage into the table
-- Make sure to replace 'my_csv_stage' with your stage name
COPY INTO HITS2_CSV FROM @my_csv_stage/synthetic_hits_data.csv
  FILE_FORMAT = (TYPE = 'CSV' SKIP_HEADER = 1);

Step 5: Query the data

USE WAREHOUSE {{STANDARD_WH_NAME}};
SELECT * FROM {{DB_NAME}}.BENCHMARK_FDN.HITS2_CSV LIMIT 100;

Performance demo

Set common variables

from snowflake.snowpark.context import get_active_session

session = get_active_session()
USER = session.sql("SELECT CURRENT_USER()").collect()[0][0]
DB_NAME = f'{USER}_MY_DEMO_DB'
INTERACTIVE_WH_NAME = f'{USER}_INT_WH'
STANDARD_WH_NAME = f'{USER}_STD_WH'

print(f"User: {USER}\nDatabase: {DB_NAME}\nInteractive WH: {INTERACTIVE_WH_NAME}\nStandard WH: {STANDARD_WH_NAME}")

Set up role, warehouse, and database

USE ROLE SYSADMIN;

-- Create the compute and database objects used throughout this notebook (idempotent)
CREATE WAREHOUSE IF NOT EXISTS {{STANDARD_WH_NAME}} WITH WAREHOUSE_SIZE = 'X-SMALL';
CREATE DATABASE IF NOT EXISTS {{DB_NAME}};

CREATE SCHEMA IF NOT EXISTS {{DB_NAME}}.BENCHMARK_FDN;

USE WAREHOUSE {{STANDARD_WH_NAME}};
USE DATABASE {{DB_NAME}};

Note: In a Snowflake Notebook, SQL and Python cells share the same session. Any USE ROLE, USE DATABASE, or USE WAREHOUSE statement you run in a SQL cell also applies to subsequent Python cells (and vice versa).

Create an interactive warehouse

CREATE OR REPLACE INTERACTIVE WAREHOUSE {{INTERACTIVE_WH_NAME}}
    WAREHOUSE_SIZE = 'XSMALL'
    MIN_CLUSTER_COUNT = 1
    MAX_CLUSTER_COUNT = 1
    COMMENT = 'Interactive warehouse demo';

Data setup and loading

USE WAREHOUSE {{STANDARD_WH_NAME}};

Note: The data is loaded from the bundled CSV using pandas and write_pandas with no external network access required. Make sure synthetic_hits_data.csv is added to the notebook's files.

import pandas as pd

DB, SCHEMA, TABLE = DB_NAME, "BENCHMARK_FDN", "HITS2_CSV"
FQ = f"{DB}.{SCHEMA}.{TABLE}"
CSV_FILE = "synthetic_hits_data.csv"  # bundled with this notebook

# Create the source table if it doesn't already exist
session.sql(f"""
CREATE TABLE IF NOT EXISTS {FQ} (
    EventDate DATE,
    CounterID INT,
    ClientIP STRING,
    SearchEngineID INT,
    SearchPhrase STRING,
    ResolutionWidth INT,
    Title STRING,
    IsRefresh INT,
    DontCountHits INT
)
""").collect()

# Idempotent load: only load when the table is empty
row_count = session.sql(f"SELECT COUNT(*) FROM {FQ}").collect()[0][0]
if row_count > 0:
    print(f"{FQ} already has {row_count:,} rows. Skipping data load.")
else:
    print(f"Loading data into {FQ} ...")
    pdf = pd.read_csv(CSV_FILE)
    pdf["EventDate"] = pd.to_datetime(pdf["EventDate"]).dt.date    
    session.write_pandas(pdf, TABLE, database=DB, schema=SCHEMA, quote_identifiers=False)
    row_count = session.sql(f"SELECT COUNT(*) FROM {FQ}").collect()[0][0]
    print(f"Loaded {row_count:,} rows into {FQ}.")

TARGET_MULTIPLIER = 20

row_count = session.sql(f"SELECT COUNT(*) FROM {FQ}").collect()[0][0]
if row_count >= TARGET_MULTIPLIER * 100_000 * 0.9:
    print(f"{FQ} already has {row_count:,} rows. Skipping data expansion.")
else:
    print(f"Expanding {FQ} to roughly {TARGET_MULTIPLIER * 100_000:,} rows ...")
    session.sql(f"""
        INSERT INTO {FQ}
        SELECT
            DATEADD(day, UNIFORM(-3, 3, RANDOM()), t.EventDate),
            t.CounterID,
            CONCAT(TO_VARCHAR(UNIFORM(1, 255, RANDOM())), '.', TO_VARCHAR(UNIFORM(1, 255, RANDOM())), '.',
                   TO_VARCHAR(UNIFORM(1, 255, RANDOM())), '.', TO_VARCHAR(UNIFORM(1, 255, RANDOM())))),
            t.SearchEngineID,
            t.SearchPhrase,
            GREATEST(1, t.ResolutionWidth + UNIFORM(-100, 100, RANDOM())),
            t.Title,
            t.IsRefresh,
            t.DontCountHits
        FROM {FQ} AS t, TABLE(GENERATOR(ROWCOUNT => {TARGET_MULTIPLIER - 1})) AS g
    """).collect()
    row_count = session.sql(f"SELECT COUNT(*) FROM {FQ}").collect()[0][0]
    print(f"Expanded {FQ} to {row_count:,} rows.")

USE WAREHOUSE {{STANDARD_WH_NAME}};
SELECT * FROM {{DB_NAME}}.BENCHMARK_FDN.HITS2_CSV LIMIT 100;

Attach a table to the interactive warehouse

USE DATABASE {{DB_NAME}};
ALTER WAREHOUSE {{INTERACTIVE_WH_NAME}} ADD TABLES(BENCHMARK_FDN.HITS2_CSV);

Note: ADD TABLES is a performance optimization, not a requirement. It proactively warms the warehouse's data cache so queries avoid a cold start. Any table you don't attach is still queryable and gets cached on demand the first time it's accessed. Proactive warming is currently limited to 10 tables.

Configure a fallback warehouse

ALTER WAREHOUSE {{INTERACTIVE_WH_NAME}} SET FALLBACK_WAREHOUSE = {{STANDARD_WH_NAME}};

SHOW WAREHOUSES LIKE '{{INTERACTIVE_WH_NAME}}';

Sequential Query Benchmark

import time
import numpy as np

cursor = session.connection.cursor()

runs = 50

def run_and_measure(count, mode):
    wh = INTERACTIVE_WH_NAME if mode == "iw" else STANDARD_WH_NAME
    table = "BENCHMARK_FDN.HITS2_CSV"
    query = f"""
        SELECT SearchEngineID, ClientIP, COUNT(*) AS c, SUM(IsRefresh), AVG(ResolutionWidth)
        FROM {table}
        WHERE SearchPhrase <> ''
        GROUP BY SearchEngineID, ClientIP
        ORDER BY c DESC LIMIT 10
    """
    cursor.execute(f"USE WAREHOUSE {wh}")
    cursor.execute('ALTER SESSION SET USE_CACHED_RESULT = FALSE;')

    timings = []
    for _ in range(count + 1):
        t0 = time.time()
        cursor.execute(query).fetchall()
        timings.append(time.time() - t0)
    return timings[1:] # skip warm-up run

counts_iw = run_and_measure(runs, "iw")
print(counts_iw)

counts_std = run_and_measure(runs, "std")
print(counts_std)

import matplotlib.pyplot as plt

titles = [(i+1) for i in range(0, len(counts_iw))]

x = np.arange(len(titles))  # the label locations
width = 0.35  # bar width

fig, ax = plt.subplots(figsize=(15, 5))
ax.bar(x - width/2, counts_std, width, label="Standard", color="#5B5B5B")
ax.bar(x + width/2, counts_iw, width, label="Interactive", color="#29B5E8")

ax.set_ylabel("Latency")
ax.set_xlabel("Query run")
ax.set_title("Standard vs Interactive warehouse")
ax.set_xticks(x)
ax.set_xticklabels(titles)
ax.legend(
    loc='upper center',
    bbox_to_anchor=(0.5, -0.15),
    ncol=2
)
plt.show()

# Calculate means and standard deviations for error bars
mean_std = np.mean(counts_std)
mean_iw = np.mean(counts_iw)
std_std = np.std(counts_std)
std_iw = np.std(counts_iw)

fig, ax = plt.subplots(figsize=(6, 5))
bars = ax.bar(["Standard", "Interactive"], [mean_std, mean_iw],
              yerr=[std_std, std_iw], capsize=8,
              color=["#5B5B5B", "#29B5E8"], width=0.5)

ax.set_ylabel("Latency (seconds)")
ax.set_title("Standard vs Interactive warehouse\n(mean over {} runs with std dev)".format(len(counts_std)))
plt.tight_layout()
plt.show()

Concurrent Query Benchmark

import random, time, numpy as np
from concurrent.futures import ThreadPoolExecutor, as_completed

QUERY_TEMPLATES = {
    "light": "SELECT * FROM {table} WHERE CounterID = 62 LIMIT 1",
    "medium": """SELECT SearchEngineID, ClientIP, COUNT(*) AS c, SUM(IsRefresh), AVG(ResolutionWidth)
        FROM {table} WHERE SearchPhrase <> ''
        GROUP BY SearchEngineID, ClientIP ORDER BY c DESC LIMIT 10""",
    "heavy": """SELECT EventDate, COUNT(*) AS hits, COUNT(DISTINCT ClientIP) AS unique_ips,
        AVG(ResolutionWidth), SUM(CASE WHEN SearchPhrase <> '' THEN 1 ELSE 0 END)
        FROM {table} GROUP BY EventDate ORDER BY EventDate""",
}

def build_query_pool(table):
    return [(k, v.format(table=table)) for k, v in QUERY_TEMPLATES.items()]

def worker(conn, wh_name, query_pool, n_queries=6, arrival_rate=5):
    cur = conn.cursor()
    cur.execute(f"USE WAREHOUSE {wh_name}")
    cur.execute("ALTER SESSION SET USE_CACHED_RESULT = FALSE")
    latencies = []
    for _ in range(n_queries):
        time.sleep(random.expovariate(arrival_rate))
        _, query = random.choice(query_pool)
        cur.execute(query).fetchall()
        qid = cur.sfqid
        ms = cur.execute(f"SELECT TOTAL_ELAPSED_TIME FROM TABLE(INFORMATION_SCHEMA.QUERY_HISTORY_BY_SESSION()) WHERE QUERY_ID = '{qid}'").fetchone()[0]
        latencies.append(ms / 1000.0)
    cur.close()
    return latencies

def run_concurrent_benchmark(conn, wh_name, table, concurrency_levels, rounds=10):
    query_pool = build_query_pool(table)
    all_results = []
    for n in concurrency_levels:
        print(f"  concurrency={n} ({rounds} rounds) ...", end=" ", flush=True)
        round_stats = []
        for _ in range(rounds):
            t0 = time.time()
            with ThreadPoolExecutor(max_workers=n) as pool:
                futures = [pool.submit(worker, conn, wh_name, query_pool) for _ in range(n)]
                lats = [l for f in as_completed(futures) for l in f.result()]
            wall = time.time() - t0
            round_stats.append({"p50": np.percentile(lats, 50), "p90": np.percentile(lats, 90),
                                "p99": np.percentile(lats, 99), "throughput_qps": len(lats) / wall})
        result = {"concurrency": n}
        for m in ["p50", "p90", "p99", "throughput_qps"]:
            vals = [r[m] for r in round_stats]
            result[f"{m}_mean"], result[f"{m}_std"] = np.mean(vals), np.std(vals)
        all_results.append(result)
        print(f"p50={result['p50_mean']:.3f}s(±{result['p50_std']:.3f})  p90={result['p90_mean']:.3f}s(±{result['p90_std']:.3f})  qps={result['throughput_qps_mean']:.1f}(±{result['throughput_qps_std']:.1f})")
    return all_results

concurrency_levels = [1, 2, 4, 8]
conn = session.connection

print("Interactive warehouse:")
results_iw = run_concurrent_benchmark(conn, INTERACTIVE_WH_NAME, "BENCHMARK_FDN.HITS2_CSV", concurrency_levels)

print("\nStandard warehouse:")
results_std = run_concurrent_benchmark(conn, STANDARD_WH_NAME, "BENCHMARK_FDN.HITS2_CSV", concurrency_levels)

print(f"\nBenchmark complete ({run_concurrent_benchmark.__defaults__[0]} rounds per level).")

import matplotlib.pyplot as plt

benchmark_rounds = run_concurrent_benchmark.__defaults__[0]
levels = [r["concurrency"] for r in results_iw]

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 6))

for results, label, color in [(results_std, "Standard", "#5B5B5B"), (results_iw, "Interactive", "#29B5E8")]:
    for metric, marker, ls, alpha in [("p50", "o", "-", 1.0), ("p90", "s", "--", 0.6), ("p99", "^", ":", 0.4)]:
        ax1.errorbar(levels, [r[f"{metric}_mean"] for r in results],
                     yerr=[r[f"{metric}_std"] for r in results],
                     fmt=f"{marker}{ls}", color=color, alpha=alpha, capsize=4, label=f"{label} {metric}")

ax1.set(xlabel="Concurrent Workers", ylabel="Latency (seconds)", xticks=levels)
ax1.set_ylim(bottom=0)
ax1.set_title(f"Concurrency vs Latency (lower is better)\nmean ± std over {benchmark_rounds} rounds")
ax1.legend(fontsize=7, ncol=2)
ax1.grid(True, alpha=0.3)

x = np.arange(len(levels))
w = 0.35
for results, label, color, offset in [(results_std, "Standard", "#5B5B5B", -w/2), (results_iw, "Interactive", "#29B5E8", w/2)]:
    ax2.bar(x + offset, [r["throughput_qps_mean"] for r in results], w,
            yerr=[r["throughput_qps_std"] for r in results], capsize=4, label=label, color=color)

ax2.set(xlabel="Concurrent Workers", ylabel="Queries / Second", xticks=x)
ax2.set_xticklabels(levels)
ax2.set_title(f"Concurrency vs Throughput (higher is better)\nmean ± std over {benchmark_rounds} rounds")
ax2.legend()
ax2.grid(True, alpha=0.3, axis="y")

plt.tight_layout()
plt.show()

Cleanup

DROP WAREHOUSE IF EXISTS {{INTERACTIVE_WH_NAME}};
DROP WAREHOUSE IF EXISTS {{STANDARD_WH_NAME}};
DROP DATABASE IF EXISTS {{DB_NAME}};

Conclusion and Resources

What You Learned

Related Resources