DEV Community: Maithreyan

My PySpark Pipeline Was Reading the Same Table 3 Times — Here's Why

Maithreyan — Mon, 31 Aug 2026 16:21:38 +0000

My PySpark pipeline was slow. I assumed it needed more cluster resources. It actually needed one line of code.

I had a job that read from a source table, applied a chain of transformations, and then used that same transformed DataFrame in three different places downstream: one write to storage, one aggregation for a summary table, and one validation check. Straightforward pattern. The pipeline worked, but it was taking far longer than the data volume justified.

When I looked at the Spark UI, the pattern became obvious. The source table was being read multiple times, and the exact same chain of transformations was showing up again and again across the job's stages — not once, but three separate times, once for each downstream action that touched the DataFrame.

This post covers why that happens, how to spot it, and the actual fix.

The root cause: transformations are lazy

Spark separates DataFrame operations into two categories: transformations and actions.

Transformations — .filter(), .select(), .join(), .withColumn(), .groupBy() — don't execute anything when you call them. They just build up a lineage: a directed acyclic graph (DAG) describing what would happen if you asked Spark to actually compute the result.

Actions — .count(), .collect(), .show(), .write() — are what actually trigger execution. When you call an action, Spark walks the lineage graph and executes every transformation needed to produce that result.

Here's the part that catches people off guard: every action re-triggers the entire lineage from scratch, all the way back to the original data source, unless you've explicitly told Spark to remember an intermediate result. Spark doesn't automatically cache anything for you between actions.

What this looked like in practice

Here's a simplified version of the pattern I had:

transformed_df = (
    source_df
    .filter(F.col("status") == "active")
    .join(reference_df, "id")
    .withColumn("flag", F.when(F.col("amount") > 1000, 1).otherwise(0))
)

transformed_df.write.parquet("output/main")

summary_df = transformed_df.groupBy("flag").count()
summary_df.show()

invalid_count = transformed_df.filter(F.col("amount") < 0).count()

There are three actions here: .write(), .show(), and .count(). Each one is a separate trigger for Spark to evaluate the lineage of transformed_df. That means:

The source table gets read from disk three separate times.
The filter, join, and withColumn transformations each run three separate times.
Any shuffle involved in the join happens three separate times.

None of this shows up as an error. The job completes successfully. It's just doing two extra passes of work that were entirely avoidable.

Spotting it in the Spark UI

The giveaway is in the Spark UI's SQL / Jobs tab. If you see the same source table scan and the same sequence of transformation stages repeated multiple times across what should logically be one pipeline, that's usually a sign the same DataFrame is being recomputed for each action instead of being reused.

A quick sanity check: count how many actions your pipeline calls on a given DataFrame, and compare that to how many times the source read shows up in the UI. If they match, you've found your redundant computation.

The fix: cache() or persist()

The fix is to tell Spark to materialize and store the result of the transformation chain the first time it's computed, so subsequent actions reuse that stored result instead of recomputing everything.

transformed_df = (
    source_df
    .filter(F.col("status") == "active")
    .join(reference_df, "id")
    .withColumn("flag", F.when(F.col("amount") > 1000, 1).otherwise(0))
    .cache()
)

transformed_df.count()  # first action materializes the cache

transformed_df.write.parquet("output/main")

summary_df = transformed_df.groupBy("flag").count()
summary_df.show()

invalid_count = transformed_df.filter(F.col("amount") < 0).count()

One important detail: .cache() itself is lazy too. Calling it doesn't immediately store anything — it just marks the DataFrame as something to be cached the next time an action runs. That's why I added an explicit .count() right after .cache(), to force the first materialization at a predictable point rather than letting it happen implicitly on whichever action runs first.

After that first action, every subsequent action on transformed_df reads from the cached result instead of re-reading the source table and re-running the transformation chain.

cache() vs. persist()

These aren't two different mechanisms — cache() is a convenience method that calls persist() with a default storage level.

