create CubeStoreResultWrapper class and switch to lazy evaluation of results set

Author: KSDaemon

packages/cubejs-backend-native/Cargo.lock

@@ -0,0 +1,77 @@
+import { getCubestoreResult, ResultRow } from './index';
+
+export class CubeStoreResultWrapper {
+  private readonly proxy: any;
+
+  private cache: any;
+
+  public constructor(private readonly nativeReference: any) {
+    this.proxy = new Proxy(this, {
+      get: (target, prop: string | symbol) => {
+        // intercept indexes
+        if (typeof prop === 'string' && !Number.isNaN(Number(prop))) {
+          const array = this.getArray();
+          return array[Number(prop)];
+        }
+
+        // intercept array methods
+        if (typeof prop === 'string' && prop in Array.prototype) {
+          const arrayMethod = (Array.prototype as any)[prop];
+          if (typeof arrayMethod === 'function') {
+            return (...args: any[]) => this.invokeArrayMethod(prop, ...args);
+          }
+        }
+
+        // intercept isNative
+        if (prop === 'isNative') {
+          return true;
+        }
+
+        // intercept array length
+        if (prop === 'length') {
+          return this.getArray().length;
+        }
+
+        // intercept JSON.stringify or toJSON()
+        if (prop === 'toJSON') {
+          return () => this.getArray();
+        }
+
+        return (target as any)[prop];
+      },
+
+      // intercept array length
+      getOwnPropertyDescriptor: (target, prop) => {
+        if (prop === 'length') {
+          const array = this.getArray();
+          return {
+            configurable: true,
+            enumerable: true,
+            value: array.length,
+            writable: false
+          };
+        }
+        return Object.getOwnPropertyDescriptor(target, prop);
+      },
+
+      ownKeys: (target) => {
+        const array = this.getArray();
+        return [...Object.keys(target), ...Object.keys(array), 'length', 'isNative'];
+      }
+    });
+
+    return this.proxy;
+  }
+
+  private getArray(): ResultRow[] {
+    if (!this.cache) {
+      this.cache = getCubestoreResult(this.nativeReference);
+    }
+    return this.cache;
+  }
+
+  private invokeArrayMethod(method: string, ...args: any[]): any {
+    const array = this.getArray();
+    return (array as any)[method](...args);
+  }
+}

packages/cubejs-backend-native/js/CubeStoreResultWrapper.ts

@@ -3,6 +3,9 @@ import fs from 'fs';
 import path from 'path';
 import { Writable } from 'stream';
 import type { Request as ExpressRequest } from 'express';
