feat(amber): columnar shuffle, aggregate, projection, config
@Ma77Ball is already working on this.
Since Sep 17, 2026.
Assessment
This issue has not been assessed yet.
Description
Feature Summary
In one sentence: make the columnar path survive multi-worker shuffles and give the common relational operators (aggregate, projection, limit, union, distinct, sort) a native Arrow path, plus move the on/off switches into config.
Parent: #8556 (opt-in columnar execution). PR 2 of the stacked series; pairs with apache/texera#8559.
What is this?
PR 1 proved a straight line (scan to filter to terminal). Real workflows fan data out across many workers by key (a "shuffle"). This issue teaches the columnar path to split one Arrow batch across receivers correctly, and adds two of the most common operators so more of a real pipeline stays columnar.
Analogy: PR 1 built one conveyor belt. This adds the sorting station that sends each item to the right bin, still by the crateful, not one item at a time.
Proposed Solution or Design
Partition-aware emit (the shuffle). When an edge sends data by key, OutputManager.emitColumnarBatch slices the Arrow batch per receiver using that edge's partitioner: one-to-one and broadcast ship the whole batch, while hash / range / round-robin cut it into per-receiver slices. So a shuffle edge stays columnar instead of silently dropping back to rows. A one-time log per worker makes any fallback visible.
%%{init: {'theme':'dark', 'themeVariables': {'background':'#000000','lineColor':'#0F766E'}}}%%
flowchart TD
B[Arrow batch] --> P{partitioner}
P -->|hash/range/round-robin| S1[slice for worker 1]
P -->|hash/range/round-robin| S2[slice for worker 2]
P -->|one-to-one/broadcast| W[whole batch]
classDef default fill:#000,color:#fff,stroke:#888,stroke-width:1px
The columnar operators.
- Aggregate reads Arrow directly, decoding only the group-key and aggregated columns, then feeds the existing COUNT / SUM / MIN / MAX / AVERAGE accumulation.
- Projection selects, reorders, renames, or drops columns on the Arrow batch with no per-row decode.
- Limit slices the Arrow batch to the remaining row cap and drops later batches once the cap is hit.
- Union passes the Arrow batch straight through (no per-row work).
- Distinct and Sort decode the batch into their existing dedup set / sort buckets and emit at finish. These are the honest cases: they need every column, so they gain little over the row path (see the parent issue's benchmark), but staying on the Arrow path avoids a redundant re-encode.
%%{init: {'theme':'dark', 'themeVariables': {'background':'#000000','lineColor':'#000000'}}}%%
flowchart TD
B[Arrow batch] --> K{operator kind}
K -->|filter/projection/limit/union| PASS[stay in Arrow, emit a batch]
K -->|aggregate/distinct/sort| STATE[decode needed columns into state, emit at finish]
classDef default fill:#000,color:#fff,stroke:#888,stroke-width:1px
style PASS stroke:#1B7F3B
style STATE stroke:#0F766E
Config. The flags move into application.conf (columnar.enable-columnar-wire, enable-vectorized-operators) via ApplicationConfig; the COLUMNAR_WIRE / FILTER_VECTORIZED env vars still override.
| Concern | Before | After |
|---|---|---|
| shuffle edge | falls back to rows | stays columnar (sliced per receiver) |
| aggregate input | full per-row decode | only the needed columns |
| projection | rebuild each row | pick columns on the batch |
| flags | env var only | application.conf + env override |
Verified: on a 2-worker hash shuffle, row output == columnar output for the aggregate across all five functions and for a projection feeding a downstream filter and aggregate.
High-level overview. Part of #8556.
- Dominant language
- Scala
- Stars
- 316
- Forks
- 189
- Avg merge
- 2d 17h
- Merged PRs (30d)
- 196
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
More from apache/texera
-
Difficulty 1/5 Under an hour Newbie friendliness 90/100
-
Difficulty 2/5 1-3 hours Newbie friendliness 82/100
-
Difficulty 2/5 1-3 hours Newbie friendliness 72/100
-
Difficulty 2/5 1-3 hours Newbie friendliness 86/100
-
Difficulty 1/5 Under an hour Newbie friendliness 90/100
Similar issues
-
requires-triage
Difficulty 2/5 1-3 hours Newbie friendliness 84/100
apache/datafusion-comet#6060 ·
-
Difficulty 2/5 1-3 hours Newbie friendliness 68/100
lichess-org/lila#21756 ·
-
Difficulty 1/5 Under an hour Newbie friendliness 90/100
-
enhancement
Difficulty 2/5 1-3 hours Newbie friendliness 82/100
salesforce/evalon#10 ·
-
Difficulty 2/5 1-3 hours Newbie friendliness 78/100
scalacenter/scalafix#2522 ·