Comparing Thousands of Snowflake Tables in Minutes (for Pennies)
Fingerprint every table with HASH_AGG, diff only the failures with EXCEPT — and watch out for the duplicate-row blind spot.
The problem
An ingestion pipeline changes, the same source tables get loaded into two Snowflake schemas, and someone has to prove that thousands of tables in schema B are identical to schema A. A typical case: the pipeline switches its file format from CSV to Parquet, each load lands in its own schema, and the switch can’t go live until every Parquet-loaded table is proven identical to its CSV twin.
Diffing every table is slow, expensive, and mostly wasted — the vast majority will match. Use a funnel instead: a cheap test clears the ~90% that are fine; real compute goes only to the tables that differ.
Stage 1 (cheap, all tables) Stage 2 (expensive, mismatches only)
┌────────────────────────────┐ ┌─────────────────────────────────┐
│ HASH_AGG fingerprint per │ │ EXCEPT / MINUS in both │
│ table pair + row counts │ ───► │ directions to count and │
│ batched + parallelized │ │ inspect the differing rows │
└────────────────────────────┘ └─────────────────────────────────┘
thousands of pairs only the failuresRepresentative numbers from a real run (rounded):
Stage 1: Fingerprint every table with HASH_AGG
HASH_AGG reduces an entire table to a single 64-bit hash, computed inside Snowflake — no data movement. Row order doesn't affect the result; duplicate rows do (that matters later).
Exclude metadata columns that legitimately differ (load timestamps, batch IDs) with SELECT * EXCLUDE (...), and grab COUNT(*) for both sides in the same query — it pays off later:
WITH a AS (SELECT * EXCLUDE (_LOAD_TIMESTAMP) FROM prod_db.staging_a.orders),
b AS (SELECT * EXCLUDE (_LOAD_TIMESTAMP) FROM prod_db.staging_b.orders)
SELECT
(SELECT HASH_AGG(*) FROM a) = (SELECT HASH_AGG(*) FROM b) AS is_equal,
(SELECT COUNT(*) FROM a) AS row_count_a,
(SELECT COUNT(*) FROM b) AS row_count_b;One comparison per query means thousands of round-trips. Packing 25 into one query with UNION ALL cuts that to a couple of hundred:
SELECT 'orders' AS table_name, (...) AS is_equal, ... , 0 AS batch_order
UNION ALL
SELECT 'customers', (...), ... , 1
-- ... 23 more ...
ORDER BY batch_order;Finally, a ThreadPoolExecutor keeps several batches in flight at once — plain Python threads sharing one session are enough (thread-safe in recent Snowpark versions), and on an X-Small warehouse 3–4 concurrent batches is the sweet spot.
The full Stage 1 code:
import os
import logging
import concurrent.futuresimport pandas as pd
from cryptography.hazmat.primitives import serialization
from snowflake.snowpark import Sessionlogging.basicConfig(level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)DATABASE = "PROD_DB"
SCHEMA_A = "STAGING_A" # baseline pipeline (e.g., CSV loads)
SCHEMA_B = "STAGING_B" # candidate pipeline (e.g., Parquet loads)
EXCLUDE_COLUMNS = ["_LOAD_TIMESTAMP"] # metadata allowed to differ# --- Connect with key-pair auth; credentials come from the environment ---
with open(os.environ["SNOWFLAKE_PRIVATE_KEY_PATH"], "rb") as f:
private_key = serialization.load_pem_private_key(f.read(), password=None)pkb = private_key.private_bytes(
encoding=serialization.Encoding.DER,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
)session = Session.builder.configs({
"account": os.environ["SNOWFLAKE_ACCOUNT"],
"user": os.environ["SNOWFLAKE_USER"],
"private_key": pkb,
"role": os.environ.get("SNOWFLAKE_ROLE", "VALIDATION_ROLE"),
"warehouse": os.environ.get("SNOWFLAKE_WAREHOUSE", "VALIDATION_WH"),
"database": DATABASE,
}).create()# --- Discover tables and pair them by name ---
def list_tables(session, database, schema):
rows = session.sql(f"SHOW TABLES IN SCHEMA {database}.{schema}").collect()
return {row["name"] for row in rows}tables_a = list_tables(session, DATABASE, SCHEMA_A)
tables_b = list_tables(session, DATABASE, SCHEMA_B)pairs = [(f"{DATABASE}.{SCHEMA_A}.{n}", f"{DATABASE}.{SCHEMA_B}.{n}")
for n in sorted(tables_a & tables_b)]print(f"{len(pairs)} pairs to compare | "
f"{len(tables_a - tables_b)} only in {SCHEMA_A} | "
f"{len(tables_b - tables_a)} only in {SCHEMA_B}")# --- Batched, parallel HASH_AGG comparison ---
EXCLUDE_CLAUSE = f" EXCLUDE ({', '.join(EXCLUDE_COLUMNS)})" if EXCLUDE_COLUMNS else ""def compare_pair_sql(table_a, table_b, order):
return f"""
SELECT
'{table_a}' AS table_a,
'{table_b}' AS table_b,
(
WITH a AS (SELECT *{EXCLUDE_CLAUSE} FROM {table_a}),
b AS (SELECT *{EXCLUDE_CLAUSE} FROM {table_b})
SELECT (SELECT HASH_AGG(*) FROM a) = (SELECT HASH_AGG(*) FROM b)
) AS is_equal,
(SELECT COUNT(*) FROM {table_a}) AS row_count_a,
(SELECT COUNT(*) FROM {table_b}) AS row_count_b,
{order} AS batch_order"""def compare_single(session, table_a, table_b):
"""Per-pair fallback so one bad table can't sink a whole batch."""
try:
r = session.sql(compare_pair_sql(table_a, table_b, 0)).collect()[0]
return {"table_a": table_a, "table_b": table_b,
"is_equal": r["IS_EQUAL"],
"row_count_a": r["ROW_COUNT_A"],
"row_count_b": r["ROW_COUNT_B"]}
except Exception as exc:
return {"table_a": table_a, "table_b": table_b, "is_equal": False,
"row_count_a": None, "row_count_b": None, "error": str(exc)}def compare_batch(session, batch):
query = " UNION ALL ".join(
compare_pair_sql(a, b, i) for i, (a, b) in enumerate(batch)
) + " ORDER BY batch_order"
try:
return [{"table_a": r["TABLE_A"], "table_b": r["TABLE_B"],
"is_equal": r["IS_EQUAL"],
"row_count_a": r["ROW_COUNT_A"],
"row_count_b": r["ROW_COUNT_B"]}
for r in session.sql(query).collect()]
except Exception as exc:
logger.error(f"Batch failed ({exc}); retrying tables individually")
return [compare_single(session, a, b) for a, b in batch]def compare_all(session, pairs, batch_size=25, max_workers=3):
batches = [pairs[i:i + batch_size]
for i in range(0, len(pairs), batch_size)]
logger.info(f"Comparing {len(pairs)} pairs in {len(batches)} batches") results, done = [], 0
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as pool:
futures = [pool.submit(compare_batch, session, b) for b in batches]
for future in concurrent.futures.as_completed(futures):
results.extend(future.result())
done += 1
if done % 10 == 0 or done == len(batches):
logger.info(f"Progress: {done}/{len(batches)} batches")
return pd.DataFrame(results)report = compare_all(session, pairs)
report.to_csv("table_comparison_results.csv", index=False)matches = (report["is_equal"] == True).sum()
print(f"{matches}/{len(report)} tables match "
f"({matches / len(report) * 100:.1f}%)")
Note the fallback: if any table in a batch is broken, the batch is retried one table at a time, so one bad object costs one result, not 25.
Get Zia Said’s stories in your inbox
Join Medium for free to get updates from this writer.
In a real run, roughly 90% of ~5,000 pairs were proven identical on the spot — only a few hundred moved on to Stage 2.
Stage 2: Explain the mismatches with EXCEPT / MINUS
A hash says that two tables differ, not how. For the failures, run MINUS (interchangeable with EXCEPT in Snowflake) in both directions — rows can exist only in A, only in B, or both:
def common_columns(session, table_a, table_b):
"""Align on shared columns so schema drift doesn't break the diff."""
def cols(fq_name):
db, schema, tbl = fq_name.split(".")
rows = session.sql(f"""
SELECT column_name
FROM {db}.INFORMATION_SCHEMA.COLUMNS
WHERE table_schema = '{schema}' AND table_name = '{tbl}'
""").collect()
return {r["COLUMN_NAME"] for r in rows}
return sorted((cols(table_a) & cols(table_b)) - set(EXCLUDE_COLUMNS))mismatches = report[report["is_equal"] == False]
print(f"Diffing {len(mismatches)} mismatched pairs...")diffs = []
for _, row in mismatches.iterrows():
a, b = row["table_a"], row["table_b"]
col_list = ", ".join(f'"{c}"' for c in common_columns(session, a, b)) a_minus_b = f"SELECT {col_list} FROM {a} MINUS SELECT {col_list} FROM {b}"
b_minus_a = f"SELECT {col_list} FROM {b} MINUS SELECT {col_list} FROM {a}" diffs.append({
"table_a": a,
"table_b": b,
"rows_only_in_a":
session.sql(f"SELECT COUNT(*) AS c FROM ({a_minus_b})").collect()[0]["C"],
"rows_only_in_b":
session.sql(f"SELECT COUNT(*) AS c FROM ({b_minus_a})").collect()[0]["C"],
})diff_report = pd.DataFrame(diffs)
diff_report.to_csv("table_difference_summary.csv", index=False)
The common_columns helper diffs only the columns both tables share, in a deterministic order. The output is a triage sheet: tables with a handful of differing rows get eyeballed; tables with thousands go back to the pipeline team as bugs.
The plot twist: EXCEPT can’t see duplicates
On one real migration, the Stage 2 numbers looked absurd: of well over a thousand tables that HASH_AGG flagged as different, more than 80% showed zero differing rows in both directions. HASH_AGG said “different”; EXCEPT said “identical.” Which one was lying?
Neither. The row counts told the story:
Every “phantom mismatch” was a duplicate-row problem: one pipeline had loaded the same row twice. HASH_AGG is sensitive to duplicates, so it correctly flagged the tables. But EXCEPT/MINUS is a set operator — it de-duplicates before comparing, and {row, row} minus {row} is empty. Both tools were right; they answer different questions.
Snowflake has no EXCEPT ALL. For a duplicate-aware diff, compare counted rows instead:
SELECT c1, c2, c3, COUNT(*) AS occurrences
FROM table_a GROUP BY ALL
MINUS
SELECT c1, c2, c3, COUNT(*) AS occurrences
FROM table_b GROUP BY ALL;A table-comparison pipeline needs three signals — HASH_AGG as the cheap "anything different?" alarm, COUNT(*) for volume and duplicates, EXCEPT for which distinct rows differ. Together they classify every mismatch: counts differ + empty EXCEPT = duplicates; counts equal + non-empty EXCEPT = changed values; both = missing or extra data.
Takeaways
- Fingerprint first, diff second. Spend real compute only on tables that fail the cheap test.
- Batch and parallelize. 25 comparisons per query and 3 threads turn thousands of round-trips into minutes.
- Collect row counts alongside hashes. Nearly free, and they expose duplicate problems instantly.
- Know your operators’ blind spots.
EXCEPTsilently ignores duplicate multiplicity;GROUP BY ALL+COUNT(*)closes the gap. - Make validation cheap enough to repeat. At minutes and pennies per sweep, you can re-validate after every pipeline fix — and ship the migration with receipts.
All account, schema, and table names are generic placeholders; performance figures are rounded from real-world runs of this approach.
