fix(cubestore): Merge for streaming union (#5554)

cube-js · Nov 7, 2022 · 310649a · 310649a
1 parent 80cfaba
commit 310649a
Show file tree

Hide file tree

Showing 2 changed files with 64 additions and 1 deletion.
diff --git a/rust/cubestore/Cargo.lock b/rust/cubestore/Cargo.lock
diff --git a/rust/cubestore/cubestore-sql-tests/src/tests.rs b/rust/cubestore/cubestore-sql-tests/src/tests.rs
@@ -194,6 +194,10 @@ pub fn sql_tests() -> Vec<(&'static str, TestFn)> {
             "unique_key_and_multi_partitions",
             unique_key_and_multi_partitions,
         ),
+        t(
+            "unique_key_and_multi_partitions_hash_aggregate",
+            unique_key_and_multi_partitions_hash_aggregate,
+        ),
         t("divide_by_zero", divide_by_zero),
         t(
             "filter_multiple_in_for_decimal",
@@ -5379,6 +5383,65 @@ async fn unique_key_and_multi_partitions(service: Box<dyn SqlClient>) {
     );
 }
 
+async fn unique_key_and_multi_partitions_hash_aggregate(service: Box<dyn SqlClient>) {
+    service.exec_query("CREATE SCHEMA test").await.unwrap();
+    service.exec_query("CREATE TABLE test.unique_parts1 (a int, b int, c int, e int, val int) unique key (a, b, c, e) ").await.unwrap();
+    service.exec_query("CREATE TABLE test.unique_parts2 (a int, b int, c int, e int, val int) unique key (a, b, c, e) ").await.unwrap();
+    service
+        .exec_query(
+            "
+            INSERT INTO test.unique_parts1
+            (a, b, c, e, val, __seq)
+            VALUES
+            (1, 1, 1, 1, 10, 1),
+            (2, 2, 2, 2, 20, 2)
+            "
+        )
+        .await
+        .unwrap();
+
+    service
+        .exec_query(
+            "
+            INSERT INTO test.unique_parts1
+            (a, b, c, e, val, __seq)
+            VALUES
+            (11, 11, 1, 11, 110, 11),
+            (22, 22, 2, 22, 220, 21)
+            "
+        )
+        .await
+        .unwrap();
+
+    service
+        .exec_query(
+            "
+            INSERT INTO test.unique_parts2
+            (a, b, c, e, val, __seq)
+            VALUES
+            (3, 3, 1, 3, 30, 3),
+            (4, 4, 1, 4, 40, 4)
+            "
+        )
+        .await
+        .unwrap();
+
+    let r = service
+        .exec_query(
+            "SELECT c, sum(val) FROM (
+                    SELECT * FROM test.unique_parts1
+                    UNION ALL 
+                    SELECT * FROM test.unique_parts2
+                ) `tt` GROUP BY 1 ORDER BY 1 LIMIT 100",
+        )
+        .await
+        .unwrap();
+    assert_eq!(
+        to_rows(&r),
+        rows(&[(1, 190), (2, 240)])
+    );
+}
+
 async fn divide_by_zero(service: Box<dyn SqlClient>) {
     service.exec_query("CREATE SCHEMA s").await.unwrap();
     service