Ma77Ball opened a new issue, #8565:
URL: https://github.com/apache/texera/issues/8565
### Feature Summary
> **In one sentence:** make the columnar path survive multi-worker shuffles
and add two more columnar operators (aggregate and projection), 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.
```mermaid
%%{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
```
**Two more 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.
**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.*
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
