perf(cubeorchestrator): Arrow-backed CubeStore results + columnar driver batches

[ovr]

Stacked on #11235. Two follow-ups that extend the columnar pipeline on both ends:

1. Keep CubeStore Arrow results columnar under a wrapper (cubeorchestrator)

CubeStore results arrive as Arrow IPC, but were materialized into per-cell DBResponsePrimitives at parse time (a String allocation per string cell), and the columnar transform then copied every cell again via transform_value(cell.clone()) — even though the transform is identity for everything except "time"-typed string cells.

ColumnarArray is now an enum: Cells (materialized; JS raw input, legacy result sets, synthetic columns) or Arrow (the Arrow arrays as they arrived, one chunk per record batch). Values are read straight from the Arrow buffers through a borrowed DBResponseValueRef view:

• parse allocates nothing per cell — column data types are still validated eagerly, so unsupported types fail at parse time as before;
• the columnar transform shares identity columns as an Arc bump per chunk instead of a per-cell copy (only "time"-typed string columns still materialize);
• JSON serialization and the SQL-API FieldValue path write string cells directly from the Arrow buffers, with byte-identical output to the materialized path;
• row-major consumers (compact/vanilla plans, getCubestoreResult) materialize once at plan build time via to_cells(), matching the old parse-time cost.

2. Stream columnar batches from the driver (base-driver / postgres-driver)

The SQL-API streaming path still received row objects from drivers and transposed them to columnar in cubejs-backend-native. New ColumnarResponse Transform in base-driver pivots position-indexed array rows into { members, columns } batches, with a columnar hint on StreamOptions:

• PostgresDriver reads rows with rowMode: 'array' (no per-row object allocation) and pipes them through ColumnarResponse;
• QueryStream remaps member aliases once per batch instead of per row;
• the native stream writer forwards batches straight to Rust, skipping row accumulation and the row→columnar transpose;
• drivers without columnar support keep emitting row objects through the old path.

Tests

• cargo test -p cubeorchestrator — 85 passed
• yarn jest test/unit/ColumnarResponse.test.ts (base-driver) — 4 passed

🤖 Generated with Claude Code

[claude]

Claude finished @ovr's task in 2m 41s —— View job

---

Review: Arrow-backed CubeStore results + columnar driver batches

Overall: solid, well-structured change with strong internal docs and a matching test on the JS side. The ArrowColumn / DBResponseValueRef split cleanly avoids the per-cell String allocation, and byte-identical JSON output (numeric-as-string, ISO timestamps) is preserved via the hand-written Serialize impl. The columnar streaming path bypasses two known hot spots (per-row object allocation in pg + row→columnar transpose in Rust).

Verdict: 0 high · 3 medium · 1 low — nothing blocking; all findings are inline.

Details

• Load PR context
• Read changed files (Rust)
• Read changed files (JS/TS)
• Post inline findings + summary

Findings (posted inline):

#	File	Severity	Summary
1	packages/cubejs-backend-native/js/index.ts:296	low	Sentinel 'ColumnarResponse' is hardcoded; import COLUMNAR_RESPONSE_TYPE from @cubejs-backend/base-driver to keep the marker in lockstep
2	rust/cube/cubeorchestrator/src/arrow_column.rs:295-301	medium	(*scale).max(0) as u32 silently drops Arrow negative decimal scales; either reject at TypedChunk::try_new or handle correctly
3	packages/cubejs-query-orchestrator/src/orchestrator/QueryStream.ts:50-61	medium	Unknown aliases become undefined entries in the batch members; also the counter/backpressure arithmetic diverges from the row path when a batch's rowCount >= HWM
4	packages/cubejs-postgres-driver/src/PostgresDriver.ts:383-389	medium	Error propagation is one-way (rowStream → columnarStream); a downstream error won't tear the pg cursor down promptly. Also worth a sanity check that batchSize = highWaterMark is intentional

Positive notes:

• The ArrowColumn::try_new up-front validation lets iter_values skip the fallible path per cell — nice.
• has_string_values() short-circuits so the columnar transform is truly a zero-copy Arc bump for non-time / non-string columns.
• Decimal256 rendering uses the generic Display helper, so no Arrow-string fallback for >i128 mantissas (see test_from_arrow_decimal256_beyond_i128).
• Byte-identical serialize behavior between Cells/Arrow variants is asserted via the shared TIMESTAMP_ITEMS and matching numeric-as-string rules on both DBResponsePrimitive and DBResponseValueRef.
• ColumnarResponse test covers batch boundaries, empty streams, and per-batch column-array freshness — good.

Not covered by tests (worth considering):

• A unit test for QueryStream's columnar branch (alias remap + passthrough).
• An integration-shaped test that a driver ignoring columnar: true still works (fallback contract).
• Rust: a ColumnarArray::Arrow → TransformedData::Vanilla path test to cover to_cells() materialization on the row-major consumer side.

· Branch: `perf/sqlapi-streaming-columnar-v2`

[claude]

Nit: this uses the string literal 'ColumnarResponse' while packages/cubejs-base-driver/src/ColumnarResponse.ts exports COLUMNAR_RESPONSE_TYPE as the canonical marker. Importing the constant here (from @cubejs-backend/base-driver) would keep the two ends in lockstep — if the sentinel ever changes, the native writer won't silently fall through to the row-accumulation path.

[claude]

(*scale).max(0) as u32 silently drops Arrow's negative decimal scales. Arrow allows negative scale to mean "multiply mantissa by 10^|scale|" (e.g. Decimal128(p, -2) with mantissa 123 represents 12300). With this clamp, that value would render as "123" instead of "12300".

If CubeStore never emits negative-scale decimals in practice this is moot, but if the intent is "unsupported", it'd be safer to reject via ParseError::UnsupportedArrowType at TypedChunk::try_new time rather than silently truncate at read time.

[claude]

Two small concerns in the columnar branch:

1. Alias remap silently drops unknown aliases. chunk.members.map((alias) => this.aliasNameToMember[alias]) will produce undefined entries when an alias is missing from the map. The row-path has the same latent issue (row[undefined] = …), but here the undefined ends up in the batch's members array that Rust decodes as JsRawColumnarData. Worth double-checking that upstream guarantees a full map, or falling back to the raw alias when a lookup misses.

2. The counter/backpressure arithmetic differs from the row path. Row path increments by 1 and compares counter < HWM. Here it's counter + rowCount < HWM — if a single batch's rowCount >= HWM, the counter is reset to 0 and the debounce timer fires on every subsequent batch. Not a correctness bug (only affects the idle-destroy timer), but the divergence is worth a comment or normalization.

[claude]

Two things worth checking here:

• batchSize: highWaterMark: highWaterMark is the pg row-buffer size (default 8192 per the native writer). Using it as the columnar batch size means each object-mode push moves ~8k rows worth of column data — fine, but if the intent was "one batch per pipe write" this is what you want; if it was "flush more often to smooth latency" you may want a distinct batchSize.
• Error propagation only wires rowStream -> columnarStream.destroy(e). If the downstream consumer errors first, columnarStream is destroyed but rowStream keeps producing until pipe()'s internal cleanup fires. Consider columnarStream.once('error', (e) => rowStream.destroy(e)) too, so a downstream failure tears down the pg cursor promptly and release() isn't blocked waiting for the source to drain.