implement custom deserializer from JS Value to rust

KSDaemon · KSDaemon · commit 0c89ee84d922 · 2024-12-17T01:05:25.000+02:00
and updated the cubestore result transformations with it
diff --git a/packages/cubejs-api-gateway/src/gateway.ts b/packages/cubejs-api-gateway/src/gateway.ts
@@ -1629,7 +1629,7 @@ class ApiGateway {
     const dataCb: TransformDataResponseCb = response.data.isNative ?
       async () => {
         const jsonData = await transformDataNative(
-          JSON.stringify(transformDataParams), response.data.getNativeRef()
+          transformDataParams, response.data.getNativeRef()
         );
         return JSON.parse(jsonData.result) as TransformDataResponse;
       }
@@ -1828,9 +1828,9 @@ class ApiGateway {
         // If all query results are from Cubestore (are native)
         // we prepare the final json result on native side
         if (allNative) {
-          const [transformDataJson, rawDataRef, cleanResultList] = results.reduce<[string[], any[], any[]]>(
+          const [transformDataJson, rawDataRef, cleanResultList] = results.reduce<[Object[], any[], Object[]]>(
             ([transformList, rawList, resultList], r) => {
-              transformList.push(JSON.stringify(r.transformDataParams));
+              transformList.push(r.transformDataParams);
               rawList.push(r.rawData.getNativeRef());
               resultList.push(cleanupResult(r));
               return [transformList, rawList, resultList];
@@ -1843,9 +1843,8 @@ class ApiGateway {
             results: cleanResultList,
             slowQuery
           };
-          const resultDataJson = JSON.stringify(responseDataObj);
 
-          res(await getFinalCubestoreResultMulti(transformDataJson, rawDataRef, resultDataJson));
+          res(await getFinalCubestoreResultMulti(transformDataJson, rawDataRef, responseDataObj));
         } else {
           // if we have mixed query results (there are js and native)
           // we prepare results separately: on js and native sides
@@ -1866,10 +1865,8 @@ class ApiGateway {
       } else if (allNative) {
         // We prepare the full final json result on native side
         const r = results[0];
-        const transformDataJson = JSON.stringify(r.transformDataParams);
         const rawDataRef = r.rawData.getNativeRef();
-        const resultDataJson = JSON.stringify(cleanupResult(r));
-        res(await getFinalCubestoreResult(transformDataJson, rawDataRef, resultDataJson));
+        res(await getFinalCubestoreResult(r.transformDataParams, rawDataRef, cleanupResult(r)));
       } else {
         const data = await results[0].dataCb();
         res({
@@ -2014,11 +2011,11 @@ class ApiGateway {
             const [transformDataJson, rawDataRef, resultDataJson] = (results as {
               transformDataParams: any;
               rawData: { getNativeRef: () => any };
-            }[]).reduce<[string[], any[], string[]]>(
+            }[]).reduce<[Object[], any[], Object[]]>(
               ([transformList, rawList, resultList], r) => {
-                transformList.push(JSON.stringify(r.transformDataParams));
+                transformList.push(r.transformDataParams);
                 rawList.push(r.rawData.getNativeRef());
-                resultList.push(JSON.stringify(cleanupResult(r)));
+                resultList.push(cleanupResult(r));
                 return [transformList, rawList, resultList];
               },
               [[], [], []]
diff --git a/packages/cubejs-backend-native/js/index.ts b/packages/cubejs-backend-native/js/index.ts
@@ -383,28 +383,28 @@ export const getCubestoreResult = (ref: CubeStoreResultWrapper): ResultRow[] =>
   return native.getCubestoreResult(ref);
 };
 
-export const transformData = (transformDataJson: string, rows: any): Promise<TransformDataResponseNative> => {
+export const transformData = (transformDataObj: Object, rows: any): Promise<TransformDataResponseNative> => {
   const native = loadNative();
 
-  return native.transformQueryData(transformDataJson, rows);
+  return native.transformQueryData(transformDataObj, rows);
 };
 
-export const getFinalCubestoreResult = (transformDataJson: string, rows: any, resultData: string): Promise<ArrayBuffer> => {
+export const getFinalCubestoreResult = (transformDataObj: Object, rows: any, resultData: Object): Promise<ArrayBuffer> => {
   const native = loadNative();
 
-  return native.getFinalCubestoreResult(transformDataJson, rows, resultData);
+  return native.getFinalCubestoreResult(transformDataObj, rows, resultData);
 };
 
-export const getFinalCubestoreResultArray = (transformDataJson: string[], rows: any[], resultDataJson: string[]): Promise<ArrayBuffer> => {
+export const getFinalCubestoreResultArray = (transformDataArr: Object[], rows: any[], resultDataArr: Object[]): Promise<ArrayBuffer> => {
   const native = loadNative();
 
-  return native.getFinalCubestoreResultArray(transformDataJson, rows, resultDataJson);
+  return native.getFinalCubestoreResultArray(transformDataArr, rows, resultDataArr);
 };
 
-export const getFinalCubestoreResultMulti = (transformDataJson: string[], rows: any[], responseData: string): Promise<ArrayBuffer> => {
+export const getFinalCubestoreResultMulti = (transformDataArr: Object[], rows: any[], responseData: Object): Promise<ArrayBuffer> => {
   const native = loadNative();
 
-  return native.getFinalCubestoreResultMulti(transformDataJson, rows, responseData);
+  return native.getFinalCubestoreResultMulti(transformDataArr, rows, responseData);
 };
 
 export interface PyConfiguration {
diff --git a/packages/cubejs-backend-native/src/lib.rs b/packages/cubejs-backend-native/src/lib.rs
@@ -11,6 +11,7 @@ pub mod cross;
 pub mod gateway;
 pub mod logger;
 pub mod node_export;
+pub mod node_obj_deserializer;
 pub mod node_obj_serializer;
 #[cfg(feature = "python")]
 pub mod python;
diff --git a/packages/cubejs-backend-native/src/node_export.rs b/packages/cubejs-backend-native/src/node_export.rs
@@ -35,13 +35,15 @@ use cubeorchestrator::cubestore_message_parser::CubeStoreResult;
 
 use cubesql::{telemetry::ReportingLogger, CubeError};
 
+use crate::node_obj_deserializer::JsValueDeserializer;
 use cubeorchestrator::cubestore_result_transform::{
     get_final_cubestore_result_array, RequestResultArray, RequestResultData,
     RequestResultDataMulti, TransformedData,
 };
 use cubeorchestrator::transport::TransformDataRequest;
 use neon::prelude::*;
 use neon::types::buffer::TypedArray;
+use serde::Deserialize;
 
 struct SQLInterface {
     services: Arc<NodeCubeServices>,
@@ -553,18 +555,19 @@ fn get_cubestore_result(mut cx: FunctionContext) -> JsResult<JsValue> {
 }
 
 fn transform_query_data(mut cx: FunctionContext) -> JsResult<JsPromise> {
-    let json_str = cx.argument::<JsString>(0)?.value(&mut cx);
+    let transform_data_js_object = cx.argument::<JsValue>(0)?;
+    let deserializer = JsValueDeserializer::new(&mut cx, transform_data_js_object);
+
+    let request_data: TransformDataRequest = match Deserialize::deserialize(deserializer) {
+        Ok(data) => data,
+        Err(err) => return cx.throw_error(err.to_string()),
+    };
 
     let cube_store_result = cx.argument::<JsBox<Arc<CubeStoreResult>>>(1)?;
     let cube_store_result = Arc::clone(&cube_store_result);
 
     let promise = cx
         .task(move || {
-            let request_data = match serde_json::from_str::<TransformDataRequest>(&json_str) {
-                Ok(data) => data,
-                Err(err) => return Err(anyhow::Error::from(err)),
-            };
-
             let transformed = TransformedData::transform(&request_data, &cube_store_result)?;
 
             match serde_json::to_string(&transformed) {
@@ -588,25 +591,25 @@ fn transform_query_data(mut cx: FunctionContext) -> JsResult<JsPromise> {
 }
 
 fn final_cubestore_result(mut cx: FunctionContext) -> JsResult<JsPromise> {
-    let transform_data_str = cx.argument::<JsString>(0)?.value(&mut cx);
+    let transform_data_js_object = cx.argument::<JsValue>(0)?;
+    let deserializer = JsValueDeserializer::new(&mut cx, transform_data_js_object);
+    let transform_request_data: TransformDataRequest = match Deserialize::deserialize(deserializer)
+    {
+        Ok(data) => data,
+        Err(err) => return cx.throw_error(err.to_string()),
+    };
+
     let cube_store_result = cx.argument::<JsBox<Arc<CubeStoreResult>>>(1)?;
     let cube_store_result = Arc::clone(&cube_store_result);
-    let result_data_str = cx.argument::<JsString>(2)?.value(&mut cx);
+    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) {
+        Ok(data) => data,
+        Err(err) => return cx.throw_error(err.to_string()),
+    };
 
     let promise = cx
         .task(move || {
-            let transform_request_data =
-                match serde_json::from_str::<TransformDataRequest>(&transform_data_str) {
-                    Ok(data) => data,
-                    Err(err) => return Err(anyhow::Error::from(err)),
-                };
-
-            let mut result_data = match serde_json::from_str::<RequestResultData>(&result_data_str)
-            {
-                Ok(data) => data,
-                Err(err) => return Err(anyhow::Error::from(err)),
-            };
-
             result_data.prepare_results(&transform_request_data, &cube_store_result)?;
 
             match serde_json::to_string(&result_data) {
@@ -633,16 +636,13 @@ fn final_cubestore_result(mut cx: FunctionContext) -> JsResult<JsPromise> {
 }
 
 fn final_cubestore_result_array(mut cx: FunctionContext) -> JsResult<JsPromise> {
-    let transform_data_array = cx.argument::<JsArray>(0)?;
-    let transform_request_strings: Vec<String> = transform_data_array
-        .to_vec(&mut cx)?
-        .into_iter()
-        .map(|js_value| {
-            js_value
-                .downcast_or_throw::<JsString, _>(&mut cx)
-                .map(|js_string| js_string.value(&mut cx))
-        })
-        .collect::<Result<_, _>>()?;
+    let transform_data_array = cx.argument::<JsValue>(0)?;
+    let deserializer = JsValueDeserializer::new(&mut cx, transform_data_array);
+    let transform_requests: Vec<TransformDataRequest> = match Deserialize::deserialize(deserializer)
+    {
+        Ok(data) => data,
+        Err(err) => return cx.throw_error(err.to_string()),
+    };
 
     let cube_store_array = cx.argument::<JsArray>(1)?;
     let cube_store_results_boxed: Vec<Handle<JsBox<Arc<CubeStoreResult>>>> = cube_store_array
@@ -655,39 +655,15 @@ fn final_cubestore_result_array(mut cx: FunctionContext) -> JsResult<JsPromise>
         .map(|handle| (**handle).clone())
         .collect();
 
-    let results_data_array = cx.argument::<JsArray>(2)?;
-    let request_result_strings: Vec<String> = results_data_array
-        .to_vec(&mut cx)?
-        .into_iter()
-        .map(|js_value| {
-            js_value
-                .downcast_or_throw::<JsString, _>(&mut cx)
-                .map(|js_string| js_string.value(&mut cx))
-        })
-        .collect::<Result<_, _>>()?;
+    let results_data_array = cx.argument::<JsValue>(2)?;
+    let deserializer = JsValueDeserializer::new(&mut cx, results_data_array);
+    let mut request_results: Vec<RequestResultData> = match Deserialize::deserialize(deserializer) {
+        Ok(data) => data,
+        Err(err) => return cx.throw_error(err.to_string()),
+    };
 
     let promise = cx
         .task(move || {
-            let transform_requests: Vec<TransformDataRequest> = transform_request_strings
-                .into_iter()
-                .map(
-                    |req_str| match serde_json::from_str::<TransformDataRequest>(&req_str) {
-                        Ok(request) => Ok(request),
-                        Err(err) => Err(anyhow::Error::from(err)),
-                    },
-                )
-                .collect::<Result<_, _>>()?;
-
-            let mut request_results: Vec<RequestResultData> = request_result_strings
-                .into_iter()
-                .map(
-                    |req_str| match serde_json::from_str::<RequestResultData>(&req_str) {
-                        Ok(request) => Ok(request),
-                        Err(err) => Err(anyhow::Error::from(err)),
-                    },
-                )
-                .collect::<Result<_, _>>()?;
-
             get_final_cubestore_result_array(
                 &transform_requests,
                 &cube_store_results,
@@ -722,16 +698,13 @@ fn final_cubestore_result_array(mut cx: FunctionContext) -> JsResult<JsPromise>
 }
 
 fn final_cubestore_result_multi(mut cx: FunctionContext) -> JsResult<JsPromise> {
-    let transform_data_array = cx.argument::<JsArray>(0)?;
-    let transform_request_strings: Vec<String> = transform_data_array
-        .to_vec(&mut cx)?
-        .into_iter()
-        .map(|js_value| {
-            js_value
-                .downcast_or_throw::<JsString, _>(&mut cx)
-                .map(|js_string| js_string.value(&mut cx))
-        })
-        .collect::<Result<_, _>>()?;
+    let transform_data_array = cx.argument::<JsValue>(0)?;
+    let deserializer = JsValueDeserializer::new(&mut cx, transform_data_array);
+    let transform_requests: Vec<TransformDataRequest> = match Deserialize::deserialize(deserializer)
+    {
+        Ok(data) => data,
+        Err(err) => return cx.throw_error(err.to_string()),
+    };
 
     let cube_store_array = cx.argument::<JsArray>(1)?;
     let cube_store_results_boxed: Vec<Handle<JsBox<Arc<CubeStoreResult>>>> = cube_store_array
@@ -744,26 +717,15 @@ fn final_cubestore_result_multi(mut cx: FunctionContext) -> JsResult<JsPromise>
         .map(|handle| (**handle).clone())
         .collect();
 
-    let result_data_str = cx.argument::<JsString>(2)?.value(&mut cx);
+    let result_data_js_object = cx.argument::<JsValue>(2)?;
+    let deserializer = JsValueDeserializer::new(&mut cx, result_data_js_object);
+    let mut result_data: RequestResultDataMulti = match Deserialize::deserialize(deserializer) {
+        Ok(data) => data,
+        Err(err) => return cx.throw_error(err.to_string()),
+    };
 
     let promise = cx
         .task(move || {
-            let transform_requests: Vec<TransformDataRequest> = transform_request_strings
-                .into_iter()
-                .map(
-                    |req_str| match serde_json::from_str::<TransformDataRequest>(&req_str) {
-                        Ok(request) => Ok(request),
-                        Err(err) => Err(anyhow::Error::from(err)),
-                    },
-                )
-                .collect::<Result<_, _>>()?;
-
-            let mut result_data =
-                match serde_json::from_str::<RequestResultDataMulti>(&result_data_str) {
-                    Ok(data) => data,
-                    Err(err) => return Err(anyhow::Error::from(err)),
-                };
-
             result_data.prepare_results(&transform_requests, &cube_store_results)?;
 
             match serde_json::to_string(&result_data) {
diff --git a/packages/cubejs-backend-native/src/node_obj_deserializer.rs b/packages/cubejs-backend-native/src/node_obj_deserializer.rs
diff --git a/rust/cubeorchestrator/src/cubestore_result_transform.rs b/rust/cubeorchestrator/src/cubestore_result_transform.rs
