feat(amber): columnar join, vectorized sink, Python passthrough

[Ma77Ball]

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 #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.

%%{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

Loading

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.

%%{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

Loading

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.