[Fix](memtable) fix shrink_memtable_by_agg should also update `_row…

be/src/olap/memtable.cpp

@@ -451,6 +451,8 @@ void MemTable::_aggregate() {
         _output_mutable_block =
                 vectorized::MutableBlock::build_mutable_block(empty_input_block.get());
         _output_mutable_block.clear_column_data();
+        _row_in_blocks = temp_row_in_blocks;
+        _last_sorted_pos = _row_in_blocks.size();
     }
 }
 

regression-test/suites/load_p0/insert/test_insert_with_aggregation_memtable.groovy

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-suite("test_insert_with_aggregation_memtable") {
+suite("test_insert_with_aggregation_memtable", "nonConcurrent") {
     def backendId_to_backendIP = [:]
     def backendId_to_backendHttpPort = [:]
     def backendId_to_params = [string:[:]]
@@ -114,6 +114,26 @@ suite("test_insert_with_aggregation_memtable") {
     sql "sync"
     qt_sql "select * from ${testTable} order by id asc"
 
+    // test with mv
+    def table_name = "agg_shrink"
+    sql "DROP TABLE IF EXISTS ${table_name}"
+    sql """
+        CREATE TABLE IF NOT EXISTS ${table_name} (
+            k bigint,
+            v text 
+        )
+        DUPLICATE KEY(`k`)
+        DISTRIBUTED BY HASH(k) BUCKETS 4
+        properties("replication_num" = "1");
+    """
+    set_be_param("write_buffer_size_for_agg", "10240") // change it to 10KB
+    sql """INSERT INTO ${table_name} SELECT *, '{"k1":1, "k2": "hello world", "k3" : [1234], "k4" : 1.10000, "k5" : [[123]]}' FROM numbers("number" = "4096")"""
+    sql """INSERT INTO ${table_name} SELECT k, v from ${table_name}"""
+    sql """INSERT INTO ${table_name} SELECT k, v from ${table_name}"""
+    createMV("""create materialized view var_cnt as select k, count(k) from ${table_name} group by k""")    
+    sql """INSERT INTO ${table_name} SELECT k, v from ${table_name} limit 8101"""
+
     reset_be_param("enable_shrink_memory")
     reset_be_param("write_buffer_size_for_agg")
+
 }