Failure Modes and Production Edge Cases

Statistics Gone Wrong: The most catastrophic failures stem from incorrect statistics. When table cardinalities are off by 10x or more, the optimizer makes decisions that destroy performance or crash executors entirely. Consider a broadcast join decision. The optimizer sees a dimension table estimated at 500 MB and decides to broadcast it to all executors. But if statistics are stale and the table has grown to 50 GB, every executor tries to materialize 50 GB in memory. With 10 GB executor memory, this triggers out of memory errors or forces spilling to disk, turning a fast broadcast into a slow shuffle that takes 100x longer. The reverse problem is equally bad. Overestimating table sizes prevents beneficial broadcast joins, forcing full shuffles of multi terabyte tables. A query that should complete in minutes instead shuffles petabytes across the network for hours, potentially overwhelming the cluster. Plan Explosion for Complex Queries: Queries with dozens of joins or deeply nested subqueries generate enormous logical plans. Rule based optimization can enter a state explosion where each rule application creates new opportunities for other rules, leading to thousands of optimization iterations. Some production systems have seen optimization take 10 to 30 seconds for queries with 50 plus joins. In interactive environments where users expect results in under 5 seconds, the planning time alone violates latency targets. Most systems enforce caps on optimization iterations or plan complexity, accepting suboptimal but "good enough" plans. Data Skew and Stragglers: Tungsten's tight loops and efficient execution make skew problems worse in relative terms. When 99% of tasks complete in 30 seconds but one hot partition takes 50 minutes, that single straggler dominates total query latency. A 50 TB table with zipfian distribution might have one partition containing 5% of all data while others have 0.001%. The task processing that hot partition runs 5000x longer than peers. Whole stage code generation does not help when the problem is data imbalance, not CPU efficiency. Code Generation Limits: Extremely complex queries can generate methods exceeding Java Virtual Machine (JVM) bytecode size limits, typically around 64 KB per method. When this happens, the JVM may refuse to load the class or deoptimize the code, falling back to slow interpretation. Some systems mitigate this by splitting generated code into smaller methods or selectively disabling code generation for pathologically complex stages. But this is a band aid; the root issue is that code generation assumes moderately complex query shapes. Memory Leak Risks: Custom memory management trades safety for performance. Off heap memory leaks are harder to debug than garbage collected heap leaks. A small leak in a tight loop processing billions of rows can exhaust cluster memory in minutes. Unlike garbage collection which eventually reclaims unreachable objects, explicitly managed memory stays allocated until you free it. Bugs in complex generated code can leak memory on error paths or edge cases that are hard to test. Production systems need robust monitoring of off heap memory usage and aggressive automated restarts when leaks are detected. Mitigation Strategies: Keep statistics fresh with automated periodic recomputation. Use dynamic broadcast join thresholds that adapt based on executor memory. Implement salting or custom partitioning schemes to address skew. Limit code generation complexity and fall back to interpretation for extreme cases. Express logic as built in functions rather than UDFs whenever possible. Monitor off heap memory closely and set conservative limits.