fix: allow empty values in reduce&test

GreptimeTeam · Sep 3, 2024 · 00c54ee · 00c54ee
1 parent 446314e
commit 00c54ee
Show file tree

Hide file tree

Showing 3 changed files with 145 additions and 0 deletions.
diff --git a/src/flow/src/compute/render/reduce.rs b/src/flow/src/compute/render/reduce.rs
@@ -983,6 +983,9 @@ impl AccumOutput {
 
     /// return (accums, output)
     fn into_accum_output(self) -> Result<(Vec<Value>, Vec<Value>), EvalError> {
+        if self.accum.is_empty() && self.output.is_empty() {
+            return Ok((vec![], vec![]));
+        }
         ensure!(
             !self.accum.is_empty() && self.accum.len() == self.output.len(),
             InternalSnafu {

diff --git a/tests/cases/standalone/common/flow/flow_basic.result b/tests/cases/standalone/common/flow/flow_basic.result
@@ -93,6 +93,105 @@ DROP TABLE out_num_cnt_basic;
 
 Affected Rows: 0
 
+-- test distinct
+CREATE TABLE distinct_basic (
+    number INT,
+    ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    PRIMARY KEY(number),
+    TIME INDEX(ts)
+);
+
+Affected Rows: 0
+
+CREATE FLOW test_distinct_basic
+SINK TO out_distinct_basic
+AS
+SELECT DISTINCT number as dis FROM distinct_basic;
+
+Affected Rows: 0
+
+-- TODO(discord9): confirm if it's necessary to flush flow here?
+-- because flush_flow result is at most 1
+admin flush_flow('test_distinct_basic');
+
++-----------------------------------------+
+| ADMIN flush_flow('test_distinct_basic') |
++-----------------------------------------+
+| 0                                       |
++-----------------------------------------+
+
+-- SQLNESS ARG restart=true
+INSERT INTO distinct_basic
+VALUES
+    (20, "2021-07-01 00:00:00.200"),
+    (22, "2021-07-01 00:00:00.600");
+
+Affected Rows: 2
+
+admin flush_flow('test_distinct_basic');
+
++-----------------------------------------+
+| ADMIN flush_flow('test_distinct_basic') |
++-----------------------------------------+
+| 1                                       |
++-----------------------------------------+
+
+SELECT dis FROM out_distinct_basic;
+
++-----+
+| dis |
++-----+
+| 20  |
+| 22  |
++-----+
+
+admin flush_flow('test_distinct_basic');
+
++-----------------------------------------+
+| ADMIN flush_flow('test_distinct_basic') |
++-----------------------------------------+
+| 0                                       |
++-----------------------------------------+
+
+INSERT INTO distinct_basic
+VALUES
+    (23,"2021-07-01 00:00:01.000"),
+    (24,"2021-07-01 00:00:01.500");
+
+Affected Rows: 2
+
+admin flush_flow('test_distinct_basic');
+
++-----------------------------------------+
+| ADMIN flush_flow('test_distinct_basic') |
++-----------------------------------------+
+| 1                                       |
++-----------------------------------------+
+
+-- note that this quote-unquote column is a column-name, **not** a aggregation expr, generated by datafusion
+SELECT dis FROM out_distinct_basic;
+
++-----+
+| dis |
++-----+
+| 20  |
+| 22  |
+| 23  |
+| 24  |
++-----+
+
+DROP FLOW test_distinct_basic;
+
+Affected Rows: 0
+
+DROP TABLE distinct_basic;
+
+Affected Rows: 0
+
+DROP TABLE out_distinct_basic;
+
+Affected Rows: 0
+
 -- test interprete interval
 CREATE TABLE numbers_input_basic (
     number INT,

diff --git a/tests/cases/standalone/common/flow/flow_basic.sql b/tests/cases/standalone/common/flow/flow_basic.sql
@@ -40,6 +40,49 @@ DROP FLOW test_numbers_basic;
 DROP TABLE numbers_input_basic;
 DROP TABLE out_num_cnt_basic;
 
+-- test distinct
+CREATE TABLE distinct_basic (
+    number INT,
+    ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    PRIMARY KEY(number),
+    TIME INDEX(ts)
+);
+
+CREATE FLOW test_distinct_basic
+SINK TO out_distinct_basic
+AS
+SELECT DISTINCT number as dis FROM distinct_basic;
+
+-- TODO(discord9): confirm if it's necessary to flush flow here?
+-- because flush_flow result is at most 1
+admin flush_flow('test_distinct_basic');
+
+-- SQLNESS ARG restart=true
+INSERT INTO distinct_basic
+VALUES
+    (20, "2021-07-01 00:00:00.200"),
+    (22, "2021-07-01 00:00:00.600");
+
+admin flush_flow('test_distinct_basic');
+
+SELECT dis FROM out_distinct_basic;
+
+admin flush_flow('test_distinct_basic');
+
+INSERT INTO distinct_basic
+VALUES
+    (23,"2021-07-01 00:00:01.000"),
+    (24,"2021-07-01 00:00:01.500");
+
+admin flush_flow('test_distinct_basic');
+
+-- note that this quote-unquote column is a column-name, **not** a aggregation expr, generated by datafusion
+SELECT dis FROM out_distinct_basic;
+
+DROP FLOW test_distinct_basic;
+DROP TABLE distinct_basic;
+DROP TABLE out_distinct_basic;
+
 -- test interprete interval
 
 CREATE TABLE numbers_input_basic (