For DataFrames, cache() is equivalent to persist(StorageLevel.MEMORY_AND_DISK). This tries to keep the cached data in executor memory, and spills to disk if it doesn't fit.

persist() lets you choose the storage level explicitly:

from pyspark import StorageLevel

transformed_df.persist(StorageLevel.MEMORY_ONLY)
transformed_df.persist(StorageLevel.MEMORY_AND_DISK)
transformed_df.persist(StorageLevel.DISK_ONLY)
transformed_df.persist(StorageLevel.MEMORY_AND_DISK_SER)

Use MEMORY_ONLY if you're confident the dataset fits comfortably in memory and want the fastest possible reuse. Use MEMORY_AND_DISK (or just .cache()) as a safer default when you're not certain. Use serialized variants (_SER suffix) if memory pressure is a concern and you're willing to trade some CPU time for a smaller memory footprint.

Things I learned that I wish I'd known earlier

Caching isn't automatically a win. Every cached DataFrame occupies executor memory. If you cache too many large DataFrames across a pipeline, you create memory pressure that can slow the job down rather than speed it up — the opposite of the intended effect.

If it doesn't fit in memory, the benefit shrinks. If a cached DataFrame is too large for available memory and has to spill to disk, the performance gain over recomputation can become marginal. It's still often better than recomputing an expensive join or shuffle from scratch, but it's not the dramatic speedup you'd get from an in-memory cache hit.

Don't forget to release it. Calling .unpersist() once you're done with a cached DataFrame frees up that memory for the rest of the job. Skipping this step means the memory stays reserved even after you no longer need the cached data, which can starve other operations running later in the same job.

transformed_df.unpersist()

Cache after your transformation chain is final, not mid-chain. If you cache an intermediate DataFrame and then keep adding more transformations on top of it, you may end up caching a version of the data you don't actually need cached, while the version that's genuinely reused downstream never gets the benefit.

Only cache what's actually reused. If a DataFrame is used in exactly one action, caching it adds overhead (the cost of materializing and storing it) without any payoff, since there's no second computation to avoid.

A simple rule of thumb

Cache a DataFrame when all of the following are true:

It's referenced by more than one action downstream.
Recomputing it is expensive — it involves a join, a wide shuffle, or a costly aggregation.
It comfortably fits within your cluster's available memory, or you're deliberately choosing a disk-backed storage level as a tradeoff.

Skip caching when a DataFrame is used exactly once, when it's cheap to recompute, or when your cluster is already under memory pressure from other cached data.

The broader lesson

Spark's lazy evaluation model is a deliberate design choice — it lets Spark's optimizer look at the full chain of transformations and plan an efficient execution strategy rather than executing operations one at a time as they're written. But that same laziness means the responsibility falls on you to recognize when a DataFrame is about to be computed more than once.

A pipeline that "needs a bigger cluster" is sometimes actually a pipeline doing the same work two or three times over. Throwing more compute at that problem makes the redundant work finish a bit faster — it doesn't eliminate the redundancy. Adding one .cache() call in the right place fixed what more executors never would have.

Have you caught a Spark job silently recomputing the same DataFrame multiple times? What tipped you off — the Spark UI, a runtime that didn't add up, or something else?

95% of My PySpark Job Finished in 4 Minutes. The Last Task Took 40. Here's Why.

Maithreyan — Mon, 31 Aug 2026 12:20:56 +0000

I had a PySpark job joining a large transactions table with a customer dimension table. Nothing exotic — a standard join, then an aggregation. On paper, it looked like it should scale fine across the cluster.

In practice, the job would race through most of its tasks and then stall. The Spark UI told the real story: almost every task finished in a few minutes, but one or two tasks ran for over 40 minutes while their executors sat at 80–90% CPU, and the rest of the cluster sat mostly idle waiting for them to finish.

This post walks through what data skew actually is, how I confirmed it was the cause, and the fix that brought the job back under control.

What data skew actually is

