feat: streaming desync (#6034)

Author: buntarb

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

@@ -119,14 +119,18 @@ function wrapNativeFunctionWithChannelCallback(
 function wrapNativeFunctionWithStream(
     fn: (extra: any) => unknown | Promise<unknown>
 ) {
+    const chunkLength = parseInt(
+        process.env.CUBEJS_DB_QUERY_STREAM_HIGH_WATER_MARK || '8192',
+        10
+    );
     return async (extra: any, writer: any) => {
         let streamResponse: any;
         try {
             streamResponse = await fn(JSON.parse(extra));
             let chunk: object[] = [];
             streamResponse.stream.on('data', (c: object) => {
                 chunk.push(c);
-                if (chunk.length >= 10000) {
+                if (chunk.length >= chunkLength) {
                     if (!writer.chunk(JSON.stringify(chunk))) {
                         streamResponse.stream.destroy({
                             stack: "Rejected by client"

packages/cubejs-backend-shared/src/env.ts

@@ -568,7 +568,7 @@ const variables: Record<string, (...args: any) => any> = {
    * Query stream `highWaterMark` value.
    */
   dbQueryStreamHighWaterMark: (): number => get('CUBEJS_DB_QUERY_STREAM_HIGH_WATER_MARK')
-    .default(5000)
+    .default(8192)
     .asInt(),
 
   /**

packages/cubejs-query-orchestrator/src/orchestrator/QueryStream.ts

@@ -2,6 +2,8 @@ import * as stream from 'stream';
 import { getEnv } from '@cubejs-backend/shared';
 
 export class QueryStream extends stream.Transform {
+  private counter = 0;
+
   public queryKey: string;
 
   public maps: {
@@ -47,7 +49,17 @@ export class QueryStream extends stream.Transform {
     Object.keys(chunk).forEach((alias) => {
       row[this.aliasNameToMember[alias]] = chunk[alias];
     });
-    callback(null, row);
+    if (this.counter < this.writableHighWaterMark) {
+      this.counter++;
+      callback(null, row);
+    } else {
+      this.pause();
+      setTimeout(() => {
+        this.resume();
+        this.counter = 0;
+        callback(null, row);
+      }, 0);
+    }
   }
 
   /**