+import { CubeStoreResultWrapper } from './CubeStoreResultWrapper';
+
+export * from './CubeStoreResultWrapper';
 
 export interface BaseMeta {
   // postgres or mysql
@@ -350,10 +353,17 @@ export const buildSqlAndParams = (cubeEvaluator: any): String => {
 
 export type ResultRow = Record<string, string>;
 
-export const parseCubestoreResultMessage = (message: ArrayBuffer): ResultRow[] => {
+export const parseCubestoreResultMessage = (message: ArrayBuffer): CubeStoreResultWrapper => {
+  const native = loadNative();
+
+  const msg = native.parseCubestoreResultMessage(message);
+  return new CubeStoreResultWrapper(msg);
+};
+
+export const getCubestoreResult = (ref: CubeStoreResultWrapper): ResultRow[] => {
   const native = loadNative();
 
-  return native.parseCubestoreResultMessage(message);
+  return native.getCubestoreResult(ref);
 };
 
 export interface PyConfiguration {

packages/cubejs-backend-native/js/index.ts

@@ -32,7 +32,7 @@ use cubenativeutils::wrappers::NativeContextHolder;
 use cubesqlplanner::cube_bridge::base_query_options::NativeBaseQueryOptions;
 use cubesqlplanner::planner::base_query::BaseQuery;
 
-use cubeorchestrator::cubestore_message_parser::parse_cubestore_ws_result;
+use cubeorchestrator::cubestore_message_parser::CubeStoreResult;
 
 use cubesql::{telemetry::ReportingLogger, CubeError};
 
@@ -508,34 +508,41 @@ fn debug_js_to_clrepr_to_js(mut cx: FunctionContext) -> JsResult<JsValue> {
 
 //============ sql orchestrator ===================
 
-fn parse_cubestore_ws_result_message(mut cx: FunctionContext) -> JsResult<JsValue> {
+fn parse_cubestore_ws_result_message(mut cx: FunctionContext) -> JsResult<JsBox<CubeStoreResult>> {
     let msg = cx.argument::<JsBuffer>(0)?;
-    let msg_data = msg.as_slice(&cx).to_vec();
-    match parse_cubestore_ws_result(&msg_data) {
+    let msg_data = msg.as_slice(&cx);
+    match CubeStoreResult::new(msg_data) {
         Ok(result) => {
-            let js_array = cx.execute_scoped(|mut cx| {
-                let js_array = JsArray::new(&mut cx, result.rows.len());
-
-                for (i, row) in result.rows.iter().enumerate() {
-                    let js_row = cx.execute_scoped(|mut cx| {
-                        let js_row = JsObject::new(&mut cx);
-                        for (key, value) in result.columns.iter().zip(row.into_iter()) {
-                            let js_value = cx.string(*value);
-                            js_row.set(&mut cx, *key, js_value)?;
-                        }
-                        Ok(js_row)
-                    })?;
-
-                    js_array.set(&mut cx, i as u32, js_row)?;
-                }
+            Ok(cx.boxed(result))
+        }
+        Err(err) => cx.throw_error(err.to_string()),
+    }
+}
 
-                Ok(js_array)
+fn get_cubestore_result(mut cx: FunctionContext) -> JsResult<JsValue> {
+    let result = cx.argument::<JsBox<CubeStoreResult>>(0)?;
+    
+    let js_array = cx.execute_scoped(|mut cx| {
+        let js_array = JsArray::new(&mut cx, result.rows.len());
+
+        for (i, row) in result.rows.iter().enumerate() {
+            let js_row = cx.execute_scoped(|mut cx| {
+                let js_row = JsObject::new(&mut cx);
+                for (key, value) in result.columns.iter().zip(row.into_iter()) {
+                    let js_key = cx.string(key);
+                    let js_value = cx.string(value);
+                    js_row.set(&mut cx, js_key, js_value)?;
+                }
+                Ok(js_row)
             })?;
 
-            Ok(js_array.upcast())
+            js_array.set(&mut cx, i as u32, js_row)?;
         }
-        Err(err) => cx.throw_error(err.to_string()),
-    }
+
+        Ok(js_array)
+    })?;
+
+    Ok(js_array.upcast())
 }
 
 pub fn register_module_exports<C: NodeConfiguration + 'static>(
@@ -556,6 +563,7 @@ pub fn register_module_exports<C: NodeConfiguration + 'static>(
         "parseCubestoreResultMessage",
         parse_cubestore_ws_result_message,
     )?;
+    cx.export_function("getCubestoreResult", get_cubestore_result)?;
 
     crate::template::template_register_module(&mut cx)?;
 

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

@@ -1,3 +1,4 @@
+use neon::prelude::Finalize;
 use cubeshared::codegen::{root_as_http_message, HttpCommand};
 
 #[derive(Debug)]
@@ -23,55 +24,61 @@ impl std::fmt::Display for ParseError {
     }
 }
 
-pub struct CubeStoreResult<'a> {
-    pub columns: Vec<&'a str>,
-    pub rows: Vec<Vec<&'a str>>,
-}
-
 impl std::error::Error for ParseError {}
 
-pub fn parse_cubestore_ws_result(msg_data: &[u8]) -> Result<CubeStoreResult, ParseError> {
-    let http_message = root_as_http_message(msg_data).map_err(|_| ParseError::FlatBufferError)?;
+pub struct CubeStoreResult {
+    pub columns: Vec<String>,
+    pub rows: Vec<Vec<String>>,
+}
 
-    let command_type = http_message.command_type();
+impl Finalize for CubeStoreResult {}
 
-    match command_type {
-        HttpCommand::HttpError => {
-            let http_error = http_message
-                .command_as_http_error()
-                .ok_or(ParseError::FlatBufferError)?;
-            let error_message = http_error.error().unwrap_or("Unknown error").to_string();
-            Err(ParseError::ErrorMessage(error_message))
-        }
-        HttpCommand::HttpResultSet => {
-            let result_set = http_message
-                .command_as_http_result_set()
-                .ok_or(ParseError::EmptyResultSet)?;
+impl CubeStoreResult {
+    pub fn new(msg_data: &[u8]) -> Result<Self, ParseError> {
+        let mut result = CubeStoreResult {
+            columns: vec![],
+            rows: vec![],
+        };
 
-            let result_set_columns = result_set.columns().ok_or(ParseError::EmptyResultSet)?;
+        let http_message = root_as_http_message(msg_data)
+            .map_err(|_| ParseError::FlatBufferError)?;
 
-            if result_set_columns.iter().any(|c| c.is_empty()) {
-                return Err(ParseError::ColumnNameNotDefined);
+        match http_message.command_type() {
+            HttpCommand::HttpError => {
+                let http_error = http_message
+                    .command_as_http_error()
+                    .ok_or(ParseError::FlatBufferError)?;
+                let error_message = http_error.error().unwrap_or("Unknown error").to_string();
+                Err(ParseError::ErrorMessage(error_message))
             }
+            HttpCommand::HttpResultSet => {
+                let result_set = http_message
+                    .command_as_http_result_set()
+                    .ok_or(ParseError::EmptyResultSet)?;
 
-            let result_set_rows = result_set.rows().ok_or(ParseError::EmptyResultSet)?;
-            let mut result = CubeStoreResult{
-                columns: result_set_columns.iter().collect(),
-                rows: Vec::with_capacity(result_set_rows.len())
-            };
+                let result_set_columns = result_set.columns().ok_or(ParseError::EmptyResultSet)?;
 
-            for row in result_set_rows.iter() {
-                let values = row.values().ok_or(ParseError::NullRow)?;
-                let row_obj: Vec<_> = values
-                    .iter()
-                    .map(|val| val.string_value().unwrap_or(""))
-                    .collect();
+                if result_set_columns.iter().any(|c| c.is_empty()) {
+                    return Err(ParseError::ColumnNameNotDefined);
+                }
 
-                result.rows.push(row_obj);
-            }
+                result.columns = result_set_columns.iter().map(|val| val.to_owned()).collect();
+                let result_set_rows = result_set.rows().ok_or(ParseError::EmptyResultSet)?;
+                result.rows = Vec::with_capacity(result_set_rows.len());
+
+                for row in result_set_rows.iter() {
+                    let values = row.values().ok_or(ParseError::NullRow)?;
+                    let row_obj: Vec<_> = values
+                        .iter()
+                        .map(|val| val.string_value().unwrap_or("").to_owned())
+                        .collect();
+
+                    result.rows.push(row_obj);
+                }
 
-            Ok(result)
+                Ok(result)
+            },
+            _ => Err(ParseError::UnsupportedCommand),
         }
-        _ => Err(ParseError::UnsupportedCommand),
     }
 }