Data skew happens when one or a few keys hold a disproportionate share of the data. When Spark distributes work across partitions — usually via hash partitioning on a join or group-by key — all the rows for a given key land in the same partition. If one key has millions of rows and most others have a few thousand, that one partition (and the single task processing it) ends up doing far more work than every other partition combined.

The result is a job where 95% of tasks look completely healthy, and the remaining 5% become the actual bottleneck. Total job time is dictated by the slowest task, not the average one — so a single skewed key can dominate your entire runtime even if it represents a tiny fraction of your total row count.

How I confirmed it

The Spark UI's stage view was the first clue: a chart of task durations with almost every bar clustered together, and one or two bars stretching far beyond the rest. That pattern — uniform short tasks plus one long outlier — is close to a signature for skew.

To confirm which key was responsible, I ran a simple aggregation on the join key before doing anything else:

from pyspark.sql import functions as F

df.groupBy("customer_id") \
  .count() \
  .orderBy(F.desc("count")) \
  .show(20)

The output made it obvious: a small number of customer IDs had row counts orders of magnitude higher than the rest of the dataset. A handful of high-volume accounts were responsible for a disproportionate share of all transactions, and Spark's default hash partitioning sent every one of those rows into a single partition.

The fix: salting

The technique that actually solved this is called salting. The core idea:

Take the skewed join key and append a random "salt" value to it, splitting what was one hot key into several smaller synthetic keys.
On the smaller side of the join, explode each row into one copy per possible salt value, so every salted variant of the key still has a matching row to join against.
Join on the new salted key instead of the original key.
Drop the salt column once the join is done, since it was only needed to spread the data across partitions.

This doesn't change the result of the join — every row still matches correctly — but it changes how the work is distributed. What used to be one massive partition handling all of a hot customer's rows gets split into several smaller partitions, each processed by a different executor in parallel.

Here's a simplified version of what I used:

from pyspark.sql.functions import rand, concat_ws, col

SALT_BUCKETS = 10

large_df = large_df.withColumn(
    "salt", (rand() * SALT_BUCKETS).cast("int")
)
large_df = large_df.withColumn(
    "join_key", concat_ws("_", col("customer_id"), col("salt"))
)

salt_values = spark.range(0, SALT_BUCKETS).withColumnRenamed("id", "salt")

small_df_expanded = small_df.crossJoin(salt_values)
small_df_expanded = small_df_expanded.withColumn(
    "join_key", concat_ws("_", col("customer_id"), col("salt"))
)

result_df = large_df.join(
    small_df_expanded,
    on="join_key",
    how="left"
).drop("salt", "join_key")

With SALT_BUCKETS = 10, each previously-hot key gets split across 10 partitions instead of 1. The slowest task in the job dropped dramatically — no more single straggler holding up the entire stage while everything else finished and sat idle.

Things that surprised me

Spark never warns you about skew. There's no error, no red flag in the logs — just one task that quietly runs far longer than the rest while the cluster utilization graph looks strangely uneven. You only notice it if you're actually looking at task-level timing in the Spark UI, not just whether the job succeeded.

Salting isn't free. Adding salt buckets increases the amount of data being shuffled and joined, since the smaller side of the join gets duplicated once per bucket. It's worth applying selectively to the specific hot keys causing the problem, not blindly salting an entire dataset — over-salting can add more overhead than it saves.

