[BugFix] Fix consistency problem

StarRocks · Jul 19, 2022 · d7ba635 · d7ba635
1 parent 7ee482c
commit d7ba635
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 14 deletions.
diff --git a/be/src/exec/pipeline/fragment_executor.cpp b/be/src/exec/pipeline/fragment_executor.cpp
@@ -546,12 +546,7 @@ Status FragmentExecutor::_decompose_data_sink_to_operator(RuntimeState* runtime_
         if (sender->get_partition_type() == TPartitionType::HASH_PARTITIONED ||
             sender->get_partition_type() == TPartitionType::BUCKET_SHUFFLE_HASH_PARTITIONED) {
             dest_dop = t_stream_sink.dest_dop;
-
-            // UNPARTITIONED mode will be performed if both num of destination and dest dop is 1
-            // So we only enable pipeline level shuffle when num of destination or dest dop is greater than 1
-            if (sender->destinations().size() > 1 || dest_dop > 1) {
-                is_pipeline_level_shuffle = true;
-            }
+            is_pipeline_level_shuffle = true;
             DCHECK_GT(dest_dop, 0);
         }
 

diff --git a/be/src/runtime/data_stream_recvr.cc b/be/src/runtime/data_stream_recvr.cc
@@ -150,7 +150,9 @@ class DataStreamRecvr::SenderQueue {
 
     typedef std::list<ChunkItem> ChunkQueue;
     ChunkQueue _chunk_queue;
+#ifdef DEBUG
     bool _is_pipeline_level_shuffle_init = false;
+#endif
     bool _is_pipeline_level_shuffle = false;
     std::vector<bool> _has_chunks_per_driver_sequence;
     serde::ProtobufChunkMeta _chunk_meta;
@@ -471,12 +473,21 @@ Status DataStreamRecvr::SenderQueue::add_chunks(const PTransmitChunkParams& requ
     ChunkQueue chunks;
     size_t total_chunk_bytes = 0;
     faststring uncompressed_buffer;
-    bool prev_is_pipeline_level_shuffle = _is_pipeline_level_shuffle;
+
+#ifdef DEBUG
+    {
+        std::lock_guard<std::mutex> l(_lock);
+        bool prev_is_pipeline_level_shuffle = _is_pipeline_level_shuffle;
+        _is_pipeline_level_shuffle =
+                _recvr->_is_pipeline && request.has_is_pipeline_level_shuffle() && request.is_pipeline_level_shuffle();
+        // _is_pipeline_level_shuffle must be stable after first assignment
+        DCHECK(!_is_pipeline_level_shuffle_init || (prev_is_pipeline_level_shuffle == _is_pipeline_level_shuffle));
+        _is_pipeline_level_shuffle_init = true;
+    }
+#else
     _is_pipeline_level_shuffle =
             _recvr->_is_pipeline && request.has_is_pipeline_level_shuffle() && request.is_pipeline_level_shuffle();
-    // _is_pipeline_level_shuffle must be stable after first assignment
-    DCHECK(!_is_pipeline_level_shuffle_init || (prev_is_pipeline_level_shuffle == _is_pipeline_level_shuffle));
-    _is_pipeline_level_shuffle_init = true;
+#endif
 
     if (use_pass_through) {
         ChunkUniquePtrVector swap_chunks;
@@ -604,12 +615,20 @@ Status DataStreamRecvr::SenderQueue::add_chunks_and_keep_order(const PTransmitCh
     size_t total_chunk_bytes = 0;
     faststring uncompressed_buffer;
     ChunkQueue local_chunk_queue;
-    bool prev_is_pipeline_level_shuffle = _is_pipeline_level_shuffle;
+#ifdef DEBUG
+    {
+        std::lock_guard<std::mutex> l(_lock);
+        bool prev_is_pipeline_level_shuffle = _is_pipeline_level_shuffle;
+        _is_pipeline_level_shuffle =
+                _recvr->_is_pipeline && request.has_is_pipeline_level_shuffle() && request.is_pipeline_level_shuffle();
+        // _is_pipeline_level_shuffle must be stable after first assignment
+        DCHECK(!_is_pipeline_level_shuffle_init || (prev_is_pipeline_level_shuffle == _is_pipeline_level_shuffle));
+        _is_pipeline_level_shuffle_init = true;
+    }
+#else
     _is_pipeline_level_shuffle =
             _recvr->_is_pipeline && request.has_is_pipeline_level_shuffle() && request.is_pipeline_level_shuffle();
-    // _is_pipeline_level_shuffle must be stable after first assignment
-    DCHECK(!_is_pipeline_level_shuffle_init || (prev_is_pipeline_level_shuffle == _is_pipeline_level_shuffle));
-    _is_pipeline_level_shuffle_init = true;
+#endif
 
     if (use_pass_through) {
         ChunkUniquePtrVector swap_chunks;