Joins and runtime pushdown

The shape of a join

{
  "type": "join",
  "with": { "format": "delta", "path": "sales.customers" },
  "with_transformations": [
    { "type": "filter", "condition": "active = true" },
    { "type": "select", "columns": ["customer_id", "segment"] },
    { "type": "distinct" }
  ],
  "on": "customer_id",
  "how": "left"
}

The right side is read inline and reshaped by with_transformations before the join. Three habits pay off immediately:

• select only what you need. A join carries every column of both sides into the shuffle.
• distinct when the right side may repeat the key, or a left join silently multiplies rows.
• Filter early, on the right side, not after the join.

on forms

"on": "customer_id"                                  // same name on both sides
"on": ["customer_id", "order_date"]                  // composite key
"on": "l.customer_id = r.id AND l.dt = r.date"       // different names, or an inequality

The left DataFrame is aliased l and the right one r. Those aliases exist only in the on expression and after the join — inside with_transformations the right side has no alias yet, so use bare column names there.

how

Value	Keeps
inner (default)	rows matching on both sides
left	every left row; right columns null when unmatched
right	every right row
full	everything from both
leftsemi	left rows that have a match — no right columns
leftanti	left rows with no match — the “what is missing?” query
cross	cartesian product

leftsemi is the one to reach for when you only need to filter by existence: it never duplicates rows and never adds columns.

Runtime pushdown

The problem: your working set is 40 000 orders, and the table you need to join against has 900 million rows. A plain join reads the big table.

The fix is to compute the key list first and push it into the read as a literal:

[
  { "type": "filter", "condition": "status = 'PENDING'" },
  { "type": "checkpoint" },
  { "type": "collect", "column": "customer_id", "as": "pending_customers" },
  {
    "type": "join",
    "with": { "format": "delta", "path": "sales.bronze_events" },
    "with_transformations": [
      { "type": "filter", "condition": "customer_id IN ({{pending_customers}})" },
      { "type": "select", "columns": ["customer_id", "last_event_at", "channel"] }
    ],
    "on": "customer_id",
    "how": "left"
  }
]

{{pending_customers}} is substituted inside the filter that runs on the right side, so Delta and Parquet can use their statistics to skip files instead of scanning the table. This is the declarative form of the collect() + isin() pattern hand-written Spark jobs use.

Rules that make it work

Caution

checkpoint before collect. collect is a driver-side action; on an unmaterialized plan Spark recomputes the entire lineage to produce the list, and then again for the join.

• Formatting is automatic: strings become 'a', 'b', numbers 1, 2.
• An empty collection renders as NULL, so IN (NULL) matches nothing — the correct answer when there is nothing to enrich.
• The variable store is shared with nested with_transformations, which is why the join’s right side can see it.
• It is cleared at the start of every run, so nothing leaks between pipelines.

When not to use it

The key list travels to the driver and into a SQL string. Tens of thousands of keys are fine; millions are not — at that size, a plain join with a broadcast hint or a bucketed table is the right tool. If the working set is unbounded, skip the pushdown.

Union is not a join

{ "type": "union", "with": { "format": "parquet", "path": "/data/orders_archive" }, "allow_missing_columns": true }

Danger

With the default allow_missing_columns: false, Spark matches columns by position. Two feeds whose schemas drifted will happily concatenate into garbage. Set it to true to match by name and fill absent columns with null.

union has no with_transformations — the right side is read as-is. Shape it upstream, or use a join if you need to.

Checking your work

Drop a debug node after the join while developing:

{ "type": "debug", "label": "after enrichment", "actions": ["count", "print_schema"] }

A row count that grew means the right side was not unique on the join key. That is the single most common join bug, and a unique validation on the key catches it permanently.