diff --git a/be/src/pipeline/exec/exchange_sink_operator.cpp b/be/src/pipeline/exec/exchange_sink_operator.cpp
index 04b9653e9c82996..65d1a5d2fb83ae2 100644
--- a/be/src/pipeline/exec/exchange_sink_operator.cpp
+++ b/be/src/pipeline/exec/exchange_sink_operator.cpp
@@ -327,7 +327,8 @@ ExchangeSinkOperatorX::ExchangeSinkOperatorX(
           _tablet_sink_txn_id(sink.tablet_sink_txn_id),
           _t_tablet_sink_exprs(&sink.tablet_sink_exprs),
           _enable_local_merge_sort(state->enable_local_merge_sort()),
-          _fragment_instance_ids(fragment_instance_ids) {
+          _fragment_instance_ids(fragment_instance_ids),
+          _dest_is_merge(sink.__isset.is_merge && sink.is_merge) {
     DCHECK_GT(destinations.size(), 0);
     DCHECK(sink.output_partition.type == TPartitionType::UNPARTITIONED ||
            sink.output_partition.type == TPartitionType::HASH_PARTITIONED ||
@@ -772,8 +773,10 @@ std::shared_ptr<ExchangeSinkBuffer> ExchangeSinkOperatorX::get_sink_buffer(
     // Therefore, sharing a sink buffer is not necessary.
     if (std::dynamic_pointer_cast<SortSourceOperatorX>(_child) ||
         std::dynamic_pointer_cast<LocalExchangeSourceOperatorX>(_child)) {
+        DCHECK_EQ(_dest_is_merge, true);
         return _create_buffer({sender_ins_id});
     }
+    DCHECK_EQ(_dest_is_merge, false);
     if (_state->enable_shared_exchange_sink_buffer()) {
         return _sink_buffer;
     }
diff --git a/be/src/pipeline/exec/exchange_sink_operator.h b/be/src/pipeline/exec/exchange_sink_operator.h
index 8d094b43f613f0d..79f16065039b7b4 100644
--- a/be/src/pipeline/exec/exchange_sink_operator.h
+++ b/be/src/pipeline/exec/exchange_sink_operator.h
@@ -294,6 +294,7 @@ class ExchangeSinkOperatorX final : public DataSinkOperatorX<ExchangeSinkLocalSt
     int _writer_count = 1;
     const bool _enable_local_merge_sort;
     const std::vector<TUniqueId>& _fragment_instance_ids;
+    const bool _dest_is_merge;
 };
 
 } // namespace pipeline