feat(amber): opt-in columnar (Apache Arrow) execution path

[Ma77Ball]

Feature Summary

In one sentence: teach Amber to move and process data one whole column at a time (Apache Arrow) instead of one row at a time, as an opt-in fast path that runs ~4-6x faster on the operators covered so far and quietly falls back to today's row engine wherever it can't help.

What is this, in plain terms?

Amber today is a row engine. Picture a spreadsheet handled one line at a time: for every row it wraps each value in a Java object, chases a pointer per field, and when a row travels between workers (the small processes that do the actual work) it gets packed and unpacked on its own. Simple, but wasteful once you have millions of rows.

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

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

Loading

It is off by default and behind a flag, so existing workflows behave exactly as before until someone turns it on.

---

Proposed Solution or Design

The whole feature is four independent, opt-in pieces. Each can ship and be reviewed on its own.

%%{init: {'theme':'dark', 'themeVariables': {'background':'#000000','lineColor':'#0F766E'}}}%%
flowchart TD
  A[1. Wire format: ship a batch as one Arrow buffer] --> B[2. Operator contract: an operator may consume columns directly]
  B --> C[3. Operators: scan, filter, projection, aggregate, join, limit, union, distinct, sort]
  C --> D[4. Optional vectorized sink: write Arrow straight to storage]
  classDef default fill:#000,color:#fff,stroke:#888,stroke-width:1px
  class A,B,C,D default

Loading

The safety net (why this can't break results). Every operator either says "I understand columns" or "I don't." If it doesn't, or hits a batch it can't handle, the engine decodes the batch back to rows and runs the normal path. There is no correctness cliff, only a speed opportunity.

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

Loading

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 the gain is dominated by the columnar scan feeding each operator. Sort is at parity by design (it must buffer everything before it can emit). Correctness is checked per operator as row output == columnar output, including a 2-worker shuffle.

Rollout. A stacked PR series, tracked by the sub-issues below. Nothing changes for users until a flag is set.

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. This issue tracks the stacked columnar-execution PR series; see the sub-issues in the comments.

[Ma77Ball]

/sub-issue #8564 #8565 #8566 #8567

[Ma77Ball]

@Yicong-Huang / @chenlica PTAL when available.

[chenlica]

This is a very cool idea. It's great to the performance improvement!

Questions:

1. How does the operator process those values in each column? If the operator needs to access all the fields, does it have additional overhead to concat those values from different columns?

2. Do we create columns for each batch? If so, can we tune the batch size to have even better performance?

If we can increase the batch size, then it will also affect the granularity of pipelined execution. There seems to be a tradeoff here.

[Ma77Ball]

/sub-issue #8684 #8673