debug and fix native transform_data()

Author: KSDaemon

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

@@ -10,6 +10,7 @@ import {
   getRealType,
   QueryAlias,
 } from '@cubejs-backend/shared';
+import { transformData as transformDataNative, TransformDataResponse } from '@cubejs-backend/native';
 import type {
   Application as ExpressApplication,
   ErrorRequestHandler,
@@ -1594,9 +1595,20 @@ class ApiGateway {
     response: any,
     responseType?: ResultType,
   ) {
-    return {
-      query: normalizedQuery,
-      data: transformData(
+    const data = response.data.isNative ?
+      JSON.parse(transformDataNative(JSON.stringify({
+        aliasToMemberNameMap: sqlQuery.aliasNameToMember,
+        annotation: {
+          ...annotation.measures,
+          ...annotation.dimensions,
+          ...annotation.timeDimensions
+        },
+        query: normalizedQuery,
+        queryType,
+        resType: responseType,
+      }), response.data.getNativeRef()).result) as TransformDataResponse
+      :
+      transformData(
         sqlQuery.aliasNameToMember,
         {
           ...annotation.measures,
@@ -1607,7 +1619,10 @@ class ApiGateway {
         normalizedQuery,
         queryType,
         responseType,
-      ),
+      );
+    return {
+      query: normalizedQuery,
+      data,
       lastRefreshTime: response.lastRefreshTime?.toISOString(),
       ...(
         getEnv('devMode') ||

packages/cubejs-api-gateway/src/types/responses.ts

@@ -1,6 +1,6 @@
 import type { ConfigItem } from '../helpers/prepareAnnotation';
 import type { NormalizedQuery } from './query';
-import type { QueryType, ResultType } from './enums';
+import type { QueryType, ResultType } from './strings';
 
 export type DBResponsePrimitive =
   null |

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

@@ -77,4 +77,8 @@ export class CubeStoreResultWrapper {
     const array = this.getArray();
     return (array as any)[method](...args);
   }
+
+  public getNativeRef() {
+    return this.nativeReference;
+  }
 }

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

@@ -3,7 +3,6 @@ import fs from 'fs';
 import path from 'path';
 import { Writable } from 'stream';
 import type { Request as ExpressRequest } from 'express';
-import type { TransformDataRequest, TransformDataResponse, } from '@cubejs-backend/api-gateway';
 import { CubeStoreResultWrapper } from './CubeStoreResultWrapper';
 
 export * from './CubeStoreResultWrapper';
@@ -102,6 +101,23 @@ export type SQLInterfaceOptions = {
   gatewayPort?: number,
 };
 
+export type DBResponsePrimitive =
+  null |
+  boolean |
+  number |
+  string;
+
+export type TransformDataResponse = {
+  members: string[],
+  dataset: DBResponsePrimitive[][]
+} | {
+  [member: string]: DBResponsePrimitive
+}[];
+
+export type TransformDataResponseNative = {
+  result: string
+};
+
 export function loadNative() {
   // Development version
   if (fs.existsSync(path.join(__dirname, '/../../index.node'))) {
@@ -367,10 +383,10 @@ export const getCubestoreResult = (ref: CubeStoreResultWrapper): ResultRow[] =>
   return native.getCubestoreResult(ref);
 };
 
-export const transformData = (data: TransformDataRequest): TransformDataResponse => {
+export const transformData = (transformDataJson: string, rows: any): TransformDataResponseNative => {
   const native = loadNative();
 
-  return native.transformQueryData(data);
+  return native.transformQueryData(transformDataJson, rows);
 };
 
 export interface PyConfiguration {

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

@@ -38,7 +38,6 @@ use cubesql::{telemetry::ReportingLogger, CubeError};
 
 use neon::prelude::*;
 use neon::types::buffer::TypedArray;
-use serde::Deserialize;
 use cubeorchestrator::cubestore_result_transform::transform_data;
 use cubeorchestrator::types::{TransformDataRequest};
 
@@ -546,37 +545,29 @@ fn get_cubestore_result(mut cx: FunctionContext) -> JsResult<JsValue> {
     Ok(js_array.upcast())
 }
 
-fn js_object_to_struct<T>(cx: &mut FunctionContext) -> Result<T, neon::result::Throw>
-where
-    T: for<'de> Deserialize<'de>,
-{
+fn transform_query_data(mut cx: FunctionContext) -> JsResult<JsObject> {
     let json_str = cx
         .argument::<JsString>(0)?
-        .value(cx);
-
-    let result: Result<T, serde_json::Error> = serde_json::from_str(&json_str);
-    match result {
-        Ok(value) => Ok(value),
-        Err(err) => cx.throw_error(err.to_string()),
-    }
-}
+        .value(&mut cx);
+    let request_data = match serde_json::from_str::<TransformDataRequest>(&json_str) {
+        Ok(data) => data,
+        Err(err) => return cx.throw_error(err.to_string()),
+    };
 
-fn transform_query_data(mut cx: FunctionContext) -> JsResult<JsObject> {
-    let request_data: TransformDataRequest = js_object_to_struct(&mut cx)?;
+    let cube_store_result = cx.argument::<JsBox<CubeStoreResult>>(1)?;
 
     let alias_to_member_name_map = &request_data.alias_to_member_name_map;
     let annotation = &request_data.annotation;
-    let data = &request_data.data;
     let query = &request_data.query;
-    let query_type = &request_data.query_type;
+    let query_type = &request_data.query_type.unwrap_or_default();
     let res_type = &request_data.res_type;
 
     let transformed = match transform_data(
         alias_to_member_name_map,
         annotation,
-        data.clone(),
+        &**cube_store_result,
         query,
-        query_type,
+        &query_type,
         res_type.clone(),
     ) {
         Ok(data) => data,

rust/cubeorchestrator/src/cubestore_message_parser.rs

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use cubeshared::codegen::{root_as_http_message, HttpCommand};
 use neon::prelude::Finalize;
 
@@ -29,6 +30,7 @@ impl std::error::Error for ParseError {}
 pub struct CubeStoreResult {
     pub columns: Vec<String>,
     pub rows: Vec<Vec<String>>,
+    pub columns_pos: HashMap<String, usize>,
 }
 
 impl Finalize for CubeStoreResult {}
@@ -38,6 +40,7 @@ impl CubeStoreResult {
         let mut result = CubeStoreResult {
             columns: vec![],
             rows: vec![],
+            columns_pos: HashMap::new(),
         };
 
         let http_message =
@@ -62,10 +65,15 @@ impl CubeStoreResult {
                     return Err(ParseError::ColumnNameNotDefined);
                 }
 
-                result.columns = result_set_columns
+                let (columns, columns_pos): (Vec<_>, HashMap<_, _>) = result_set_columns
                     .iter()
-                    .map(|val| val.to_owned())
-                    .collect();
+                    .enumerate()
+                    .map(|(index, column_name)| (column_name.to_owned(), (column_name.to_owned(), index)))
+                    .unzip();
+
+                result.columns = columns;
+                result.columns_pos = columns_pos;
+
                 let result_set_rows = result_set.rows().ok_or(ParseError::EmptyResultSet)?;
                 result.rows = Vec::with_capacity(result_set_rows.len());