Check broadcast joins first. If one side of your join is small enough to fit comfortably in executor memory (a common threshold cited is under roughly 500MB, though this depends on your cluster's spark.sql.autoBroadcastJoinThreshold setting), a broadcast join avoids the shuffle — and the skew problem — entirely. It's simpler than salting and should usually be tried first if it's feasible for your data size.

Not all skew comes from joins. The same problem shows up in groupBy aggregations, window functions partitioned by a skewed column, and even repartitioning by a skewed key. The detection method (checking row counts per key) and the fix (salting, or in some cases using Spark's adaptive query execution skew join optimization) generalize beyond just joins.

A quick detection checklist

If your Spark job "succeeds" but feels slower than it should be for the data volume involved, this is a fast way to check for skew before assuming the cluster just needs more resources:

Open the Spark UI and look at the stage's task duration distribution — a long tail with one or two extreme outliers is the classic signature.
Run a groupBy(key).count().orderBy(desc("count")) on your join or aggregation key and look for one or a few values that dominate the row count.
Check executor CPU utilization during the slow stage — if most executors are idle while one or two are pegged, that's a strong secondary signal.
If confirmed, evaluate a broadcast join first if feasible, and fall back to salting for genuinely large-to-large joins where broadcasting isn't an option.

The real lesson

"My job finished" and "my job finished efficiently" are two very different claims. A completed Spark job can still be hiding a massive inefficiency in plain sight, and the only way to catch it is to look past the green checkmark and into task-level timing.

Adding more cluster resources would not have fixed this — the bottleneck wasn't total compute capacity, it was that all the work for the hottest keys was funneled into a single partition no matter how many executors were available. Understanding why a job is slow matters more than just throwing more hardware at it.

Have you run into a skewed join in production? What tipped you off — the Spark UI, a cost spike, or something else?

Why Apache Airflow Instead of Cron? A Deep Dive Into How Airflow Actually Schedules Your DAGs

Maithreyan — Wed, 12 Aug 2026 00:07:25 +0000

"Why not just use a cron job?" is the first question I get whenever someone sees an Airflow DAG. Fair question. Cron works. It's been around for decades. It's simple.

The real answer isn't that cron is bad — it's that cron solves a different problem than Airflow does.

Cron is a job scheduler. It runs a command at a fixed time. That's it. It doesn't know whether the command succeeded, whether its dependencies are satisfied, or whether it should even run at all today. It just fires the command and moves on.

Airflow is a workflow orchestrator. It doesn't just schedule tasks — it models them as a graph of dependencies, tracks their state, retries failed ones, and gives you a UI to see what ran, what failed, and why.

Here's where that difference actually matters.

The problem cron can't solve

Imagine a simple ETL pipeline:

Extract raw data from an API
Validate and clean it
Load into a warehouse
Run a transformation
Send a Slack alert if anything fails

With cron, you'd write five separate cron entries, one per step, and hope the timing works out. If step 2 fails but step 3 runs anyway, you now have bad data in your warehouse. If step 4 takes twice as long one day, you've silently broken your SLA. Nobody gets notified unless you manually add alerting logic to every script.

With Airflow, you model this as a DAG:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

with DAG(
    dag_id="daily_etl",
    schedule="0 6 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

    extract = PythonOperator(task_id="extract", python_callable=extract_data)
    validate = PythonOperator(task_id="validate", python_callable=validate_data)
    load = PythonOperator(task_id="load", python_callable=load_to_warehouse)
    transform = PythonOperator(task_id="transform", python_callable=run_transformation)

    extract >> validate >> load >> transform

Airflow guarantees the order. If validate fails, load and transform never run. You get automatic retries, failure alerts, and a web UI that shows exactly where the pipeline broke and why.

How Airflow actually schedules a DAG

This is where it gets interesting. Airflow doesn't just "run your Python script at a fixed time" the way cron does.

When you define a DAG with a schedule (or schedule_interval in older versions), Airflow doesn't pass that cron expression directly to the OS scheduler. Instead, it converts it into a timetable — an internal object that determines when a DAG run should be created.

The scheduler process runs continuously, checking every few seconds whether any DAGs are ready to run based on their timetable. When a DAG is due, the scheduler creates a DagRun object for that execution date and queues up the tasks. The actual execution happens on worker processes (via the executor you've configured — Local, Celery, or Kubernetes), not directly from the scheduler itself.

This matters for two reasons:

1. Airflow schedules based on data intervals, not wall-clock time.

A DAG with schedule="0 6 * * *" (daily at 6 AM) doesn't run at 6 AM to process data at that moment. It runs at 6 AM to process data for the previous interval — typically yesterday, if you're on a daily schedule. Airflow's execution_date is the start of the data interval, not the time the task actually runs.

This is why catchup=True (the default in older Airflow versions) can surprise you: if you deploy a new daily DAG on January 10th with a start date of January 1st, Airflow will immediately create DagRuns for every day from Jan 1–9 and try to backfill them all, because it thinks you're behind on processing those intervals.

2. The scheduler is stateful and centralized.

Unlike cron, which runs independently on each machine, Airflow's scheduler is a single process (or a small cluster in HA setups) that maintains a global view of all DAGs, their schedules, and their current state. It knows which tasks are running, which are queued, which have failed, and which are blocked by upstream dependencies. This is what enables features like automatic retries, SLA monitoring, and the ability to pause or unpause a DAG from the UI without touching the server.

The part that actually matters in production

The real difference isn't features — it's what happens when things go wrong.

Cron jobs fail silently. Logs are scattered across servers. Backfilling a missed run means manually re-running scripts in the right order. Scaling to 50+ pipelines means managing hundreds of crontab entries across multiple machines, with no central visibility into what's running or what's broken.

Airflow tracks everything — task state, execution history, retry counts, SLAs. You can backfill a date range with one command. You can see which tasks are blocking others. You can add sensors that wait for external data to arrive before starting a downstream task. None of this exists in cron; you'd have to build it yourself, and you'd build it worse than Airflow already has.

I've seen this play out directly: a team running 30+ cron-based ETL scripts had no idea when a critical pipeline silently failed for three days because the script exited with a 0 status code even though the data was stale. Moving to Airflow meant that same pipeline would have automatically retried, alerted on failure, and shown up in red on a dashboard — impossible to miss.

When cron is actually the right choice

I'm not saying "never use cron." Cron is perfect for:

Simple, independent tasks (daily backups, log rotation, health checks)
Small-scale automation (fewer than ~5 scripts, no dependencies between them)
Situations where silent failure is acceptable or you have other monitoring in place

If your pipeline is "run this one script at 2 AM and that's it," cron is fine. If your pipeline has dependencies, retries, alerts, or cross-team visibility needs, Airflow pays for itself quickly.

The takeaway

Cron tells a task when to run. Airflow decides what should run, in which order, and what happens if something fails.

That difference becomes critical the moment your workflows grow beyond "one script, one schedule."

Do you still use cron for anything in production, or has everything moved to an orchestrator?

How SQL Actually Works Under the Hood — A Deep Dive Into Snowflake and Redshift

Maithreyan — Mon, 10 Aug 2026 11:30:00 +0000

Most people think SQL is fast because "the database is optimized." That's true, but it's such a shallow answer it's almost useless. If you've ever had a query run fine on a 50K-row staging table and crawl on a 40M-row production table with the exact same SQL, you already know the real answer is more interesting — and more mechanical — than "the engine is smart."

This post breaks down what actually happens between you hitting Run and getting your rows back, specifically on Snowflake and Redshift, since the two take genuinely different architectural approaches to solving the same problem.

Your SQL is a request, not an instruction

The first thing to internalize: SQL is declarative. You describe what you want, not how to get it. The engine's query optimizer is the component that decides how. It parses your query, checks statistics about your tables (row counts, value distributions, min/max ranges per partition), and evaluates multiple possible execution plans before picking the one it estimates will cost the least.

This is why identical SQL can perform completely differently depending on data size, indexing, or even how recently statistics were refreshed. The query text never changes; the plan behind it does.

Snowflake's architecture: separating storage, compute, and metadata

Snowflake splits itself into three distinct layers, and understanding this split explains almost everything about why it behaves the way it does:

Storage layer — your data sits in cloud object storage (S3, Azure Blob, or GCS) as compressed, columnar micro-partitions, typically 50–500MB of uncompressed data each.
Compute layer — "virtual warehouses" are independent MPP compute clusters you spin up to run queries. Multiple warehouses can query the same underlying data simultaneously without contending for resources, because compute is fully decoupled from storage.
Services layer — this is the part people forget about. It manages metadata, security, and — critically — the optimizer itself.

Why Snowflake queries can be fast without indexes at all

Snowflake doesn't use traditional indexes. Instead, every micro-partition carries metadata about the min/max values of each column stored inside it. When you filter with a WHERE clause, Snowflake's optimizer uses this metadata to skip entire micro-partitions that can't possibly contain matching rows — a technique called partition pruning. If your table is well-clustered on the columns you filter by often, this can mean scanning a tiny fraction of the actual data on disk, even on a table with billions of rows.

Snowflake also processes data using vectorized execution — operating on batches of column values at once rather than row-by-row — and worker nodes exchange data directly during joins, avoiding some of the shuffle-heavy overhead that plagues naive distributed joins.

The catch

None of this is free if your data isn't organized well. If a table isn't clustered on the columns you actually filter by, pruning doesn't help much, and Snowflake falls back to scanning far more micro-partitions than necessary — which shows up directly in your compute cost, since Snowflake bills by warehouse runtime.

Redshift's architecture: a leader node and a fleet of workers

Redshift takes a more classically MPP (massively parallel processing) approach, structured around actual physical clusters rather than fully decoupled layers:

Leader node — receives your query, parses it, builds the execution plan, and compiles the plan into executable code. It also coordinates communication with your SQL client and handles a small set of functions that run exclusively on the leader node itself.
Compute nodes — each stores a slice of your table's data and executes its portion of the plan in parallel. Results get sent back to the leader node for final aggregation.

Why distribution keys and sort keys matter so much

Because Redshift's compute nodes each own a physical slice of the data, how that data is distributed across nodes directly determines whether a join can run locally on each node or requires shuffling rows across the network first. If you join two tables on a column that isn't the distribution key, Redshift has to redistribute rows across nodes to align matching keys before the join can proceed — this is often the single biggest hidden cost in a slow Redshift query.

Sort keys work similarly to Snowflake's micro-partition metadata: Redshift tracks the min/max range of sorted columns per block, and can skip blocks that fall outside your filter range entirely — this is Redshift's version of partition pruning, and it depends entirely on your table being sorted on the columns you commonly filter by.

Compiled code, not interpreted SQL

One detail that surprises people: Redshift doesn't re-interpret your SQL text every time you run a query. The leader node compiles the execution plan into actual executable code tailored to your specific query and schema, and that compiled code can be reused for repeated executions of similar queries — meaning stable, frequently-run queries genuinely get faster the more they're run.

Where the two architectures actually diverge in practice

Aspect	Snowflake	Redshift
Compute/storage	Fully decoupled, independently scalable	Coupled to node/cluster you provision
Data organization	Automatic micro-partitions + clustering	Manual/auto distribution keys + sort keys
Pruning mechanism	Micro-partition min/max metadata	Sort key block ranges
Join cost driver	Data co-location via clustering	Distribution key alignment across nodes
Concurrency model	Multiple independent virtual warehouses	Shared cluster, workload manager (WLM) queues
Query execution	Vectorized, direct worker-to-worker exchange	Compiled code per query, reused on repeat runs

The part that actually matters for you as an engineer

Neither engine's speed comes from magic — it comes from giving the optimizer good conditions to work with. On Snowflake, that means understanding clustering and not fighting it with poorly organized load patterns. On Redshift, that means deliberately choosing distribution and sort keys that match your actual join and filter patterns, not just letting DISTSTYLE AUTO guess forever.

I've seen this play out directly: a query that ran in seconds on a small staging table took over 4 minutes in production. The SQL hadn't changed. What had changed was that the join column in the production table didn't match the distribution key, so Redshift redistributed rows across every node before it could even start the join — pure network and shuffle overhead, invisible in the query text itself. Aligning the distribution key to the join column brought it back down to seconds.

The real lesson

SQL performance isn't about clever syntax. It's a systems problem — the optimizer, the storage layout, the distribution of data across compute, and the statistics the engine has about your tables all matter more than how you phrase your SELECT. Writing fast SQL means understanding what your specific engine needs to make a good decision, and then giving it that, deliberately, instead of hoping the optimizer figures it out on its own.

Once you start thinking in terms of "what will the optimizer actually do with this," you stop debugging queries by rewriting SQL syntax and start debugging them by checking EXPLAIN, clustering keys, and distribution keys — which is a completely different (and much more effective) way to work.

What's the deepest "it wasn't the SQL, it was the architecture" bug you've run into on Snowflake or Redshift?

How I Stopped Recalculating the Same MAX(date) Subquery on Redshift

Maithreyan — Wed, 05 Aug 2026 12:00:00 +0000

A subquery that looked perfectly fine on its own was quietly running dozens of times a day across our Redshift pipeline. Here's the story of how I found it, why it hurt specifically on Redshift, and the fix that made it a non-issue.

The setup

I had a query that needed the most recent record for a given entity — a fairly common pattern:

SELECT *
FROM events
WHERE event_date = (
  SELECT MAX(event_date)
  FROM events e2
  WHERE e2.entity_id = events.entity_id
);

This is a correlated subquery — the inner SELECT references the outer query's row (e2.entity_id = events.entity_id), so it can't just run once. It worked, it was correct, and it was fast enough on its own. I moved on.

Where it went wrong

The problem wasn't this one query — it was that the same "latest record per entity" logic kept showing up downstream: a dashboard filter, a data quality check, a reconciliation job, a couple of reports. Each of those had its own copy of a similar MAX(date) subquery.

Individually, every query looked fine in isolation. Nobody flagged it in review because each query's runtime on its own was acceptable. The real cost only showed up in aggregate — the same underlying calculation was being recomputed from scratch every time something needed it, hitting the same table repeatedly.

Why this hurts more on Redshift specifically

This is where Redshift's architecture makes the problem worse than it might be elsewhere. Redshift is a columnar, MPP (massively parallel processing) engine, and correlated subqueries interact badly with that design for a few specific reasons:

Limited parallelism. Because a correlated subquery is logically re-evaluated per outer row, it constrains how much the query can be parallelized across compute nodes — you lose some of the benefit of Redshift's distributed architecture.
Sub-optimal query plans. Redshift's optimizer has documented issues generating efficient plans for correlated subqueries, particularly with EXISTS / NOT EXISTS patterns — sometimes producing nested loop joins, which are the slowest join type Redshift supports.
Data movement across nodes. If the subquery's join key isn't the table's distribution key, Redshift has to redistribute rows across the cluster to evaluate the correlation, adding network I/O on top of the recomputation cost.

AWS's own query design guidance is explicit about this: use a CASE expression for complex aggregations instead of scanning the same table multiple times, and prefer subqueries only when they return a small result set (under roughly 200 rows) used purely as a filter — not as a repeated per-row calculation.

The fix

Instead of repeating the subquery everywhere it was needed, I computed the max date once and joined it back to the base table as an extra column:

SELECT
  t1.*,
  t2.max_date
FROM events t1
JOIN (
  SELECT entity_id, MAX(event_date) AS max_date
  FROM events
  GROUP BY entity_id
) t2 ON t1.entity_id = t2.entity_id;

Two things mattered for making this actually fast on Redshift, not just "correct":

The join key matches the table's distribution key. When you join on the distribution key, Redshift can complete the join on each node in parallel without shuffling rows across the cluster. If entity_id isn't your distribution key, this join can still trigger a redistribution step — worth checking with EXPLAIN before assuming it's free.
The result gets reused, not recalculated. Now max_date is a column that lives on the row. Any downstream query, dashboard filter, or reconciliation check just reads that column and compares (event_date = max_date) instead of running its own version of the subquery.

An alternative: window functions

A window function version is also worth considering on Redshift, especially if you want to flag which row is the latest, not just know the date:

SELECT *
FROM (
  SELECT
    *,
    MAX(event_date) OVER (PARTITION BY entity_id) AS max_date,
    ROW_NUMBER() OVER (PARTITION BY entity_id ORDER BY event_date DESC) AS rn
  FROM events
) ranked
WHERE rn = 1;

This avoids a self-join entirely and often performs better than either the correlated subquery or the join approach on large tables, since Redshift can compute it as a single sort-and-scan rather than a join. Redshift performance guides specifically recommend window functions over self-joins as a general optimization pattern.

Materializing beyond a single query

If this value needs to be read very frequently across many downstream consumers, you have a few options on Redshift specifically:

A CTE if it's scoped to a single pipeline run — just be aware that Redshift sometimes rewrites WITH clauses into temporary volt_tt tables internally, which can add overhead on complex CTEs. Simpler joins or window functions sometimes outperform an equivalent CTE for this reason.
A materialized view if the underlying data doesn't change constantly — Redshift materialized views precompute the result set, which is ideal for a value like "max date per entity" that many queries read but few queries update. Note that Redshift disables automatic materialized view refresh by default (citing planning-time overhead), so you'd trigger a manual refresh as the last step of your ETL job.
A physical column updated via your ETL job if this needs extremely frequent reads and refresh timing is predictable — this trades some storage and write complexity for guaranteed fast reads.

Takeaway

Correlated subqueries repeated across multiple queries multiply their cost silently — nothing looks wrong until you count the total load, and Redshift's MPP architecture makes that cost worse due to limited parallelism and potential data redistribution
Precomputing shared logic once — via a join on the distribution key, a window function, or a materialized view — turns N recalculations into one
Always check the query plan with EXPLAIN when in doubt. Don't assume a subquery, join, or window function is faster without testing against your actual table's distribution and sort keys

Have you run into logic that looked fine solo but got expensive once it was reused across a Redshift pipeline? Curious how others have handled this.

Why CTE vs Subquery Matters More on Redshift Than Anywhere Else

Maithreyan — Mon, 03 Aug 2026 12:00:00 +0000

On Snowflake or modern Postgres, CTE vs subquery barely matters for performance. On Redshift, it can matter a lot — and I learned this the hard way.

The assumption I carried over

I carried an assumption over from Snowflake: that the optimizer would treat a CTE and an equivalent subquery the same way. Redshift doesn't play by those rules. Redshift's query planner is derived from an older Postgres lineage, and it historically doesn't inline CTEs the way newer engines do — it can materialize them as a separate step before the outer query even runs.

Why this matters on an MPP engine

This matters a lot on Redshift specifically because it's a columnar, MPP (massively parallel processing) engine. Materializing a CTE means writing an intermediate result set across the cluster's compute nodes before the next step even starts. On a small CTE, that's harmless. On a CTE scanning millions of rows from a big fact table, that's a real cost — extra I/O and redistribution across nodes that a well-written subquery or a join might avoid entirely.

What actually happened

I hit this on a Redshift ETL query pulling from a large events table. I had it as a CTE, referenced once in the outer query, nothing complex. When I flattened it into a subquery instead, the query planner pushed filters down earlier and pruned way more data before the expensive join — noticeably faster on that specific table.

My rule of thumb for Redshift now

Use a subquery when it's referenced once and simple — let the planner push predicates down early
Use a CTE mainly for readability on complex multi-step logic, but check EXPLAIN before trusting it won't materialize unnecessarily
Always check the query plan (EXPLAIN) rather than assuming CTE behavior — Redshift doesn't guarantee the same optimizations as Snowflake or Postgres 12+
For genuinely reused logic, a temp table often beats both if the CTE is being scanned multiple times

The takeaway

Patterns that are "safe defaults" on one warehouse can be real performance traps on another. Always validate with EXPLAIN before assuming.

Anyone else been burned by carrying optimizer assumptions across warehouses?