feat(cubesql): Streams - cancel query and drop conection handling

Author: gandronchik

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

@@ -100,8 +100,15 @@ function wrapNativeFunctionWithChannelCallback(
                     e
                 });
             }
-
-            channel.reject(e.message || 'Unknown JS exception');
+            try {
+                channel.reject(e.message || 'Unknown JS exception');
+            } catch(e) {
+                if (process.env.CUBEJS_NATIVE_INTERNAL_DEBUG) {
+                    console.debug("[js] channel.reject exception", {
+                        e
+                    });
+                }
+            }
 
             // throw e;
           }
@@ -121,7 +128,9 @@ function wrapNativeFunctionWithStream(
                 chunk.push(c);
                 if (chunk.length >= 10000) {
                     if (!writer.chunk(JSON.stringify(chunk))) {
-                        streamResponse.stream.removeAllListeners();
+                        streamResponse.stream.destroy({
+                            stack: "Rejected by client"
+                        });
                     }
                     chunk = [];
                 }

rust/cubesql/cubesql/src/compile/engine/df/scan.rs

@@ -407,6 +407,7 @@ struct CubeScanMemoryStream {
     /// Schema representing the data
     schema: SchemaRef,
     member_fields: Vec<MemberField>,
+    is_completed: bool,
 }
 
 impl CubeScanMemoryStream {
@@ -419,17 +420,33 @@ impl CubeScanMemoryStream {
             stream,
             schema,
             member_fields,
+            is_completed: false,
         }
     }
 
     fn poll_next(&mut self) -> Option<ArrowResult<RecordBatch>> {
         match self.stream.poll_next() {
-            Ok(chunk) => parse_chunk(chunk, self.schema.clone(), &self.member_fields),
+            Ok(chunk) => {
+                let chunk = parse_chunk(chunk, self.schema.clone(), &self.member_fields);
+                if chunk.is_none() {
+                    self.is_completed = true;
+                }
+
+                chunk
+            }
             Err(err) => Some(Err(ArrowError::ComputeError(err.to_string()))),
         }
     }
 }
 
+impl Drop for CubeScanMemoryStream {
+    fn drop(&mut self) {
+        if !self.is_completed {
+            self.stream.reject();
+        }
+    }
+}
+
 struct CubeScanStreamRouter {
     main_stream: CubeScanMemoryStream,
     one_shot_stream: CubeScanOneShotStream,

rust/cubesql/cubesql/src/transport/service.rs

@@ -81,9 +81,7 @@ impl CubeReadStream for CubeDummyStream {
         panic!("CubeDummyStream.poll_next - can not be called");
     }
 
-    fn reject(&self) {
-        panic!("CubeDummyStream.reject - can not be called");
-    }
+    fn reject(&self) {}
 }
 
 #[derive(Debug)]