perf(cubeorchestrator): keep CubeStore Arrow results in Arrow memory

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

@@ -1,5 +1,6 @@
 use crate::node_obj_deserializer::JsValueDeserializer;
 use crate::transport::MapCubeErrExt;
+use cubeorchestrator::direct_result::DirectData;
 use cubeorchestrator::query_message_parser::QueryResult;
 use cubeorchestrator::query_result_transform::{
     DBResponsePrimitive, RequestResultData, RequestResultDataMulti, TransformedData,
@@ -257,8 +258,10 @@ pub fn get_cubestore_result(mut cx: FunctionContext) -> JsResult<JsValue> {
             result.members().iter().map(|k| cx.string(k)).collect();
 
         let row_count = result.row_count();
+        // One reader per column, so an Arrow-backed column resolves its type once
+        // instead of on every cell.
         let columns: Vec<_> = (0..js_keys.len())
-            .map(|i| result.column(i))
+            .map(|i| result.reader(i))
             .collect::<Result<_, _>>()
             .or_else(|err| cx.throw_error(err.to_string()))?;
         let js_array = JsArray::new(&mut cx, row_count);
@@ -268,11 +271,11 @@ pub fn get_cubestore_result(mut cx: FunctionContext) -> JsResult<JsValue> {
                 let js_row = JsObject::new(&mut cx);
 
                 for (col_idx, js_key) in js_keys.iter().enumerate() {
-                    let value = &columns[col_idx][row_idx];
-                    let js_value: Handle<'_, JsValue> = match value {
-                        DBResponsePrimitive::Null => cx.null().upcast(),
-                        // For compatibility, we convert all primitives to strings
-                        other => cx.string(other.to_string()).upcast(),
+                    // For compatibility, we convert all primitives to strings
+                    let js_value: Handle<'_, JsValue> = match columns[col_idx].value_as_str(row_idx)
+                    {
+                        None => cx.null().upcast(),
+                        Some(text) => cx.string(text).upcast(),
                     };
 
                     js_row.set(&mut cx, *js_key, js_value)?;
@@ -307,14 +310,17 @@ pub fn final_query_result(mut cx: FunctionContext) -> JsResult<JsPromise> {
 
     let result_data_js_object = cx.argument::<JsValue>(2)?;
     let deserializer = JsValueDeserializer::new(&mut cx, result_data_js_object);
-    let mut result_data: RequestResultData = match Deserialize::deserialize(deserializer) {
+    let result_data: RequestResultData = match Deserialize::deserialize(deserializer) {
         Ok(data) => data,
         Err(err) => return cx.throw_error(err.to_string()),
     };
 
     let promise = cx
         .task(move || {
-            result_data.prepare_results(&transform_request_data, &cube_store_result)?;
+            // The result is serialized and dropped, so render `data` straight
+            // from the source columns instead of materializing it first.
+            let result_data =
+                result_data.with_data(DirectData::new(&transform_request_data, &cube_store_result));
 
             match serde_json::to_string(&result_data) {
                 Ok(json) => Ok(json),
@@ -353,7 +359,17 @@ pub fn final_query_result_multi(mut cx: FunctionContext) -> JsResult<JsPromise>
 
     let promise = cx
         .task(move || {
-            result_data.prepare_results(&transform_requests, &cube_store_results)?;
+            result_data.prepare_pivot_query()?;
+
+            // As in `final_query_result`: render each `data` member while
+            // serializing, straight from its source columns.
+            let result_data = result_data.with_data(
+                transform_requests
+                    .iter()
+                    .zip(cube_store_results.iter())
+                    .map(|(request, source)| DirectData::new(request, source))
+                    .collect(),
+            )?;
 
             match serde_json::to_string(&result_data) {
                 Ok(json) => Ok(json),

rust/cube/cubeorchestrator/benches/common/mod.rs

@@ -1,7 +1,13 @@
 #![allow(dead_code)]
 
+use cubeorchestrator::query_message_parser::QueryResult;
 use cubeorchestrator::query_result_transform::{ColumnarArray, DBResponsePrimitive};
 use cubeorchestrator::transport::JsRawColumnarData;
+use cubeshared::codegen::{
+    HttpCommand, HttpMessage, HttpMessageArgs, HttpQueryResult, HttpQueryResultArgs,
+    HttpQueryResultArrow, HttpQueryResultArrowArgs, HttpQueryResultData,
+};
+use cubeshared::flatbuffers::FlatBufferBuilder;
 
 pub const ROW_COUNTS: &[usize] = &[1_000, 10_000, 50_000, 100_000];
 pub const COLUMN_COUNTS: &[usize] = &[8, 16, 32, 64];
@@ -79,17 +85,38 @@ pub fn build_dataset(
     JsRawColumnarData { members, columns }
 }
 
+/// How measure columns are typed in an Arrow fixture. CubeStore answers `SUM`
+/// with `Decimal128`, so that is the shape most measure cells really have; the
+/// `Float64` variant is kept because it is the cheaper baseline to compare against.
+#[derive(Clone, Copy, PartialEq, Eq)]
+pub enum MeasureKind {
+    Float64,
+    Decimal128,
+}
+
+impl MeasureKind {
+    pub fn label(self) -> &'static str {
+        match self {
+            MeasureKind::Float64 => "arrow",
+            MeasureKind::Decimal128 => "arrow_dec",
+        }
+    }
+}
+
 /// Build an Arrow IPC **stream** payload with the same logical data shape as
-/// [`build_dataset`]: dimensions as Utf8, measures as Float64, time dimensions
-/// as Timestamp(Millisecond). Used to compare Arrow parse throughput against the
-/// JSON path.
+/// [`build_dataset`]: dimensions as Utf8, measures per `measure_kind`, time
+/// dimensions as Timestamp(Millisecond). Used to compare Arrow parse throughput
+/// against the JSON path.
 pub fn build_arrow_ipc(
     row_count: usize,
     dimensions: &[(String, String)],
     measures: &[(String, String)],
     time_dims: &[TimeColumn],
+    measure_kind: MeasureKind,
 ) -> Vec<u8> {
-    use arrow::array::{ArrayRef, Float64Array, StringArray, TimestampMillisecondArray};
+    use arrow::array::{
+        ArrayRef, Decimal128Array, Float64Array, StringArray, TimestampMillisecondArray,
+    };
     use arrow::datatypes::{DataType, Field, Schema, TimeUnit};
     use arrow::ipc::writer::StreamWriter;
     use arrow::record_batch::RecordBatch;
@@ -99,19 +126,40 @@ pub fn build_arrow_ipc(
     let mut fields = Vec::with_capacity(total_cols);
     let mut columns: Vec<ArrayRef> = Vec::with_capacity(total_cols);
 
+    // Every column is built with `from_iter_values`, which fills the Arrow buffer
+    // straight from the iterator. Collecting into a `Vec` first would hold a
+    // second copy of the whole column — and for the string case, `row_count` live
+    // `String` allocations — just to hand it over.
     for (j, (_, alias)) in dimensions.iter().enumerate() {
         fields.push(Field::new(alias.clone(), DataType::Utf8, false));
-        let values: Vec<String> = (0..row_count)
-            .map(|i| format!("dim_{}_{}", j, i % 1000))
-            .collect();
-        columns.push(Arc::new(StringArray::from(values)));
+        columns.push(Arc::new(StringArray::from_iter_values(
+            (0..row_count).map(|i| format!("dim_{}_{}", j, i % 1000)),
+        )));
     }
     for (j, (_, alias)) in measures.iter().enumerate() {
-        fields.push(Field::new(alias.clone(), DataType::Float64, false));
-        let values: Vec<f64> = (0..row_count)
-            .map(|i| ((i * (j + 1)) as f64) * 0.5)
-            .collect();
-        columns.push(Arc::new(Float64Array::from(values)));
+        match measure_kind {
+            MeasureKind::Float64 => {
+                fields.push(Field::new(alias.clone(), DataType::Float64, false));
+                columns.push(Arc::new(Float64Array::from_iter_values(
+                    (0..row_count).map(|i| ((i * (j + 1)) as f64) * 0.5),
+                )));
+            }
+            MeasureKind::Decimal128 => {
+                fields.push(Field::new(
+                    alias.clone(),
+                    DataType::Decimal128(38, 2),
+                    false,
+                ));
+                // Same magnitudes as the Float64 arm, as a scale-2 mantissa.
+                columns.push(Arc::new(
+                    Decimal128Array::from_iter_values(
+                        (0..row_count).map(|i| ((i * (j + 1)) as i128) * 50),
+                    )
+                    .with_precision_and_scale(38, 2)
+                    .expect("decimal precision"),
+                ));
+            }
+        }
     }
     for (j, td) in time_dims.iter().enumerate() {
         fields.push(Field::new(
@@ -120,10 +168,9 @@ pub fn build_arrow_ipc(
             false,
         ));
         // One day apart, offset per column — arbitrary but realistic spread.
-        let values: Vec<i64> = (0..row_count)
-            .map(|i| ((i + j) as i64) * 86_400_000)
-            .collect();
-        columns.push(Arc::new(TimestampMillisecondArray::from(values)));
+        columns.push(Arc::new(TimestampMillisecondArray::from_iter_values(
+            (0..row_count).map(|i| ((i + j) as i64) * 86_400_000),
+        )));
     }
 
     let schema = Arc::new(Schema::new(fields));
@@ -137,3 +184,56 @@ pub fn build_arrow_ipc(
     }
     buf
 }
+
+/// Wrap raw Arrow IPC bytes in an `HttpMessage` FlatBuffer carrying
+/// `HttpQueryResultArrow`, exactly as CubeStore sends it.
+pub fn build_cubestore_fb_arrow_message(arrow_ipc: &[u8]) -> Vec<u8> {
+    let mut builder = FlatBufferBuilder::new();
+    let data_vec = builder.create_vector(arrow_ipc);
+    let arrow = HttpQueryResultArrow::create(
+        &mut builder,
+        &HttpQueryResultArrowArgs {
+            data: Some(data_vec),
+            is_last: true,
+        },
+    );
+    let query_result = HttpQueryResult::create(
+        &mut builder,
+        &HttpQueryResultArgs {
+            data_type: HttpQueryResultData::HttpQueryResultArrow,
+            data: Some(arrow.as_union_value()),
+        },
+    );
+    let connection_id = builder.create_string("bench_connection");
+    let message = HttpMessage::create(
+        &mut builder,
+        &HttpMessageArgs {
+            message_id: 1,
+            command_type: HttpCommand::HttpQueryResult,
+            command: Some(query_result.as_union_value()),
+            connection_id: Some(connection_id),
+        },
+    );
+    builder.finish(message, None);
+    builder.finished_data().to_vec()
+}
+
+/// An Arrow-backed `QueryResult` with the same logical shape as
+/// [`build_dataset`], so transform throughput can be compared column storage
+/// against column storage.
+///
+/// Note the time dimensions differ in kind, not just encoding: the Arrow fixture
+/// carries `Timestamp(Millisecond)` cells, which the `time` member type passes
+/// straight through, while [`build_dataset`] carries ISO strings that
+/// `transform_value` re-parses and re-formats per cell.
+pub fn build_arrow_query_result(
+    row_count: usize,
+    dimensions: &[(String, String)],
+    measures: &[(String, String)],
+    time_dims: &[TimeColumn],
+    measure_kind: MeasureKind,
+) -> QueryResult {
+    let ipc = build_arrow_ipc(row_count, dimensions, measures, time_dims, measure_kind);
+    let payload = build_cubestore_fb_arrow_message(&ipc);
+    QueryResult::from_cubestore_fb(&payload).expect("arrow query result")
+}

rust/cube/cubeorchestrator/benches/parser.rs

@@ -4,17 +4,16 @@ use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Through
 use cubeorchestrator::query_message_parser::QueryResult;
 use cubeorchestrator::transport::JsRawColumnarData;
 use cubeshared::codegen::{
-    HttpColumnValue, HttpColumnValueArgs, HttpCommand, HttpMessage, HttpMessageArgs,
-    HttpQueryResult, HttpQueryResultArgs, HttpQueryResultArrow, HttpQueryResultArrowArgs,
-    HttpQueryResultData, HttpResultSet, HttpResultSetArgs, HttpRow, HttpRowArgs,
+    HttpColumnValue, HttpColumnValueArgs, HttpCommand, HttpMessage, HttpMessageArgs, HttpResultSet,
+    HttpResultSetArgs, HttpRow, HttpRowArgs,
 };
 use cubeshared::flatbuffers::FlatBufferBuilder;
 
 #[path = "common/mod.rs"]
 mod common;
 use common::{
-    build_arrow_ipc, build_dataset, make_member_aliases, split_dim_measure, COLUMN_COUNTS,
-    ROW_COUNTS,
+    build_arrow_ipc, build_cubestore_fb_arrow_message, build_dataset, make_member_aliases,
+    split_dim_measure, MeasureKind, COLUMN_COUNTS, ROW_COUNTS,
 };
 
 /// Build a FlatBuffer `HttpMessage` payload mirroring CubeStore's wire format
@@ -153,38 +152,6 @@ fn bench_from_js_raw_data(c: &mut Criterion) {
     group.finish();
 }
 
-/// Wrap raw Arrow IPC bytes in an `HttpMessage` FlatBuffer carrying
-fn build_cubestore_fb_arrow_message(arrow_ipc: &[u8]) -> Vec<u8> {
-    let mut builder = FlatBufferBuilder::new();
-    let data_vec = builder.create_vector(arrow_ipc);
-    let arrow = HttpQueryResultArrow::create(
-        &mut builder,
-        &HttpQueryResultArrowArgs {
-            data: Some(data_vec),
-            is_last: true,
-        },
-    );
-    let query_result = HttpQueryResult::create(
-        &mut builder,
-        &HttpQueryResultArgs {
-            data_type: HttpQueryResultData::HttpQueryResultArrow,
-            data: Some(arrow.as_union_value()),
-        },
-    );
-    let connection_id = builder.create_string("bench_connection");
-    let message = HttpMessage::create(
-        &mut builder,
-        &HttpMessageArgs {
-            message_id: 1,
-            command_type: HttpCommand::HttpQueryResult,
-            command: Some(query_result.as_union_value()),
-            connection_id: Some(connection_id),
-        },
-    );
-    builder.finish(message, None);
-    builder.finished_data().to_vec()
-}
-
 fn bench_from_cubestore_fb_arrow(c: &mut Criterion) {
     let mut group = c.benchmark_group("QueryResult::from_cubestore_fb_arrow");
 
@@ -202,7 +169,8 @@ fn bench_from_cubestore_fb_arrow(c: &mut Criterion) {
         let dimensions = make_member_aliases("dim", dim_count);
         let measures = make_member_aliases("measure", measure_count);
 
-        let arrow_ipc = build_arrow_ipc(row_count, &dimensions, &measures, &[]);
+        let arrow_ipc =
+            build_arrow_ipc(row_count, &dimensions, &measures, &[], MeasureKind::Float64);
         let payload = build_cubestore_fb_arrow_message(&arrow_ipc);
         let payload_len = payload.len();
 
@@ -214,8 +182,10 @@ fn bench_from_cubestore_fb_arrow(c: &mut Criterion) {
         group.throughput(Throughput::Elements((row_count * col_count) as u64));
 
         let id = format!("c{:02}_r{}", col_count, row_count);
-        // Arrow IPC parse always materializes the QueryResult, so this measures
-        // the equivalent of from_js_raw_data's `parse_plus_build`.
+        // Arrow columns are kept in Arrow memory, so this measures the IPC decode
+        // and the column wiring — not a per-cell conversion. The conversion cost
+        // moved to `TransformedData::transform`; see the `arrow` axis in
+        // benches/transform.rs.
         group.bench_with_input(BenchmarkId::from_parameter(id), &(), |b, _| {
             b.iter(|| {
                 let built = QueryResult::from_cubestore_fb(black_box(&payload))