[bug](partition_sort)partition sort need sort all data in two pahse g…

…lobal

be/src/vec/exec/vpartition_sort_node.cpp

@@ -64,6 +64,7 @@ Status VPartitionSortNode::init(const TPlanNode& tnode, RuntimeState* state) {
     _has_global_limit = tnode.partition_sort_node.has_global_limit;
     _top_n_algorithm = tnode.partition_sort_node.top_n_algorithm;
     _partition_inner_limit = tnode.partition_sort_node.partition_inner_limit;
+    _topn_phase = tnode.partition_sort_node.ptopn_phase;
     return Status::OK();
 }
 
@@ -181,7 +182,9 @@ Status VPartitionSortNode::sink(RuntimeState* state, vectorized::Block* input_bl
             _value_places[0]->append_whole_block(input_block, child(0)->row_desc());
         } else {
             //just simply use partition num to check
-            if (_num_partition > config::partition_topn_partition_threshold &&
+            //if is TWO_PAHSE_GLOBAL, must be sort all data thought partition num threshold have been exceeded.
+            if (_topn_phase != TPartTopNPhase::TWO_PAHSE_GLOBAL &&
+                _num_partition > config::partition_topn_partition_threshold &&
                 child_input_rows < 10000 * _num_partition) {
                 {
                     std::lock_guard<std::mutex> lock(_buffer_mutex);

be/src/vec/exec/vpartition_sort_node.h

@@ -435,6 +435,7 @@ class VPartitionSortNode : public ExecNode {
     std::queue<Block> _blocks_buffer;
     int64_t child_input_rows = 0;
     std::mutex _buffer_mutex;
+    TPartTopNPhase::type _topn_phase;
 
     RuntimeProfile::Counter* _build_timer;
     RuntimeProfile::Counter* _emplace_key_timer;