Ma77Ball opened a new issue, #8556: URL: https://github.com/apache/texera/issues/8556
### Feature Summary Amber executes row by row: a batch is an array of `Tuple`, and each `Tuple` is a boxed `Array[Any]`, so every field is a Java object and data is serialized one tuple at a time between workers. That boxes every value, chases a pointer per field, and re-pays serialization per row. This proposes an opt-in columnar path on Apache Arrow, where a batch is one Arrow structure that stores each column as a contiguous native array. Operators that opt in process a whole batch of columns at once, and workers ship the batch as a single Arrow buffer instead of N tuples. It is off by default and flag-gated, so the row path is unchanged when disabled. On the operators covered so far it runs ~4-6x faster on a single worker, with output verified identical to the row path. ### Proposed Solution or Design Four independent, opt-in pieces. (1) Wire format: a `ColumnarFrame` payload carries one Arrow IPC batch plus row count and format version; a worker decodes once per batch instead of once per row, and falls back to the per-tuple `DataFrame` when the flag is off. (2) Operator contract: an operator may implement `processColumnarBatch(bytes, port)` returning `Emit(arrowBatch)` (produced columns, e.g. filter/projection), `EmitRows(rows)` (consumed columns, emits rows), `Consumed` (absorbed into state, e.g. sort/distinct), or `Unsupported` (engine decodes and runs the row path). `Unsupported` is the safety net, so there is no correctness cliff. (3) Operators covered: scan, filter, projection, aggregate, join (selective-probe), limit, union, distinct, sort; Python UDFs pass the Arrow batch straight through to the existing Flight bridge with no per-row round-trip. (4) Optional vectorized sink: the Iceberg writer can consume the Arrow batch directly through a record view that reads values from the Arrow columns, materializing no intermediate rows. Config lives under `columnar.*` in `application.conf` (`enable-columnar-wire`, `enable-vectorized-operators`, `enable-vectorized-sink`, `wire-compression`), each with an env override, default off. Measured on 2M-row lineitem, single worker, scan to op to terminal: filter 34.4s to 5.7s (~6.0x), projection ~6.0x, limit ~6.3x, aggregate ~4.5x, union ~4.3x, distinct ~3.9x, join high-miss 73.7s to 13.1s (~5.6x), sort at parity (it must buffer before emitting). These are whole-pipeline numbers dominated by the columnar scan; correctness was checked per operator as row output == columnar output, including a 2-worker hash shuffle. Rollout is a stacked PR series (wire+scan+filter, then shuffle+aggregate+projection+config, then join+sink+Python passthrough, then hardening) so each piece reviews on its own. Open questions: keep flag-gated opt-in as the default or flip once coverage is broad; which operators get a native path next vs. staying on fallback; and larger multi-node validation beyond the current single-worker and 2-worker shuffle. -- 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]
