feat(api-gateway): Async native query results transformations (#8961)

* create a basic cubeorchestrator project structure

* wip

* move flatbuffer schema/code to separate crate

* implement parse_cubestore_ws_result

* add cubeorchestrator/parse_cubestore_ws_result_message export

* use native parseCubestoreResultMessage

* init hashmap with capacity

* cargo fmt

* some optimizations and improvements

* a bit optimized version

* use cx.execute_scoped for optimization

* a bit more rust idiomatic code

* put native parseCubestoreResultMessage behind the flag

* tiny improvement

* cargo fmt

* cargo fmt

* cargo clippy fix

* update cubestore Dockerfile

* cargo fmt

* update cubestore Docker builds

* introduce CubeStoreResult struct

* create CubeStoreResultWrapper class and switch to lazy evaluation of results set

* add resToRawResultFn in API GW

* cargo fmt

* update resToResultFn

* call prepareAnnotation later

* remove bytes

* add cached flag to CubeStoreResultWrapper

* convert core data types from api gateway

* implement transformData and related helpers

* cargo fmt

* down chrono to same version as in cubesql (0.4.31)

* fail fast in api gw load()

* update cargo.lock (syn crate)

* linter fix

* implement get_query_granularities & get_pivot_query

* prepare transformQueryData native wrapper

* small optimization: do not use native parsing for short messages

* refactor transformValue and related

* types restructure

* debug and fix native transform_data()

* lazy transformData evaluation

* omplement get_final_cubestore_result & get_final_cubestore_result_multi in native

* cargo fmt

* cargo clippy fix

* refactor getVanillaRow

* implement get_final_cubestore_result_array() native

* fix native response flow for sqlApiLoad

* add postgres with native cubestore results driver tests

* Build native (without Python) in drivers tests

* workaround for native build in testings-drivers

* small improvements in CubeStoreResultWrapper

* make parse_cubestore_ws_result_message async

* make all native cubestore_result_transform functions async

* cargo fmt

* refactor results transformations

* yarn sync

* implement custom deserializer from JS Value to rust

and updated the cubestore result transformations with it

* refactor json_to_array_buffer()

* a bit of refactoring

* code rearrangement

* switch to use DBResponsePrimitive instead of just Strings

* refactoring

* always use transform data native for all results

* fix dtos for native query processing (thnx unit tests)

* cache loadNative

* remove not needed anymore native.transformQueryData

* add Build native for unit tests in CI

* refactor serde annotations

* remove unused

* annotate native query results processing functions

* add Build native for unit tests in CI for Debian without pushing

* add few unit tests

* fix some tests

* attempt to fix native build/test in docker-dev CI

* fix empty result set issue

* another fix in datetime parsing

* another fix in deserialization

* another fix in datetime parsing

* update postgres-native-cubestore-response-full.test.ts.snap

* another fix in deserialization

* cargo fmt

* another fix in datetime parsing

* update postgres-native-cubestore-response-full.test.ts.snap

* attempt to fix native build/test in docker-dev CI

* add some comments

* edits in result processing when streaming

* fix native response processing via websocket

* fix yarn lock

* attempt to fix native build/test in cloud integration tests in CI

* recreated unit tests for transform data in native

* run rust unit tests on push

* commented out cargo fmt/build/test for cubesql, cubesqlplanner

* remove transformdata JS implementation

* refactor push CI jobs

* rename CubestoreResultWrapper → ResultWrapper

* add isNative flag to ResultWrapper

* rename getResultInternal → prepareResultTransformData

* use ResultWrapper for all results

* fix getArray in ResultWrapper

* lint fix

* encapsulate rootResultObject into ResultWrapper

* add DataResult interface + implement it in result wrappers

* add isWrapper property to wrappers

* transform wrapped result later before returning the response

* fix lint warn

* fix async request porcessing

* some node version fix

* weird yarn.lock change. Wtf?

* full transition of result native-js-native

* fix @cubejs-backend/native version ref

* refactor ResultWrapper constructor

* fix string deserialize in transport

* extend DataResult interface

* support iterative access in ResultWrapper

* refactor ResultWrapper classes

* pass ResultWrapper to native and offload transformData from the eventloop

* remove obsolete getFinalQueryResultArray

* fix ws api subscriptionServer results processing

* pass SchemaRef to load methods.

* trying to fix flaky ws failing tests

* remove cubeorchestrator dependency from cubesql

* rewrite all load api calls to return RecordBatch

* fix @cubejs-backend/native version ref after rebase

* set prototype of ResultWrapper proxy to class prototype

* fix linter warnings

* fix yarn.lock

* linter fix

* Set CUBEJS_TESSERACT_ORCHESTRATOR: true for all tests workflows

* fix old refs after rebase

* update integration-cubestore job with the build of backend-native

* remove postgres-native-cubestore-response from testing drivers as we turnon new orchestrator for all tests

Author: KSDaemon

.github/workflows/cloud.yml

@@ -25,6 +25,9 @@ on:
       - 'package.json'
       - 'yarn.lock'
 
+env:
+  CUBEJS_TESSERACT_ORCHESTRATOR: true
+
 jobs:
   latest-tag-sha:
     runs-on: ubuntu-20.04
@@ -59,15 +62,31 @@ jobs:
       matrix:
         node-version: [ 20.x ]
         db: [ 'athena', 'bigquery', 'snowflake' ]
+        target: [ "x86_64-unknown-linux-gnu" ]
       fail-fast: false
 
     steps:
       - name: Checkout
         uses: actions/checkout@v4
+      - name: Install Rust
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+        with:
+          toolchain: nightly-2024-07-15
+          # override: true # this is by default on
+          rustflags: ""
+          components: rustfmt
+          target: ${{ matrix.target }}
       - name: Install Node.js ${{ matrix.node-version }}
         uses: actions/setup-node@v4
         with:
           node-version: ${{ matrix.node-version }}
+      - name: Install cargo-cp-artifact
+        run: npm install -g [email protected]
+      - uses: Swatinem/rust-cache@v2
+        with:
+          workspaces: ./packages/cubejs-backend-native
+          key: native-${{ runner.OS }}-${{ matrix.target }}
+          shared-key: native-${{ runner.OS }}-${{ matrix.target }}
       - name: Get yarn cache directory path
         id: yarn-cache-dir-path
         run: echo "dir=$(yarn cache dir)" >> "$GITHUB_OUTPUT"
@@ -87,6 +106,8 @@ jobs:
         run: yarn build
       - name: Lerna tsc
         run: yarn tsc
+      - name: Build native (no python)
+        run: cd packages/cubejs-backend-native && npm run native:build-release
       - name: Run Integration tests for ${{ matrix.db }} matrix
         timeout-minutes: 30
         env:

.github/workflows/drivers-tests.yml

@@ -56,6 +56,9 @@ on:
       - 'packages/cubejs-backend-native/**'
       - 'rust/cubesql/**'
 
+env:
+  CUBEJS_TESSERACT_ORCHESTRATOR: true
+
 jobs:
   latest-tag-sha:
     runs-on: ubuntu-20.04
@@ -117,8 +120,8 @@ jobs:
       - uses: Swatinem/rust-cache@v2
         with:
           workspaces: ./packages/cubejs-backend-native
-          key: native-${{ runner.OS }}-x86_64-unknown-linux-gnu
-          shared-key: native-${{ runner.OS }}-x86_64-unknown-linux-gnu
+          key: native-${{ runner.OS }}-${{ matrix.target }}
+          shared-key: native-${{ runner.OS }}-${{ matrix.target }}
       - name: Build native (fallback)
         if: (matrix.python-version == 'fallback')
         env:
@@ -165,7 +168,12 @@ jobs:
         uses: actions/download-artifact@v4
         with:
           name: backend-native
-          path: packages/cubejs-backend-native/
+          path: packages/cubejs-backend-native
+      # current .dockerignore prevents use of native build
+      - name: Unignore native from .dockerignore
+        run: |
+          grep -v -E "packages/cubejs-backend-native/((native)|(index.node))" .dockerignore > .dockerignore.tmp
+          mv .dockerignore.tmp .dockerignore
       - name: Build and push
         uses: docker/build-push-action@v6
         with:

.github/workflows/master.yml

@@ -16,6 +16,8 @@ on:
       - 'rust/cubesql/**'
     branches:
       - master
+env:
+  CUBEJS_TESSERACT_ORCHESTRATOR: true
 jobs:
   latest-tag-sha:
     runs-on: ubuntu-20.04

.github/workflows/publish.yml

@@ -552,7 +552,7 @@ jobs:
       - name: Push to Docker Hub
         uses: docker/build-push-action@v6
         with:
-          context: ./rust/cubestore/
+          context: ./rust/
           file: ./rust/cubestore/Dockerfile
           platforms: ${{ matrix.platforms }}
           build-args: ${{ matrix.build-args }}

.github/workflows/push.yml

@@ -9,6 +9,10 @@ on:
       - 'packages/**'
       - 'rust/cubestore/**'
       - 'rust/cubesql/**'
+      - 'rust/cubenativeutils/**'
+      - 'rust/cubeorchestrator/**'
+      - 'rust/cubeshared/**'
+      - 'rust/cubesqlplanner/**'
       - '.eslintrc.js'
       - '.prettierrc'
       - 'package.json'
@@ -24,13 +28,20 @@ on:
       - 'packages/**'
       - 'rust/cubestore/**'
       - 'rust/cubesql/**'
+      - 'rust/cubenativeutils/**'
+      - 'rust/cubeorchestrator/**'
+      - 'rust/cubeshared/**'
+      - 'rust/cubesqlplanner/**'
       - '.eslintrc.js'
       - '.prettierrc'
       - 'package.json'
       - 'lerna.json'
       - 'rollup.config.js'
       - 'yarn.lock'
 
+env:
+  CUBEJS_TESSERACT_ORCHESTRATOR: true
+
 jobs:
   unit:
     runs-on: ubuntu-20.04
@@ -95,14 +106,15 @@ jobs:
           command: yarn install --frozen-lockfile
       - name: Lerna tsc
         run: yarn tsc
+      - name: Build native (no python)
+        run: cd packages/cubejs-backend-native && npm run native:build-release
       - name: Build client
         run: yarn build
       - name: Build cubejs-backend-native (with Python)
         run: yarn run native:build-release-python
         working-directory: ./packages/cubejs-backend-native
         env:
           PYO3_PYTHON: python${{ matrix.python-version }}
-
       - name: Lerna test
         run: yarn lerna run --concurrency 1 --stream --no-prefix unit
 #      - uses: codecov/codecov-action@v1
@@ -111,6 +123,21 @@ jobs:
 #          files: ./packages/*/coverage/clover.xml
 #          flags: cube-backend
 #          verbose: true # optional (default = false)
+      - name: Cargo test cubeorchestrator
+        run: |
+          cargo test --manifest-path rust/cubeorchestrator/Cargo.toml -j 1
+      - name: Cargo test cubenativeutils
+        run: |
+          cargo test --manifest-path rust/cubenativeutils/Cargo.toml -j 1
+      - name: Cargo test cubeshared
+        run: |
+          cargo test --manifest-path rust/cubeshared/Cargo.toml -j 1
+#      - name: Cargo test cubesql
+#        run: |
+#          cargo test --manifest-path rust/cubesql/Cargo.toml -j 1
+#      - name: Cargo test cubesqlplanner
+#        run: |
+#          cargo test --manifest-path rust/cubesqlplanner/cubesqlplanner/Cargo.toml -j 1
 
   lint:
     runs-on: ubuntu-20.04
@@ -159,6 +186,21 @@ jobs:
         run: yarn lint:npm
       - name: Lerna lint
         run: yarn lerna run --concurrency 1 lint
+      - name: Cargo fmt cubeorchestrator
+        run: |
+          cargo fmt --manifest-path rust/cubeorchestrator/Cargo.toml -- --check
+      - name: Cargo fmt cubenativeutils
+        run: |
+          cargo fmt --manifest-path rust/cubenativeutils/Cargo.toml -- --check
+      - name: Cargo fmt cubeshared
+        run: |
+          cargo fmt --manifest-path rust/cubeshared/Cargo.toml -- --check
+#      - name: Cargo fmt cubesql
+#        run: |
+#          cargo fmt --manifest-path rust/cubesql/Cargo.toml -- --check
+#      - name: Cargo fmt cubesqlplanner
+#        run: |
+#          cargo fmt --manifest-path rust/cubesqlplanner/cubesqlplanner/Cargo.toml -- --check
 
   build:
     runs-on: ubuntu-20.04
@@ -211,6 +253,21 @@ jobs:
         run: yarn lerna run --concurrency 1 build
         env:
           NODE_OPTIONS: --max_old_space_size=4096
+      - name: Cargo build cubeorchestrator
+        run: |
+          cargo build --manifest-path rust/cubeorchestrator/Cargo.toml -j 4
+      - name: Cargo build cubenativeutils
+        run: |
+          cargo build --manifest-path rust/cubenativeutils/Cargo.toml -j 4
+      - name: Cargo build cubeshared
+        run: |
+          cargo build --manifest-path rust/cubeshared/Cargo.toml -j 4
+#      - name: Cargo build cubesql
+#        run: |
+#          cargo build --manifest-path rust/cubesql/Cargo.toml -j 4
+#      - name: Cargo build cubesqlplanner
+#        run: |
+#          cargo build --manifest-path rust/cubesqlplanner/cubesqlplanner/Cargo.toml -j 4
 
   build-cubestore:
     needs: [latest-tag-sha]
@@ -276,6 +333,13 @@ jobs:
           df -h
       - name: Checkout
         uses: actions/checkout@v4
+      - name: Install Rust
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+        with:
+          toolchain: nightly-2024-07-15
+          # override: true # this is by default on
+          rustflags: ""
+          components: rustfmt
       - name: Install Node.js ${{ matrix.node-version }}
         uses: actions/setup-node@v4
         with:
@@ -305,6 +369,9 @@ jobs:
           command: yarn install --frozen-lockfile
       - name: Lerna tsc
         run: yarn tsc
+      - name: Build cubejs-backend-native (without Python)
+        run: yarn run native:build-release
+        working-directory: ./packages/cubejs-backend-native
       - name: Download cubestored-x86_64-unknown-linux-gnu-release artifact
         uses: actions/download-artifact@v4
         with:
@@ -539,6 +606,8 @@ jobs:
           - 5000:5000
     strategy:
       matrix:
+        node-version: [ 20 ]
+        target: [ "x86_64-unknown-linux-gnu" ]
         dockerfile:
           - dev.Dockerfile
         include:
@@ -565,21 +634,29 @@ jobs:
           df -h
       - name: Checkout
         uses: actions/checkout@v4
-      - name: Set up QEMU
-        uses: docker/setup-qemu-action@v3
-      - name: Build image
-        uses: docker/build-push-action@v6
-        timeout-minutes: 30
+      - name: Install Rust
+        uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
-          context: .
-          file: ./packages/cubejs-docker/${{ matrix.dockerfile }}
-          platforms: linux/amd64
-          push: true
-          tags: localhost:5000/cubejs/cube:${{ matrix.tag }}
-      - name: Use Node.js 20.x
+          toolchain: nightly-2024-07-15
+          # override: true # this is by default on
+          rustflags: ""
+          components: rustfmt
+          target: ${{ matrix.target }}
+      - name: Install Node.js ${{ matrix.node-version }}
         uses: actions/setup-node@v4
         with:
-          node-version: 20.x
+          node-version: ${{ matrix.node-version }}
+      - name: Install Yarn
+        run: npm install -g yarn
+      - name: Set Yarn version
+        run: yarn policies set-version v1.22.22
+      - name: Install cargo-cp-artifact
+        run: npm install -g [email protected]
+      - uses: Swatinem/rust-cache@v2
+        with:
+          workspaces: ./packages/cubejs-backend-native
+          key: native-${{ runner.OS }}-${{ matrix.target }}
+          shared-key: native-${{ runner.OS }}-${{ matrix.target }}
       - name: Get yarn cache directory path
         id: yarn-cache-dir-path
         run: echo "dir=$(yarn cache dir)" >> "$GITHUB_OUTPUT"
@@ -591,8 +668,6 @@ jobs:
           key: ${{ runner.os }}-yarn-${{ hashFiles('**/yarn.lock') }}
           restore-keys: |
             ${{ runner.os }}-yarn-
-      - name: Set Yarn version
-        run: yarn policies set-version v1.22.22
       - name: Yarn install
         uses: nick-fields/retry@v3
         env:
@@ -607,6 +682,24 @@ jobs:
         run: yarn build
       - name: Lerna tsc
         run: yarn tsc
+      - name: Build native (no python)
+        run: cd packages/cubejs-backend-native && npm run native:build-release
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v3
+      # current .dockerignore prevents use of native build
+      - name: Unignore native from .dockerignore
+        run: |
+          grep -v -E "packages/cubejs-backend-native/((native)|(index.node))" .dockerignore > .dockerignore.tmp
+          mv .dockerignore.tmp .dockerignore
+      - name: Build image
+        uses: docker/build-push-action@v6
+        timeout-minutes: 30
+        with:
+          context: .
+          file: ./packages/cubejs-docker/${{ matrix.dockerfile }}
+          platforms: linux/amd64
+          push: true
+          tags: localhost:5000/cubejs/cube:${{ matrix.tag }}
       - name: Testing CubeJS (container mode) via BirdBox
         run: |
           cd packages/cubejs-testing/

.github/workflows/rust-cubestore-master.yml

@@ -151,7 +151,7 @@ jobs:
       - name: Push to Docker Hub
         uses: docker/build-push-action@v6
         with:
-          context: ./rust/cubestore
+          context: ./rust
           file: ./rust/cubestore/Dockerfile
           platforms: ${{ matrix.platforms }}
           build-args: ${{ matrix.build-args }}

.github/workflows/rust-cubestore.yml

@@ -99,7 +99,7 @@ jobs:
       - name: Build only
         uses: docker/build-push-action@v6
         with:
-          context: ./rust/cubestore/
+          context: ./rust/
           file: ./rust/cubestore/Dockerfile
           platforms: ${{ matrix.platforms }}
           build-args: ${{ matrix.build-args }}

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

@@ -19,7 +19,7 @@ const calcMessageLength = (message: unknown) => Buffer.byteLength(
   typeof message === 'string' ? message : JSON.stringify(message)
 );
 
-export type WebSocketSendMessageFn = (connectionId: string, message: any) => void;
+export type WebSocketSendMessageFn = (connectionId: string, message: any) => Promise<void>;
 
 export class SubscriptionServer {
   public constructor(
@@ -31,7 +31,7 @@ export class SubscriptionServer {
   }
 
   public resultFn(connectionId: string, messageId: string, requestId: string | undefined) {
-    return (message, { status } = { status: 200 }) => {
+    return async (message, { status } = { status: 200 }) => {
       this.apiGateway.log({
         type: 'Outgoing network usage',
         service: 'api-ws',