You ship the job. It passes CI. The data quality checks return green. Then, at 3:14 AM, the PagerDuty alert fires. Your GCP billing dashboard is currently reporting a "spend anomaly," and your Databricks SQL Warehouse is throwing an OUT_OF_MEMORY error that is currently cascading into a service-wide outage.
I’ve been here. Twice. Once on BigQuery, once on Databricks. They aren't the same beast, and treating them as interchangeable "SQL engines" is exactly how you end up explaining a $4,000 hourly burn rate to a VP who doesn't care about your partition pruning strategy.
What we saw
It started with a simple request: join our 4TB events table with a 500GB user_metadata table. In our staging environment, which used a 10% sample of the data, the query finished in 42 seconds. In production, it didn't finish. It hung for 45 minutes, then the BigQuery slot usage spiked to 10,000+ slots, and the Databricks SQL Warehouse (a 2XL Serverless instance) simply evaporated, dumping a java.lang.OutOfMemoryError: Java heap space into the logs.
The false lead was the index. My junior dev thought we missed a partition key on user_id. We spent two hours adding a CLUSTER BY on the BigQuery side and a Z-Order on the Databricks side. It changed absolutely nothing. The execution plan remained a disaster. The symptoms—massive latency and OOM errors—weren't about data locality. They were about the engine trying to perform a broadcast join on a table that was too damn big to fit in a single worker's memory.
Photo by Salah Ait Mokhtar on Unsplash
Root cause
In BigQuery, the culprit is almost always "Shuffling." When BigQuery can't perform a map-side join, it triggers a massive shuffle phase. I saw the job details: Bytes shuffled: 12.4 TB. BigQuery charges for data processed, but when you hit a join that forces a massive shuffle, you aren't just paying for the read; you're paying for the compute slots spinning in circles trying to sort and move data across the network fabric.
On Databricks, the mechanism is different but equally lethal. We were using a Databricks SQL Warehouse with AUTO join optimization. The Catalyst optimizer looked at our 500GB metadata table and incorrectly estimated it could fit in the memory of a single node for a broadcast join. It couldn't. The executor tried to pull the entire 500GB into memory, hit the heap limit, and triggered a hard OOM. The spark.sql.autoBroadcastJoinThreshold default is 10MB, but our specific configuration had been bumped to 1GB by a previous team trying to "speed up" smaller queries. It was a ticking time bomb.
Photo by Gorilla ROI Data Connector on Unsplash
The fix
We stopped trying to outsmart the optimizers and forced the physical plan.
For BigQuery, we implemented a sub-query re-write. Instead of a standard JOIN, we used a JOIN with a WHERE clause that explicitly limited the join keys to a pre-filtered subset of the data. We also switched from SELECT * to specific columns—a boring, manual fix that reduced our shuffle volume from 12TB down to 800GB. We also had to switch the job to INTERACTIVE priority; the BATCH queue was holding onto slots for too long, causing a resource deadlock.
For Databricks, we had to disable the broadcast join for that specific query. We added the hint /*+ MERGE(events, user_metadata) */ to the SQL. This forced the engine to use a Shuffle Sort-Merge Join, which is much slower than a broadcast join but significantly more stable. It spilled to disk instead of exploding the heap. The query took 12 minutes instead of failing, but it finished. I'll take a 12-minute success over a $4,000 failed job any day.
What we changed so it never happens again
First, we implemented a "Query Size Budget." We now have a post-commit hook in our CI pipeline that runs EXPLAIN on every SQL file. If the estimated bytes processed in BigQuery exceed 1TB, or if the Databricks plan shows a BroadcastExchange on a table larger than 500MB, the PR is automatically blocked. You don't get to deploy code that could bankrupt the department.
Second, we stopped using AUTO settings for warehouse sizing. We now use specific warehouse tags for specific workloads. If a job is heavy on joins, it runs on a warehouse with enable_serverless_compute = true but with min_serverless_worker_count set to a higher floor. We treat the warehouse like a dedicated resource, not a magic bucket.
Third, we enforced a culture of looking at the "Query Profile." In BigQuery, that means checking the "Wait time" and "Shuffle stage" metrics in the UI. In Databricks, it means looking at the "Metrics" tab in the SQL query history to see how much data was spilled to disk. If you see "Spilled to disk," you haven't failed; you've just found the physical limit of your cluster.
Finally, I stopped trusting the "Big Data" marketing. BigQuery and Databricks are incredible tools, but they assume you know how a distributed join works. If you treat them like a MySQL instance on a laptop, they will treat your credit card like an unlimited ATM. Don't look at the execution time. Look at the data movement. Data movement is the tax you pay for poor query design. Keep the data local, keep the shuffles minimal, and for the love of god, keep an eye on your broadcast thresholds.
Cover photo by Albert Stoynov on Unsplash.
Top comments (0)