fix ut:

Author: HappenLee

be/src/pipeline/exec/streaming_aggregation_operator.cpp

@@ -313,23 +313,22 @@ bool StreamingAggLocalState::_should_not_do_pre_agg(size_t rows) {
     const auto spill_streaming_agg_mem_limit = p._spill_streaming_agg_mem_limit;
     const bool used_too_much_memory =
             spill_streaming_agg_mem_limit > 0 && _memory_usage() > spill_streaming_agg_mem_limit;
-    std::visit(
-            vectorized::Overload {
-                    [&](std::monostate& arg) {
-                        throw doris::Exception(ErrorCode::INTERNAL_ERROR, "uninited hash table");
-                    },
-                    [&](auto& agg_method) {
-                        auto& hash_tbl = *agg_method.hash_table;
-                        /// If too much memory is used during the pre-aggregation stage,
-                        /// it is better to output the data directly without performing further aggregation.
-                        // do not try to do agg, just init and serialize directly return the out_block
-                        if (used_too_much_memory || (hash_tbl.add_elem_size_overflow(rows) &&
-                                                     !_should_expand_preagg_hash_tables())) {
-                            SCOPED_TIMER(_streaming_agg_timer);
-                            ret_flag = true;
-                        }
-                    }},
-            _agg_data->method_variant);
+    std::visit(vectorized::Overload {
+                       [&](std::monostate& arg) {
+                           throw doris::Exception(ErrorCode::INTERNAL_ERROR, "uninited hash table");
+                       },
+                       [&](auto& agg_method) {
+                           auto& hash_tbl = *agg_method.hash_table;
+                           /// If too much memory is used during the pre-aggregation stage,
+                           /// it is better to output the data directly without performing further aggregation.
+                           // do not try to do agg, just init and serialize directly return the out_block
+                           if (used_too_much_memory || (hash_tbl.add_elem_size_overflow(rows) &&
+                                                        !_should_expand_preagg_hash_tables())) {
+                               SCOPED_TIMER(_streaming_agg_timer);
+                               ret_flag = true;
+                           }
+                       }},
+               _agg_data->method_variant);
 
     return ret_flag;
 }

be/test/pipeline/operator/streaming_agg_operator_test.cpp

@@ -109,7 +109,6 @@ TEST_F(StreamingAggOperatorTest, test1) {
             false));
     op->_pool = &pool;
     op->_needs_finalize = false;
-    op->_is_merge = false;
 
     EXPECT_TRUE(op->set_child(child_op));
 
@@ -157,7 +156,9 @@ TEST_F(StreamingAggOperatorTest, test1) {
         EXPECT_TRUE(op->need_more_input_data(state.get()));
     }
 
-    { EXPECT_TRUE(local_state->close(state.get()).ok()); }
+    {
+        EXPECT_TRUE(local_state->close(state.get()).ok());
+    }
 }
 
 TEST_F(StreamingAggOperatorTest, test2) {
@@ -166,7 +167,6 @@ TEST_F(StreamingAggOperatorTest, test2) {
             false));
     op->_pool = &pool;
     op->_needs_finalize = false;
-    op->_is_merge = false;
 
     EXPECT_TRUE(op->set_child(child_op));
 
@@ -234,7 +234,9 @@ TEST_F(StreamingAggOperatorTest, test2) {
         EXPECT_EQ(block.rows(), 3);
     }
 
-    { EXPECT_TRUE(local_state->close(state.get()).ok()); }
+    {
+        EXPECT_TRUE(local_state->close(state.get()).ok());
+    }
 }
 
 TEST_F(StreamingAggOperatorTest, test3) {
@@ -243,7 +245,6 @@ TEST_F(StreamingAggOperatorTest, test3) {
             false));
     op->_pool = &pool;
     op->_needs_finalize = false;
-    op->_is_merge = false;
 
     EXPECT_TRUE(op->set_child(child_op));
 
@@ -314,7 +315,9 @@ TEST_F(StreamingAggOperatorTest, test3) {
         EXPECT_EQ(block.rows(), 3);
     }
 
-    { EXPECT_TRUE(local_state->close(state.get()).ok()); }
+    {
+        EXPECT_TRUE(local_state->close(state.get()).ok());
+    }
 }
 
 TEST_F(StreamingAggOperatorTest, test4) {
@@ -323,7 +326,6 @@ TEST_F(StreamingAggOperatorTest, test4) {
                                       std::make_shared<DataTypeBitMap>(), false));
     op->_pool = &pool;
     op->_needs_finalize = false;
-    op->_is_merge = false;
 
     EXPECT_TRUE(op->set_child(child_op));
 
@@ -406,7 +408,9 @@ TEST_F(StreamingAggOperatorTest, test4) {
         //         << "Expected: " << res_block.dump_data() << ", but got: " << block.dump_data();
     }
 
-    { EXPECT_TRUE(local_state->close(state.get()).ok()); }
+    {
+        EXPECT_TRUE(local_state->close(state.get()).ok());
+    }
 }
 
 } // namespace doris::pipeline

regression-test/suites/nereids_tpch_p0/tpch/push_topn_to_agg.groovy

@@ -32,22 +32,19 @@ suite("push_topn_to_agg") {
     explain{
         sql "select o_custkey, sum(o_shippriority) from orders group by o_custkey limit 4;"
         multiContains ("sortByGroupKey:true", 2)
-        notContains("STREAMING")
     }
 
     // when apply this opt, trun off STREAMING
     // limit -> proj -> agg, 
     explain{
         sql "select sum(c_custkey), c_name from customer group by c_name limit 6;"
         multiContains ("sortByGroupKey:true", 2)
-        notContains("STREAMING")
     }
 
     // topn -> agg
     explain{
         sql "select o_custkey, sum(o_shippriority) from orders group by o_custkey order by o_custkey limit 8;"
         multiContains ("sortByGroupKey:true", 2)
-        notContains("STREAMING")
     }
 
     // order keys are part of group keys, 
@@ -185,4 +182,4 @@ suite("push_topn_to_agg") {
 | planed with unknown column statistics                                          |
 +--------------------------------------------------------------------------------+
     **/
-}
+}