Queries in Snowflake try to use the entire standard warehouse. So what happens when we’re running dozens of queries on the same warehouse at the same time, and each of those queries is attempting to use all of the resources of the warehouse? All of the queries often run slower as they compete for resources. Often queries can intersperse and share resources reasonably, but it depends on what the workload looks like.
Concurrency as a factor comes as a surprise to some customers. Queries that run concurrently often run slower than when run in isolation. This is one reason that is important to both test query performance in isolation and also run load tests that come as close to simulating a real-world workload as possible. Test and do the work to understand or even reduce the impact of concurrency for your workloads
Multi-Cluster Warehouses
Many customers expect multi-cluster warehouses to solve this issue. Multi-cluster warehouses are available on Enterprise and higher. They involve spinning up multiple warehouse clusters of the same size to handle more queries at a time. This can be more expensive because you are charged for the additional compute. Multi-cluster warehouses allow the specification of the minimum and maximum number of clusters. The maximum allows you to control cost through limiting how wide a multi-cluster warehouse can get. The minimum allows you to start with a higher amount of compute from the beginning. I’ve worked with customers who set their max above 100, though more commonly I work with customers using single digits.
The scaling policy determines how Snowflake scales up and spins down clusters in the warehouse. The standard policy triggers the start of a new cluster on queuing due to load. There are mechanisms to prevent mercurial changes and starting too many clusters off of one queuing event. As you’ll see in the example later in this article, queuing may not be happening as much as you expect it to.
MAX_CONCURRENCY_LEVEL
By default, the setting of MAX_CONCURRENCY_LEVEL is 8. This is a parameter you can adjust for each warehouse. Note the naming of this parameter – it is the maximum number of full-slot queries Snowflake will run on a warehouse before queuing. There are other determinants based on how busy the warehouse is. Snowflake can choose to queue when there are only 6 or 4 or fewer queries, if it determines that the warehouse clusters are already too busy.
To completely eliminate the possibility of any concurrency impact, it is possible set MAX_CONCURRENCY_LEVEL to 1. This is NOT something that I recommend nor something that Snowflake recommends. It tends to be excessively expensive and can lead to either a lot of queuing or a lot of MCW scaling, or both. I cannot remember ever recommending that a customer consider increasing MAX_CONCURENCY_LEVEL, I have recommended that they lower it after detailed conversations about their workload and their priorities and goals.
Performance Impacts of Concurrency
As mentioned at the start, Snowflake tries to make use of all of the compute available when running a query. This means that even two queries running at the same time can impact the performance of one or both queries. This is one of the reasons that realistic load testing is critical. Balancing the cost of more compute against your own Service Level Objectives (SLOs) is one of the things that load testing can really help with.
Example Concurrency Impact
Dearest gentle reader, if you’ve read this blog for more than a few articles, you probably know that I love to put data behind my statements, and this is no exception. I used Cortex Code to help me come up with a meaningful example here. I’ve found that with some back-and-forth, Cortex Code is really excellent at demonstrating concepts or problems.
This example uses the database/schema SNOWFLAKE_SAMPLE_DATA.TPCH_SF10, which should be available in all Snowflake accounts. The python script used to generate this test is included in the code block at the bottom of this article. I do not warranty this code or promise it will work for anything – use it at your own risk. I provide it so you can see the methodology used to get the numbers I share here. I ran it on a free trial account. The general process was that each of 8 light, medium, and heavy queries was tested alone to establish a performance baseline. Then 8 threads ran three queries each – one light, one medium, and one heavy. The order in which each thread ran it’s light medium, and heavy query was rotated, so that we weren’t running all the light queries at once. There was no ramp-up period, which I normally consider a cardinal sin.
Overall, some queuing was seen, but only 5.8% of the concurrent slowdown was from queuing (QUEUED_OVERLOAD_TIME). The other 94.2% was resource contention or execution itself. Here is what the timing differences looked like per query:
For this chart, each query was given a number with a starting letter indicating whether it was a light(L), Medium(M), or Heavy(H) query. The chart shows the solo timing for the query first in a lighter shade, followed by the concurrent timing in a darker shade.
Most queries (21 of 24) had zero queue time — they were admitted immediately but ran slower because queries were fighting over one extra-small warehouse’s resources.
Known Issues
I do not represent this as a perfect representation of experimentation. The baseline includes only one run of each query rather than multiple runs at multiple times that are then aggregated. The goal was to introduce concurrency, so all threads started at the same time. There was no ramp-up time or think time between queries. These are all things I would include in a real world test that are not needed for this simple demonstration.
Conclusions
The goal of this test was not to establish a benchmark or to expose a weakness – it is to demonstrate the impact of concurrency on performance and to illustrate one of the most powerful ways of combining isolated query testing with load testing – understanding the difference in performance that concurrency can make.
The way I ran queries here was very naive. It was pretty random, as it didn’t matter a lot for a pure test like this. In the real world, using a tool like JMeter with a reasonable ramp-up period (at least 15 minutes, in steps), and think time between queries makes a lot more sense. Simulate your real workload or your peak workload to the best of your ability to get the best results.
Future Improvements
When adaptive warehouses (announced at Snowflake Summit, last year) become available, they may change how the picture here looks in several different ways. It will be interesting to revisit this kind of experiment once they’re available.
Python Script
#!/usr/bin/env python3
"""
Snowflake Concurrency Performance Test
=======================================
Demonstrates the impact of concurrent query execution on an X-Small warehouse.
Breaks down slowdown into queuing time vs execution time using QUERY_HISTORY.
Runs 24 queries against SNOWFLAKE_SAMPLE_DATA.TPCH_SF10:
- Phase 1: Each query runs solo (sequential baseline)
- Phase 2: 8 threads x 3 queries each run concurrently
Usage:
SNOWFLAKE_CONNECTION_NAME=MY_CONNECTION python3 concurrency_test.py
SNOWFLAKE_CONNECTION_NAME=MY_CONNECTION python3 concurrency_test.py --keep-warehouse
"""
import argparse
import os
import time
import uuid
from concurrent.futures import ThreadPoolExecutor, as_completed
import snowflake.connector
CONNECTION_NAME = os.getenv("SNOWFLAKE_CONNECTION_NAME") or "MY_CONNECTION"
WAREHOUSE = "CONCURRENCY_TEST_WH"
SCHEMA = "SNOWFLAKE_SAMPLE_DATA.TPCH_SF10"
RUN_ID = uuid.uuid4().hex[:8]
LIGHT_QUERIES = [
("L1", f"SELECT COUNT(*) FROM {SCHEMA}.LINEITEM"),
("L2", f"SELECT AVG(L_EXTENDEDPRICE) FROM {SCHEMA}.LINEITEM"),
("L3", f"SELECT COUNT(DISTINCT O_CUSTKEY) FROM {SCHEMA}.ORDERS"),
("L4", f"SELECT SUM(L_QUANTITY) FROM {SCHEMA}.LINEITEM WHERE L_SHIPDATE >= '1995-01-01'"),
("L5", f"SELECT MIN(O_ORDERDATE), MAX(O_ORDERDATE) FROM {SCHEMA}.ORDERS"),
("L6", f"SELECT COUNT(*) FROM {SCHEMA}.CUSTOMER WHERE C_MKTSEGMENT = 'BUILDING'"),
("L7", f"SELECT AVG(PS_SUPPLYCOST) FROM {SCHEMA}.PARTSUPP"),
("L8", f"SELECT SUM(L_DISCOUNT * L_EXTENDEDPRICE) FROM {SCHEMA}.LINEITEM WHERE L_SHIPDATE BETWEEN '1994-01-01' AND '1994-12-31'"),
]
MEDIUM_QUERIES = [
("M1", f"""
SELECT O_ORDERPRIORITY, COUNT(*) AS cnt
FROM {SCHEMA}.ORDERS
WHERE O_ORDERDATE >= '1996-01-01' AND O_ORDERDATE < '1996-04-01'
GROUP BY O_ORDERPRIORITY ORDER BY cnt DESC
"""),
("M2", f"""
SELECT L_RETURNFLAG, L_LINESTATUS, SUM(L_QUANTITY), SUM(L_EXTENDEDPRICE),
SUM(L_EXTENDEDPRICE * (1 - L_DISCOUNT)) AS disc_price
FROM {SCHEMA}.LINEITEM
WHERE L_SHIPDATE <= '1998-09-02'
GROUP BY L_RETURNFLAG, L_LINESTATUS ORDER BY L_RETURNFLAG, L_LINESTATUS
"""),
("M3", f"""
SELECT N_NAME, COUNT(*) AS supplier_count
FROM {SCHEMA}.SUPPLIER s JOIN {SCHEMA}.NATION n ON s.S_NATIONKEY = n.N_NATIONKEY
GROUP BY N_NAME ORDER BY supplier_count DESC
"""),
("M4", f"""
SELECT C_MKTSEGMENT, SUM(L_EXTENDEDPRICE * (1 - L_DISCOUNT)) AS revenue
FROM {SCHEMA}.CUSTOMER c
JOIN {SCHEMA}.ORDERS o ON c.C_CUSTKEY = o.O_CUSTKEY
JOIN {SCHEMA}.LINEITEM l ON o.O_ORDERKEY = l.L_ORDERKEY
WHERE O_ORDERDATE < '1995-03-15' AND L_SHIPDATE > '1995-03-15'
GROUP BY C_MKTSEGMENT ORDER BY revenue DESC
"""),
("M5", f"""
SELECT YEAR(O_ORDERDATE) AS yr, COUNT(*) AS order_count, SUM(O_TOTALPRICE) AS total
FROM {SCHEMA}.ORDERS
GROUP BY yr ORDER BY yr
"""),
("M6", f"""
SELECT L_SHIPMODE, SUM(CASE WHEN O_ORDERPRIORITY IN ('1-URGENT','2-HIGH') THEN 1 ELSE 0 END) AS high_priority,
SUM(CASE WHEN O_ORDERPRIORITY NOT IN ('1-URGENT','2-HIGH') THEN 1 ELSE 0 END) AS low_priority
FROM {SCHEMA}.ORDERS o JOIN {SCHEMA}.LINEITEM l ON o.O_ORDERKEY = l.L_ORDERKEY
WHERE L_SHIPMODE IN ('MAIL','SHIP') AND L_RECEIPTDATE >= '1994-01-01' AND L_RECEIPTDATE < '1995-01-01'
GROUP BY L_SHIPMODE ORDER BY L_SHIPMODE
"""),
("M7", f"""
SELECT P_TYPE, AVG(L_EXTENDEDPRICE) AS avg_price
FROM {SCHEMA}.PART p JOIN {SCHEMA}.LINEITEM l ON p.P_PARTKEY = l.L_PARTKEY
WHERE P_SIZE BETWEEN 1 AND 10
GROUP BY P_TYPE ORDER BY avg_price DESC LIMIT 20
"""),
("M8", f"""
SELECT C_NATIONKEY, COUNT(DISTINCT O_ORDERKEY) AS orders, SUM(O_TOTALPRICE) AS spend
FROM {SCHEMA}.CUSTOMER c JOIN {SCHEMA}.ORDERS o ON c.C_CUSTKEY = o.O_CUSTKEY
WHERE O_ORDERDATE >= '1997-01-01'
GROUP BY C_NATIONKEY ORDER BY spend DESC
"""),
]
HEAVY_QUERIES = [
("H1", f"""
SELECT s.S_NAME, COUNT(*) AS numwait
FROM {SCHEMA}.SUPPLIER s
JOIN {SCHEMA}.LINEITEM l1 ON s.S_SUPPKEY = l1.L_SUPPKEY
JOIN {SCHEMA}.ORDERS o ON l1.L_ORDERKEY = o.O_ORDERKEY
WHERE o.O_ORDERSTATUS = 'F'
AND l1.L_RECEIPTDATE > l1.L_COMMITDATE
AND EXISTS (SELECT 1 FROM {SCHEMA}.LINEITEM l2 WHERE l2.L_ORDERKEY = l1.L_ORDERKEY AND l2.L_SUPPKEY <> l1.L_SUPPKEY)
AND NOT EXISTS (SELECT 1 FROM {SCHEMA}.LINEITEM l3 WHERE l3.L_ORDERKEY = l1.L_ORDERKEY AND l3.L_SUPPKEY <> l1.L_SUPPKEY AND l3.L_RECEIPTDATE > l3.L_COMMITDATE)
GROUP BY s.S_NAME ORDER BY numwait DESC, s.S_NAME LIMIT 100
"""),
("H2", f"""
SELECT C_CUSTKEY, C_NAME, SUM(L_EXTENDEDPRICE * (1 - L_DISCOUNT)) AS revenue,
C_ACCTBAL, N_NAME, C_ADDRESS, C_PHONE, C_COMMENT
FROM {SCHEMA}.CUSTOMER c
JOIN {SCHEMA}.ORDERS o ON c.C_CUSTKEY = o.O_CUSTKEY
JOIN {SCHEMA}.LINEITEM l ON o.O_ORDERKEY = l.L_ORDERKEY
JOIN {SCHEMA}.NATION n ON c.C_NATIONKEY = n.N_NATIONKEY
WHERE O_ORDERDATE >= '1993-10-01' AND O_ORDERDATE < '1994-01-01' AND L_RETURNFLAG = 'R'
GROUP BY C_CUSTKEY, C_NAME, C_ACCTBAL, C_PHONE, N_NAME, C_ADDRESS, C_COMMENT
ORDER BY revenue DESC LIMIT 20
"""),
("H3", f"""
SELECT YEAR(O_ORDERDATE) AS o_year,
SUM(CASE WHEN nation = 'BRAZIL' THEN volume ELSE 0 END) / SUM(volume) AS mkt_share
FROM (
SELECT O_ORDERDATE, L_EXTENDEDPRICE * (1 - L_DISCOUNT) AS volume, n2.N_NAME AS nation
FROM {SCHEMA}.PART p
JOIN {SCHEMA}.LINEITEM l ON p.P_PARTKEY = l.L_PARTKEY
JOIN {SCHEMA}.SUPPLIER s ON l.L_SUPPKEY = s.S_SUPPKEY
JOIN {SCHEMA}.ORDERS o ON l.L_ORDERKEY = o.O_ORDERKEY
JOIN {SCHEMA}.CUSTOMER c ON o.O_CUSTKEY = c.C_CUSTKEY
JOIN {SCHEMA}.NATION n1 ON c.C_NATIONKEY = n1.N_NATIONKEY
JOIN {SCHEMA}.REGION r ON n1.N_REGIONKEY = r.R_REGIONKEY
JOIN {SCHEMA}.NATION n2 ON s.S_NATIONKEY = n2.N_NATIONKEY
WHERE r.R_NAME = 'AMERICA' AND O_ORDERDATE BETWEEN '1995-01-01' AND '1996-12-31' AND P_TYPE = 'ECONOMY ANODIZED STEEL'
) sub
GROUP BY o_year ORDER BY o_year
"""),
("H4", f"""
SELECT nation, o_year, SUM(amount) AS sum_profit
FROM (
SELECT n.N_NAME AS nation, YEAR(O_ORDERDATE) AS o_year,
L_EXTENDEDPRICE * (1 - L_DISCOUNT) - PS_SUPPLYCOST * L_QUANTITY AS amount
FROM {SCHEMA}.PART p
JOIN {SCHEMA}.LINEITEM l ON p.P_PARTKEY = l.L_PARTKEY
JOIN {SCHEMA}.SUPPLIER s ON l.L_SUPPKEY = s.S_SUPPKEY
JOIN {SCHEMA}.PARTSUPP ps ON l.L_SUPPKEY = ps.PS_SUPPKEY AND l.L_PARTKEY = ps.PS_PARTKEY
JOIN {SCHEMA}.ORDERS o ON l.L_ORDERKEY = o.O_ORDERKEY
JOIN {SCHEMA}.NATION n ON s.S_NATIONKEY = n.N_NATIONKEY
WHERE P_NAME LIKE '%green%'
) sub
GROUP BY nation, o_year ORDER BY nation, o_year DESC
"""),
("H5", f"""
SELECT L_ORDERKEY,
SUM(L_EXTENDEDPRICE * (1 - L_DISCOUNT)) AS revenue,
O_ORDERDATE, O_SHIPPRIORITY,
RANK() OVER (ORDER BY SUM(L_EXTENDEDPRICE * (1 - L_DISCOUNT)) DESC) AS rnk
FROM {SCHEMA}.CUSTOMER c
JOIN {SCHEMA}.ORDERS o ON c.C_CUSTKEY = o.O_CUSTKEY
JOIN {SCHEMA}.LINEITEM l ON o.O_ORDERKEY = l.L_ORDERKEY
WHERE C_MKTSEGMENT = 'BUILDING' AND O_ORDERDATE < '1995-03-15' AND L_SHIPDATE > '1995-03-15'
GROUP BY L_ORDERKEY, O_ORDERDATE, O_SHIPPRIORITY
ORDER BY revenue DESC, O_ORDERDATE
LIMIT 100
"""),
("H6", f"""
SELECT C_CUSTKEY, C_NAME, C_ACCTBAL,
SUM(O_TOTALPRICE) AS total_spend,
COUNT(O_ORDERKEY) AS num_orders,
NTILE(100) OVER (ORDER BY SUM(O_TOTALPRICE) DESC) AS spend_percentile
FROM {SCHEMA}.CUSTOMER c JOIN {SCHEMA}.ORDERS o ON c.C_CUSTKEY = o.O_CUSTKEY
GROUP BY C_CUSTKEY, C_NAME, C_ACCTBAL
ORDER BY total_spend DESC
LIMIT 200
"""),
("H7", f"""
SELECT supp_nation, cust_nation, l_year, SUM(volume) AS revenue
FROM (
SELECT n1.N_NAME AS supp_nation, n2.N_NAME AS cust_nation, YEAR(L_SHIPDATE) AS l_year,
L_EXTENDEDPRICE * (1 - L_DISCOUNT) AS volume
FROM {SCHEMA}.SUPPLIER s
JOIN {SCHEMA}.LINEITEM l ON s.S_SUPPKEY = l.L_SUPPKEY
JOIN {SCHEMA}.ORDERS o ON l.L_ORDERKEY = o.O_ORDERKEY
JOIN {SCHEMA}.CUSTOMER c ON o.O_CUSTKEY = c.C_CUSTKEY
JOIN {SCHEMA}.NATION n1 ON s.S_NATIONKEY = n1.N_NATIONKEY
JOIN {SCHEMA}.NATION n2 ON c.C_NATIONKEY = n2.N_NATIONKEY
WHERE ((n1.N_NAME = 'FRANCE' AND n2.N_NAME = 'GERMANY') OR (n1.N_NAME = 'GERMANY' AND n2.N_NAME = 'FRANCE'))
AND L_SHIPDATE BETWEEN '1995-01-01' AND '1996-12-31'
) shipping
GROUP BY supp_nation, cust_nation, l_year ORDER BY supp_nation, cust_nation, l_year
"""),
("H8", f"""
SELECT P_BRAND, P_TYPE, P_SIZE, COUNT(DISTINCT PS_SUPPKEY) AS supplier_cnt,
RANK() OVER (ORDER BY COUNT(DISTINCT PS_SUPPKEY) DESC) AS rnk
FROM {SCHEMA}.PARTSUPP ps JOIN {SCHEMA}.PART p ON ps.PS_PARTKEY = p.P_PARTKEY
WHERE P_BRAND <> 'Brand#45'
AND P_TYPE NOT LIKE 'MEDIUM POLISHED%'
AND P_SIZE IN (49, 14, 23, 45, 19, 3, 36, 9)
AND PS_SUPPKEY NOT IN (SELECT S_SUPPKEY FROM {SCHEMA}.SUPPLIER WHERE S_COMMENT LIKE '%Customer%Complaints%')
GROUP BY P_BRAND, P_TYPE, P_SIZE
ORDER BY supplier_cnt DESC, P_BRAND, P_TYPE, P_SIZE
"""),
]
THREAD_ASSIGNMENTS = [
[HEAVY_QUERIES[0], MEDIUM_QUERIES[0], LIGHT_QUERIES[0]],
[LIGHT_QUERIES[1], HEAVY_QUERIES[1], MEDIUM_QUERIES[1]],
[MEDIUM_QUERIES[2], LIGHT_QUERIES[2], HEAVY_QUERIES[2]],
[HEAVY_QUERIES[3], LIGHT_QUERIES[3], MEDIUM_QUERIES[3]],
[LIGHT_QUERIES[4], MEDIUM_QUERIES[4], HEAVY_QUERIES[4]],
[MEDIUM_QUERIES[5], HEAVY_QUERIES[5], LIGHT_QUERIES[5]],
[HEAVY_QUERIES[6], MEDIUM_QUERIES[6], LIGHT_QUERIES[6]],
[LIGHT_QUERIES[7], MEDIUM_QUERIES[7], HEAVY_QUERIES[7]],
]
ALL_QUERIES = [q for thread_queries in THREAD_ASSIGNMENTS for q in thread_queries]
def init_connection(conn):
cur = conn.cursor()
cur.execute(f"USE WAREHOUSE {WAREHOUSE}")
cur.execute("ALTER SESSION SET USE_CACHED_RESULT = FALSE")
return conn
def open_connections(n):
print(f" Opening {n} connections...", end="", flush=True)
conns = []
for _ in range(n):
conn = snowflake.connector.connect(connection_name=CONNECTION_NAME)
init_connection(conn)
conns.append(conn)
print(" done.")
return conns
def run_query_solo(conn, label, sql):
cur = conn.cursor()
tag = f"{RUN_ID}:solo:{label}"
cur.execute(f"ALTER SESSION SET QUERY_TAG = '{tag}'")
start = time.time()
try:
cur.execute(sql)
cur.fetchall()
qid = cur.sfqid
finally:
elapsed = time.time() - start
return label, elapsed, qid
def run_thread_batch(conn, queries):
results = []
cur = conn.cursor()
for label, sql in queries:
tag = f"{RUN_ID}:conc:{label}"
cur.execute(f"ALTER SESSION SET QUERY_TAG = '{tag}'")
start = time.time()
try:
cur.execute(sql)
cur.fetchall()
qid = cur.sfqid
finally:
elapsed = time.time() - start
results.append((label, elapsed, qid))
return results
def fetch_query_breakdown(conn, query_ids):
cur = conn.cursor()
ids_str = ",".join(f"'{qid}'" for qid in query_ids)
cur.execute(f"""
SELECT QUERY_ID, QUERY_TAG,
QUEUED_OVERLOAD_TIME,
QUEUED_PROVISIONING_TIME,
COMPILATION_TIME,
EXECUTION_TIME,
TOTAL_ELAPSED_TIME
FROM TABLE(SNOWFLAKE.INFORMATION_SCHEMA.QUERY_HISTORY())
WHERE QUERY_ID IN ({ids_str})
""")
rows = cur.fetchall()
breakdown = {}
for row in rows:
qid = row[0]
tag = row[1] or ""
parts = tag.split(":")
label = parts[2] if len(parts) >= 3 else qid
breakdown[label] = {
"queued_overload_ms": row[2],
"queued_provisioning_ms": row[3],
"compilation_ms": row[4],
"execution_ms": row[5],
"total_elapsed_ms": row[6],
}
return breakdown
def setup_warehouse(conn):
cur = conn.cursor()
print(f"Creating warehouse {WAREHOUSE} (X-Small)...")
cur.execute(f"""
CREATE WAREHOUSE IF NOT EXISTS {WAREHOUSE}
WITH WAREHOUSE_SIZE = 'XSMALL'
AUTO_SUSPEND = 300
AUTO_RESUME = TRUE
INITIALLY_SUSPENDED = FALSE
""")
cur.execute(f"ALTER WAREHOUSE {WAREHOUSE} RESUME IF SUSPENDED")
time.sleep(3)
print("Warehouse ready.\n")
def reset_warehouse(conn):
cur = conn.cursor()
print("Suspending warehouse to clear cache...")
cur.execute(f"ALTER WAREHOUSE {WAREHOUSE} SUSPEND")
time.sleep(5)
cur.execute(f"ALTER WAREHOUSE {WAREHOUSE} RESUME")
time.sleep(3)
print("Warehouse restarted.\n")
def teardown_warehouse(conn):
cur = conn.cursor()
print(f"\nDropping warehouse {WAREHOUSE}...")
cur.execute(f"DROP WAREHOUSE IF EXISTS {WAREHOUSE}")
print("Done.")
def print_results(solo_times, concurrent_times, solo_breakdown, conc_breakdown):
query_labels = [label for label, _ in ALL_QUERIES]
solo_map = {label: t for label, t, _ in solo_times}
conc_map = {label: t for label, t, _ in concurrent_times}
hdr = (f"{'Query':<7} {'Type':<7} {'Solo':>7} {'Conc':>7} {'Slow':>6} "
f"{'Queued':>8} {'Compile':>9} {'Exec':>8} {'Total':>8} {'Queue%':>7}")
sep = "-" * len(hdr)
print("\n" + sep)
print("RESULTS: Solo vs Concurrent — with Server-Side Breakdown (concurrent phase)")
print(f"Warehouse: {WAREHOUSE} (X-Small) | Threads: 8 | Queries/thread: 3 | Run ID: {RUN_ID}")
print(sep)
print(hdr)
print(f"{'':>22} {'(s)':>7} {'(s)':>7} {'':>6} {'(ms)':>8} {'(ms)':>9} {'(ms)':>8} {'(ms)':>8}")
print(sep)
slowdowns = []
total_queued_ms = 0
total_elapsed_ms = 0
for label in query_labels:
solo_t = solo_map.get(label, 0)
conc_t = conc_map.get(label, 0)
slowdown = conc_t / solo_t if solo_t > 0 else 0
slowdowns.append(slowdown)
qtype = "Light" if label.startswith("L") else ("Medium" if label.startswith("M") else "Heavy")
bd = conc_breakdown.get(label, {})
queued = bd.get("queued_overload_ms", 0) + bd.get("queued_provisioning_ms", 0)
compile_ms = bd.get("compilation_ms", 0)
exec_ms = bd.get("execution_ms", 0)
total_ms = bd.get("total_elapsed_ms", 0)
queue_pct = (queued / total_ms * 100) if total_ms > 0 else 0
total_queued_ms += queued
total_elapsed_ms += total_ms
print(f"{label:<7} {qtype:<7} {solo_t:>7.2f} {conc_t:>7.2f} {slowdown:>5.2f}x "
f"{queued:>8} {compile_ms:>9} {exec_ms:>8} {total_ms:>8} {queue_pct:>6.1f}%")
print(sep)
solo_total = sum(t for _, t, _ in solo_times)
conc_wall = max(t for _, t, _ in concurrent_times) if concurrent_times else 0
avg_slowdown = sum(slowdowns) / len(slowdowns) if slowdowns else 0
sorted_sd = sorted(slowdowns)
p50 = sorted_sd[len(sorted_sd) // 2] if sorted_sd else 0
p95_idx = min(int(len(sorted_sd) * 0.95), len(sorted_sd) - 1)
p95 = sorted_sd[p95_idx] if sorted_sd else 0
total_queue_pct = (total_queued_ms / total_elapsed_ms * 100) if total_elapsed_ms > 0 else 0
total_exec_ms = total_elapsed_ms - total_queued_ms
total_exec_pct = 100 - total_queue_pct
print(f"\n{'TIMING SUMMARY'}")
print("-" * 55)
print(f"{'Solo total (sequential sum):':<40} {solo_total:>10.2f}s")
print(f"{'Concurrent wall-clock time:':<40} {conc_wall:>10.2f}s")
print()
print(f"{'Average per-query slowdown:':<40} {avg_slowdown:>10.2f}x")
print(f"{'Median (p50) slowdown:':<40} {p50:>10.2f}x")
print(f"{'p95 slowdown:':<40} {p95:>10.2f}x")
print()
print(f"{'QUEUING vs EXECUTION BREAKDOWN (concurrent phase)'}")
print("-" * 55)
print(f"{'Total queuing time (all 24 queries):':<40} {total_queued_ms:>8}ms ({total_queue_pct:.1f}%)")
print(f"{'Total non-queue time (compile+exec):':<40} {total_exec_ms:>8}ms ({total_exec_pct:.1f}%)")
print(f"{'Total server elapsed (all 24 queries):':<40} {total_elapsed_ms:>8}ms")
print(f"{'Avg queuing per query:':<40} {total_queued_ms / 24:>8.0f}ms")
print()
def main():
parser = argparse.ArgumentParser(description="Snowflake concurrency performance test")
parser.add_argument("--keep-warehouse", action="store_true", help="Don't drop the warehouse after the test")
args = parser.parse_args()
admin_conn = snowflake.connector.connect(connection_name=CONNECTION_NAME)
all_conns = []
try:
setup_warehouse(admin_conn)
print("Pre-opening connections to avoid OAuth timeouts during test...")
solo_conn = open_connections(1)[0]
conc_conns = open_connections(8)
all_conns = [solo_conn] + conc_conns
print()
print("=" * 60)
print(f"PHASE 1: Solo Baseline (24 queries, one at a time) [run={RUN_ID}]")
print("=" * 60)
solo_results = []
solo_qids = []
for i, (label, sql) in enumerate(ALL_QUERIES, 1):
print(f" [{i:>2}/24] Running {label}...", end="", flush=True)
lbl, elapsed, qid = run_query_solo(solo_conn, label, sql)
solo_results.append((lbl, elapsed, qid))
solo_qids.append(qid)
print(f" {elapsed:.2f}s")
print("\nFetching solo server-side breakdown...")
solo_breakdown = fetch_query_breakdown(admin_conn, solo_qids)
reset_warehouse(admin_conn)
print("=" * 60)
print("PHASE 2: Concurrent (8 threads x 3 queries each)")
print("=" * 60)
concurrent_results = []
conc_qids = []
overall_start = time.time()
with ThreadPoolExecutor(max_workers=8) as pool:
futures = {}
for tid, queries in enumerate(THREAD_ASSIGNMENTS, 1):
labels = ", ".join(q[0] for q in queries)
print(f" Thread {tid}: {labels}")
futures[pool.submit(run_thread_batch, conc_conns[tid - 1], queries)] = tid
for future in as_completed(futures):
tid = futures[future]
results = future.result()
for label, elapsed, qid in results:
concurrent_results.append((label, elapsed, qid))
conc_qids.append(qid)
print(f" Thread {tid} | {label}: {elapsed:.2f}s")
overall_elapsed = time.time() - overall_start
print(f"\n Total concurrent wall-clock time: {overall_elapsed:.2f}s")
print("\nFetching concurrent server-side breakdown...")
conc_breakdown = fetch_query_breakdown(admin_conn, conc_qids)
print_results(solo_results, concurrent_results, solo_breakdown, conc_breakdown)
finally:
for c in all_conns:
try:
c.close()
except Exception:
pass
if not args.keep_warehouse:
teardown_warehouse(admin_conn)
else:
print(f"\nWarehouse {WAREHOUSE} kept alive (will auto-suspend in 5 min).")
admin_conn.close()
if __name__ == "__main__":
main()