Ma77Ball opened a new issue, #8566:
URL: https://github.com/apache/texera/issues/8566
### Feature Summary
> **In one sentence:** add the columnar join, let the result writer and
Python operators consume Arrow directly, so a batch avoids being decoded back
to rows on the join, the sink, and the Python bridge.
Parent: #8556 (opt-in columnar execution). PR 3 of the stacked series; pairs
with apache/texera#8560.
**What is this?**
By now filter, projection and aggregate are columnar. This issue tackles the
three remaining places a batch would otherwise be forced back into rows: the
join, the write-to-storage step, and the handoff to Python user code.
---
### Proposed Solution or Design
**Selective-probe join.** A hash join has a "build" side (a lookup table)
and a "probe" side (the stream that checks the table). The columnar probe reads
keys straight from the Arrow batch and only decodes the rows that actually
match, so a batch with few matches does almost no per-row work.
```mermaid
%%{init: {'theme':'dark', 'themeVariables':
{'background':'#000000','lineColor':'#0F766E'}}}%%
flowchart LR
BATCH[Arrow probe batch] --> K[read key column]
K --> M{key in build table?}
M -->|miss| SKIP[skip, no decode]
M -->|hit| DEC[decode only this row and join]
classDef default fill:#000,color:#fff,stroke:#888,stroke-width:1px
style SKIP stroke:#1B7F3B
style DEC stroke:#0F766E
```
**Sink offload.** Instead of decoding a batch to rows on the hot loop,
`OutputManager.saveArrowBatchToStorageIfNeeded` hands the whole Arrow batch to
the result-writer thread (`OutputPortStorageWriterThread`), moving the work off
the per-worker loop. Terminal operators return `Consumed` for a columnar batch
rather than forcing a decode.
**Python passthrough.** A `ColumnarFrame` that reaches the Python bridge
(`PythonProxyClient`) is streamed straight to Arrow Flight, with no tuple
round-trip, because Python already speaks Arrow.
```mermaid
%%{init: {'theme':'dark', 'themeVariables':
{'background':'#000000','lineColor':'#000000'}}}%%
flowchart LR
CF[ColumnarFrame] --> PY[stream Arrow to Flight]
PY --> UDF[Python UDF reads Arrow directly]
classDef default fill:#000,color:#fff,stroke:#888,stroke-width:1px
```
**Contract addition.** `ColumnarResult` gains `EmitRows`, and
`processColumnarBatch` takes a `port`, so an operator can consume columns but
emit rows.
| Place | Before | After |
| --- | --- | --- |
| join probe | decode every probe row | decode only matches |
| sink | decode on the DP thread | hand Arrow to the writer thread |
| Python | decode to rows, re-encode | pass Arrow straight through |
Verified: 2-worker shuffle row == columnar through the join, and the Arrow
write path stores the same rows as the row path.
---
*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]
