Skip to content

Commit 1a64945

Browse files
ovrovr
authored
perf(cubeorchestrator): Use columnar format between JS->Rust data transport (-20%) (cube-js#10837)
Switch the JS->Rust raw query results wire format from row-oriented `Vec<IndexMap<String, DBResponsePrimitive>>` (each row carrying its own copy of every column name) to columnar `JsRawColumnarData { members, columns }`. The pivot is done once on the JS side in `ResultWrapper.serialize` before `JSON.stringify`, then `from_js_raw_data` transposes back to the existing row-major `QueryResult` on Rust without per-cell clones. Also drops the `JsArray` + `JsValueDeserializer` ingest path in `orchestrator.rs` — only `JsBox<Arc<QueryResult>>` and `JsBuffer` are accepted now. The `JsValueDeserializer` walk was the slow path the buffer route was already winning against. Bench: synthetic dataset, 60% string dims / 40% f64 measures, no nulls, `cargo bench --bench transform -- --sample-size 30 --measurement-time 3`. Rust ingest path (`serde_json::from_slice` + `from_js_raw_data`, median): | Dataset | rows (master) | columnar (this) | Faster by | Speedup | | ------------------- | ------------: | --------------: | --------: | ------: | | 8 cols x 10k rows | 52.4 ms | 40.8 ms | -22 % | 1.29 x | | 16 cols x 10k rows | 107.2 ms | 82.9 ms | -23 % | 1.29 x | | 16 cols x 100k rows | 1024.7 ms | 827.7 ms | -19 % | 1.24 x | | 32 cols x 100k rows | 2019.8 ms | 1633.9 ms | -19 % | 1.24 x |
1 parent 10f249e commit 1a64945

7 files changed

Lines changed: 335 additions & 227 deletions

File tree

‎packages/cubejs-api-gateway/src/gateway.ts‎

Lines changed: 1 addition & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ import {
1919
ResultArrayWrapper,
2020
ResultMultiWrapper,
2121
ResultWrapper,
22+
rowsToColumnar,
2223
} from '@cubejs-backend/native';
2324
import type {
2425
Application as ExpressApplication,
@@ -129,40 +130,6 @@ function systemAsyncHandler(handler: (req: Request & { context: ExtendedRequestC
129130
};
130131
}
131132

132-
function rowsToColumnar(rawData: any): { members: string[]; columns: any[][] } {
133-
let rows: any[];
134-
135-
if (Array.isArray(rawData)) {
136-
rows = rawData;
137-
} else if (rawData) {
138-
rows = Array.from(rawData as Iterable<any>);
139-
} else {
140-
rows = [];
141-
}
142-
143-
const rowCount = rows.length;
144-
if (rowCount === 0) {
145-
return { members: [], columns: [] };
146-
}
147-
148-
const members = Object.keys(rows[0]);
149-
const memberCount = members.length;
150-
const columns: any[][] = new Array(memberCount);
151-
152-
for (let j = 0; j < memberCount; j++) {
153-
const member = members[j];
154-
const col = new Array(rowCount);
155-
156-
for (let i = 0; i < rowCount; i++) {
157-
col[i] = rows[i][member];
158-
}
159-
160-
columns[j] = col;
161-
}
162-
163-
return { members, columns };
164-
}
165-
166133
// Prepared CheckAuthFn, default or from config: always async
167134
type PreparedCheckAuthFn = (ctx: any, authorization?: string) => Promise<{
168135
securityContext: any;

‎packages/cubejs-backend-native/js/ResultWrapper.ts‎

Lines changed: 48 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,45 @@ export interface DataResult {
1515
getResults(): ResultWrapper[];
1616
}
1717

18+
export interface JsRawColumnarData {
19+
members: string[];
20+
columns: any[][];
21+
}
22+
23+
export function rowsToColumnar(rawData: any): JsRawColumnarData {
24+
let rows: any[];
25+
26+
if (Array.isArray(rawData)) {
27+
rows = rawData;
28+
} else if (rawData) {
29+
rows = Array.from(rawData as Iterable<any>);
30+
} else {
31+
rows = [];
32+
}
33+
34+
const rowCount = rows.length;
35+
if (rowCount === 0) {
36+
return { members: [], columns: [] };
37+
}
38+
39+
const members = Object.keys(rows[0]);
40+
const memberCount = members.length;
41+
const columns: any[][] = new Array(memberCount);
42+
43+
for (let j = 0; j < memberCount; j++) {
44+
const member = members[j];
45+
const col = new Array(rowCount);
46+
47+
for (let i = 0; i < rowCount; i++) {
48+
col[i] = rows[i][member];
49+
}
50+
51+
columns[j] = col;
52+
}
53+
54+
return { members, columns };
55+
}
56+
1857
class BaseWrapper {
1958
public readonly isWrapper: boolean = true;
2059
}
@@ -140,14 +179,19 @@ export class ResultWrapper extends BaseWrapper implements DataResult {
140179
return [this.nativeReference];
141180
}
142181

182+
// Pivot to columnar before serializing: the row-oriented form repeats
183+
// every column name on every row, which inflates JSON size and forces
184+
// the Rust side to allocate a per-row map before transposing back to
185+
// its native columnar `QueryResult` representation.
186+
//
143187
// Serialize to a Buffer so the Rust side can decode via
144-
// serde_json::from_slice instead of walking a JsArray through the
188+
// serde_json::from_slice instead of walking a JsValue through the
145189
// Neon bridge with JsValueDeserializer. On 5 MB of AoO rows
146-
// (~21k rows × 8 fields) the JsArray walk costs ~80 ms locally;
190+
// (~21k rows × 8 fields) the JsValue walk costs ~80 ms locally;
147191
// Buffer + serde_json is ~7× faster (M3 MAX) and tracks V8's JSON.parse
148192
// (~11 ms on the same payload). On a real server it should be 3-6× slower,
149-
// so avoiding the JsArray walk matters even more there.
150-
return [Buffer.from(JSON.stringify(this.jsResult))];
193+
// so avoiding the JsValue walk matters even more there.
194+
return [Buffer.from(JSON.stringify(rowsToColumnar(this.jsResult)))];
151195
}
152196

153197
public setTransformData(td: any) {

‎packages/cubejs-backend-native/src/orchestrator.rs‎

Lines changed: 24 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ use cubeorchestrator::query_result_transform::{
55
DBResponsePrimitive, DBResponseValue, RequestResultData, RequestResultDataMulti,
66
TransformedData,
77
};
8-
use cubeorchestrator::transport::{JsRawData, TransformDataRequest};
8+
use cubeorchestrator::transport::{JsRawColumnarData, TransformDataRequest};
99
use cubesql::compile::engine::df::scan::{ColumnarValueObject, FieldValue, ValueObject};
1010
use cubesql::CubeError;
1111
use neon::context::{Context, FunctionContext, ModuleContext};
@@ -87,41 +87,27 @@ impl ResultWrapper {
8787

8888
let raw_data_js = raw_data_js_arr.first().unwrap();
8989

90-
let query_result =
91-
if let Ok(js_box) = raw_data_js.downcast::<JsBox<Arc<QueryResult>>, _>(cx) {
92-
Arc::clone(&js_box)
93-
} else if let Ok(js_buffer) = raw_data_js.downcast::<JsBuffer, _>(cx) {
94-
let bytes = js_buffer.as_slice(cx);
95-
let js_raw_data: JsRawData = serde_json::from_slice(bytes).map_err(|e| {
96-
CubeError::internal(format!(
97-
"Can't parse raw data JSON from JS ResultWrapper: {}",
98-
e
99-
))
100-
})?;
101-
102-
QueryResult::from_js_raw_data(js_raw_data)
103-
.map(Arc::new)
104-
.map_cube_err("Can't build results data from JS rawData")?
105-
} else if let Ok(js_array) = raw_data_js.downcast::<JsArray, _>(cx) {
106-
let deserializer = JsValueDeserializer::new(cx, js_array.upcast());
107-
let js_raw_data: JsRawData = match Deserialize::deserialize(deserializer) {
108-
Ok(data) => data,
109-
Err(_) => {
110-
return Err(CubeError::internal(
111-
"Can't deserialize results raw data from JS ResultWrapper object"
112-
.to_string(),
113-
));
114-
}
115-
};
90+
let query_result = if let Ok(js_box) =
91+
raw_data_js.downcast::<JsBox<Arc<QueryResult>>, _>(cx)
92+
{
93+
Arc::clone(&js_box)
94+
} else if let Ok(js_buffer) = raw_data_js.downcast::<JsBuffer, _>(cx) {
95+
let bytes = js_buffer.as_slice(cx);
96+
let js_raw_data: JsRawColumnarData = serde_json::from_slice(bytes).map_err(|e| {
97+
CubeError::internal(format!(
98+
"Can't parse raw data JSON from JS ResultWrapper: {}",
99+
e
100+
))
101+
})?;
116102

117-
QueryResult::from_js_raw_data(js_raw_data)
118-
.map(Arc::new)
119-
.map_cube_err("Can't build results data from JS rawData")?
120-
} else {
121-
return Err(CubeError::internal(
122-
"Can't deserialize results raw data from JS ResultWrapper object".to_string(),
123-
));
124-
};
103+
QueryResult::from_js_raw_data(js_raw_data)
104+
.map(Arc::new)
105+
.map_cube_err("Can't build results data from JS rawData")?
106+
} else {
107+
return Err(CubeError::internal(
108+
"Can't deserialize results raw data from JS ResultWrapper object".to_string(),
109+
));
110+
};
125111

126112
Ok(Self {
127113
transform_data: transform_request,
@@ -315,21 +301,14 @@ fn extract_query_result(
315301
Ok(Arc::clone(&js_box))
316302
} else if let Ok(js_buffer) = data_arg.downcast::<JsBuffer, _>(cx) {
317303
let bytes = js_buffer.as_slice(cx);
318-
let js_raw_data: JsRawData = serde_json::from_slice(bytes)?;
319-
320-
QueryResult::from_js_raw_data(js_raw_data)
321-
.map(Arc::new)
322-
.map_err(anyhow::Error::from)
323-
} else if let Ok(js_array) = data_arg.downcast::<JsArray, _>(cx) {
324-
let deserializer = JsValueDeserializer::new(cx, js_array.upcast());
325-
let js_raw_data: JsRawData = Deserialize::deserialize(deserializer)?;
304+
let js_raw_data: JsRawColumnarData = serde_json::from_slice(bytes)?;
326305

327306
QueryResult::from_js_raw_data(js_raw_data)
328307
.map(Arc::new)
329308
.map_err(anyhow::Error::from)
330309
} else {
331310
Err(anyhow::anyhow!(
332-
"Second argument must be an Array of JsBox<Arc<QueryResult>> or JsArray"
311+
"Second argument must be a JsBox<Arc<QueryResult>> or a JsBuffer with columnar JsRawColumnarData JSON"
333312
))
334313
}
335314
}
@@ -357,7 +336,7 @@ pub fn get_cubestore_result(mut cx: FunctionContext) -> JsResult<JsValue> {
357336
for (i, row) in result.rows.iter().enumerate() {
358337
let js_row = cx.execute_scoped(|mut cx| {
359338
let js_row = JsObject::new(&mut cx);
360-
for (key, value) in result.columns.iter().zip(row.iter()) {
339+
for (key, value) in result.members.iter().zip(row.iter()) {
361340
let js_key = cx.string(key);
362341
let js_value: Handle<'_, JsValue> = match value {
363342
DBResponseValue::Primitive(DBResponsePrimitive::Null) => cx.null().upcast(),

‎rust/cube/cubeorchestrator/benches/transform.rs‎

Lines changed: 95 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,9 @@ use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Through
55
use cubeorchestrator::query_message_parser::QueryResult;
66
use cubeorchestrator::query_result_transform::{DBResponsePrimitive, TransformedData};
77
use cubeorchestrator::transport::{
8-
ConfigItem, JsRawData, MemberOrMemberExpression, NormalizedQuery, QueryType, ResultType,
9-
TransformDataRequest,
8+
ConfigItem, JsRawColumnarData, MemberOrMemberExpression, NormalizedQuery, QueryType,
9+
ResultType, TransformDataRequest,
1010
};
11-
use indexmap::IndexMap;
1211

1312
const ROW_COUNTS: &[usize] = &[1_000, 10_000, 50_000, 100_000];
1413
const COLUMN_COUNTS: &[usize] = &[8, 16, 32, 64];
@@ -171,40 +170,48 @@ fn build_dataset(
171170
dimensions: &[(String, String)],
172171
measures: &[(String, String)],
173172
time_dims: &[TimeColumn],
174-
) -> JsRawData {
173+
) -> JsRawColumnarData {
175174
let total_cols = dimensions.len() + measures.len() + time_dims.len();
176-
let mut rows = Vec::with_capacity(row_count);
177-
178-
for i in 0..row_count {
179-
let mut row = IndexMap::with_capacity(total_cols);
180-
for (j, (_, alias)) in dimensions.iter().enumerate() {
181-
row.insert(
182-
alias.clone(),
183-
DBResponsePrimitive::String(format!("dim_{}_{}", j, i % 1000)),
184-
);
175+
let mut members = Vec::with_capacity(total_cols);
176+
let mut columns: Vec<Vec<DBResponsePrimitive>> = Vec::with_capacity(total_cols);
177+
178+
for (j, (_, alias)) in dimensions.iter().enumerate() {
179+
members.push(alias.clone());
180+
let mut col = Vec::with_capacity(row_count);
181+
for i in 0..row_count {
182+
col.push(DBResponsePrimitive::String(format!(
183+
"dim_{}_{}",
184+
j,
185+
i % 1000
186+
)));
185187
}
186-
for (j, (_, alias)) in measures.iter().enumerate() {
187-
row.insert(
188-
alias.clone(),
189-
DBResponsePrimitive::Number(((i * (j + 1)) as f64) * 0.5),
190-
);
188+
columns.push(col);
189+
}
190+
for (j, (_, alias)) in measures.iter().enumerate() {
191+
members.push(alias.clone());
192+
let mut col = Vec::with_capacity(row_count);
193+
for i in 0..row_count {
194+
col.push(DBResponsePrimitive::Number(((i * (j + 1)) as f64) * 0.5));
191195
}
192-
for (j, td) in time_dims.iter().enumerate() {
196+
columns.push(col);
197+
}
198+
for (j, td) in time_dims.iter().enumerate() {
199+
members.push(td.alias.clone());
200+
let mut col = Vec::with_capacity(row_count);
201+
for i in 0..row_count {
193202
// Format mirrors typical CubeStore output: ISO-8601 with millisecond
194-
// fractional and no timezone. None of `transform_value`'s six chrono
195-
// parsers fully match this shape, so the function falls through to
196-
// `s.clone()` — measuring the production worst case.
203+
// fractional and no timezone.
197204
let month = ((i + j) % 12) + 1;
198205
let day = ((i / 12) % 28) + 1;
199-
row.insert(
200-
td.alias.clone(),
201-
DBResponsePrimitive::String(format!("2024-{:02}-{:02}T00:00:00.000", month, day)),
202-
);
206+
col.push(DBResponsePrimitive::String(format!(
207+
"2024-{:02}-{:02}T00:00:00.000",
208+
month, day
209+
)));
203210
}
204-
rows.push(row);
211+
columns.push(col);
205212
}
206213

207-
rows
214+
JsRawColumnarData { members, columns }
208215
}
209216

210217
fn bench_transform(c: &mut Criterion) {
@@ -305,5 +312,64 @@ fn bench_transform_time_scenarios(c: &mut Criterion) {
305312
group.finish();
306313
}
307314

308-
criterion_group!(benches, bench_transform, bench_transform_time_scenarios);
315+
/// Bench the JS→Rust raw-data ingest path: `serde_json::from_slice` then
316+
/// `QueryResult::from_js_raw_data`. This is the part of the pipeline that the
317+
/// columnar wire format change actually touches; `bench_transform` above
318+
/// consumes an already-built `QueryResult` and is unaffected.
319+
fn bench_from_js_raw_data(c: &mut Criterion) {
320+
let mut group = c.benchmark_group("QueryResult::from_js_raw_data");
321+
322+
let combos: &[(usize, usize)] = &[(8, 10_000), (16, 10_000), (16, 100_000), (32, 100_000)];
323+
324+
for &(col_count, row_count) in combos {
325+
let (dim_count, measure_count) = split_dim_measure(col_count);
326+
let dimensions = make_member_aliases("dim", dim_count);
327+
let measures = make_member_aliases("measure", measure_count);
328+
329+
let dataset = build_dataset(row_count, &dimensions, &measures, &[]);
330+
let payload = serde_json::to_vec(&dataset).expect("to_vec");
331+
let payload_len = payload.len();
332+
333+
eprintln!(
334+
"from_js_raw_data: c{:02}_r{} payload_bytes={}",
335+
col_count, row_count, payload_len
336+
);
337+
338+
group.throughput(Throughput::Elements((row_count * col_count) as u64));
339+
340+
let id_param = format!("c{:02}_r{}", col_count, row_count);
341+
342+
// Parse only: serde_json::from_slice into the wire type.
343+
group.bench_with_input(BenchmarkId::new("parse_only", &id_param), &(), |b, _| {
344+
b.iter(|| {
345+
let parsed: JsRawColumnarData =
346+
serde_json::from_slice(black_box(&payload)).expect("from_slice");
347+
black_box(parsed);
348+
});
349+
});
350+
351+
// End-to-end: parse + transpose into QueryResult — what the Neon bridge does.
352+
group.bench_with_input(
353+
BenchmarkId::new("parse_plus_build", &id_param),
354+
&(),
355+
|b, _| {
356+
b.iter(|| {
357+
let parsed: JsRawColumnarData =
358+
serde_json::from_slice(black_box(&payload)).expect("from_slice");
359+
let built = QueryResult::from_js_raw_data(parsed).expect("from_js_raw_data");
360+
black_box(built);
361+
});
362+
},
363+
);
364+
}
365+
366+
group.finish();
367+
}
368+
369+
criterion_group!(
370+
benches,
371+
bench_transform,
372+
bench_transform_time_scenarios,
373+
bench_from_js_raw_data
374+
);
309375
criterion_main!(benches);

0 commit comments

Comments
 (0)