Columnar (Apache Arrow) execution in Amber: the opt-in PR series and why

[Ma77Ball]

In one sentence: a stacked, opt-in PR series (#8558-#8561) that teaches Amber to move and process data one whole column at a time (Apache Arrow) instead of one row at a time, about 4-6x faster where it helps, and quietly falling back to today's row engine everywhere else.

Tracking issue: #8556. Sub-issues: #8564, #8565, #8566, #8567.

---

Why we are doing this

Amber is a row engine today. For every row it wraps each value in a Java object, follows a pointer per field, and packs then unpacks it on its own when it crosses between workers (the small processes that do the actual work). That is simple, but at millions of rows most of the time goes into boxing and serialization, not the real work.

Columnar execution turns a batch sideways. Instead of a stack of rows, a batch becomes a few columns, each stored as one tight native array. That is what Apache Arrow (a standard in-memory column format) gives us. An operator can then act on a whole column at once, and a batch crosses the network as a few big buffers instead of thousands of little envelopes.

%%{init: {'theme':'dark', 'themeVariables': {'background':'#000000','lineColor':'#0F766E'}}}%%
flowchart LR
  subgraph ROW["Row engine (today)"]
    r1[row 1] --> r2[row 2] --> r3[row 3]
  end
  subgraph COL["Columnar (Arrow)"]
    c1[column A]
    c2[column B]
    c3[column C]
  end
  ROW --> COL
  classDef default fill:#000,color:#fff,stroke:#888,stroke-width:1px

Loading

---

Why it is safe

Every operator either understands columns or it does not. If it does not, or hits a batch it cannot handle, the engine decodes the batch back to rows and runs the normal path. It is off by default, behind a flag. There is no correctness cliff, only a speed opportunity.

%%{init: {'theme':'dark', 'themeVariables': {'background':'#000000','lineColor':'#000000'}}}%%
flowchart TD
  IN[Arrow batch] --> Q{operator has a<br/>columnar path?}
  Q -->|yes| FAST[process columns directly]
  Q -->|no| SLOW[decode to rows, run row path]
  classDef default fill:#000,color:#fff,stroke:#888,stroke-width:1px
  style FAST stroke:#1B7F3B
  style SLOW stroke:#B0451E

Loading

---

The PR series (review and merge top to bottom)

PR	Issue	What it adds
#8558	#8564	the foundation: Arrow wire format, the opt-in operator contract, columnar scan and filter
#8559	#8565	multi-worker shuffle, the operators (aggregate, projection, limit, union, distinct, sort), config flags
#8560	#8566	join (selective probe), vectorized sink, Python passthrough
#8561	#8567	hardening: wire versioning, optional LZ4/ZSTD compression

Each PR builds on the one before it; #8561 closes the tracking issue.

---

How much faster (measured)

2M-row lineitem, single worker, scan to operator to terminal:

Operator	Row	Columnar	Speedup
filter	34.4s	5.7s	~6.0x
projection	34.4s	5.8s	~6.0x
limit	34.2s	5.4s	~6.3x
aggregate	34.3s	7.7s	~4.5x
union	37.1s	8.5s	~4.3x
distinct	34.5s	8.9s	~3.9x
join (high-miss)	73.7s	13.1s	~5.6x
sort	75.9s	76.9s	~1.0x

These are whole-pipeline numbers, so most of the win is the columnar scan building zero row tuples. Sort is at parity by design: it is a blocking operator that must buffer everything and needs every column, so the scan saving is cancelled out. Correctness is checked per operator as row output == columnar output, including a 2-worker shuffle.

---

Open questions for discussion

• Keep flag-gated opt-in as the default, or flip it on once coverage is broad enough?
• Which operators earn a native columnar path next, versus staying on the row fallback?
• Distributed validation beyond the current single-worker and 2-worker shuffle.

---

High-level overview of the columnar-execution work. Details and diagrams per piece live in the linked issues.