fix the problem of multishard incorrect result on distinct (#831)

* fix the problem of multishard incorrect result on distinct

* code revise

Author: Jasmine-ge

src/AggregateFunctions/Streaming/AggregateFunctionDistinct.h

@@ -107,9 +107,18 @@ struct AggregateFunctionDistinctSingleNumericData
         for (auto it = extra_data_since_last_finalize.begin(); it != extra_data_since_last_finalize.end();)
         {
             if (rhs.set.find(*it) != rhs.set.end())
-                it = extra_data_since_last_finalize.erase(it);
+            {
+                bool is_new_data = std::find(rhs.extra_data_since_last_finalize.begin(), rhs.extra_data_since_last_finalize.end(), *it)
+                    != rhs.extra_data_since_last_finalize.end();
+                if (is_new_data)
+                    ++it;
+                else
+                    it = extra_data_since_last_finalize.erase(it);
+            }
             else
+            {
                 ++it;
+            }
         }
 
         /// Merge and deduplicate rhs' extra data
@@ -176,9 +185,18 @@ struct AggregateFunctionDistinctGenericData
         for (auto next = extra_data_since_last_finalize.begin(); next != extra_data_since_last_finalize.end();)
         {
             if (rhs.set.find(*next) != rhs.set.end())
-                next = extra_data_since_last_finalize.erase(next);
+            {
+                bool is_new_data = std::find(rhs.extra_data_since_last_finalize.begin(), rhs.extra_data_since_last_finalize.end(), *next)
+                    != rhs.extra_data_since_last_finalize.end();
+                if (is_new_data)
+                    ++next;
+                else
+                    next = extra_data_since_last_finalize.erase(next);
+            }
             else
+            {
                 ++next;
+            }
         }
 
         /// Merge and deduplicate rhs' extra data

tests/stream/test_stream_smoke/0012_multishards7.yaml

@@ -606,14 +606,14 @@ tests:
             query_type: table
             depends_on: '1366'
             wait: 1
-            query: insert into test13_multishard_7(id, val) values(1, 30), (2, 40), (3, 60)
+            query: insert into test13_multishard_7(id, val) values(1, 40), (1, 30)
 
           - client: python
             query_type: table
             kill: '1366'
             kill_wait: 2
             wait: 1
-            query: insert into test13_multishard_7(id, val) values(1, 30), (1, 40), (1, 60), (2, 40), (3, 40)
+            query: insert into test13_multishard_7(id, val) values(1, 30), (1, 50), (2, 50), (3, 50)
 
           - client: python
             query_type: table
@@ -624,5 +624,5 @@ tests:
     expected_results:
       - query_id: '1366'
         expected_results:
-          - [130, 130]
-          - [130, 340]
+          - [70, 70]
+          - [120, 250]