[fix](join) Fix partitioned hash join strategy (#40164)

## Proposed changes

pick #40163

<!--Describe your changes.-->

be/src/pipeline/pipeline_x/pipeline_x_fragment_context.cpp

@@ -1161,6 +1161,8 @@ Status PipelineXFragmentContext::_create_operator(ObjectPool* pool, const TPlanN
 
             _pipeline_parent_map.push(op->node_id(), cur_pipe);
             _pipeline_parent_map.push(op->node_id(), build_side_pipe);
+            sink->set_followed_by_shuffled_join(sink->is_shuffled_hash_join());
+            op->set_followed_by_shuffled_join(op->is_shuffled_hash_join());
         } else {
             op.reset(new HashJoinProbeOperatorX(pool, tnode, next_operator_id(), descs));
             RETURN_IF_ERROR(cur_pipe->add_operator(op));
@@ -1181,6 +1183,8 @@ Status PipelineXFragmentContext::_create_operator(ObjectPool* pool, const TPlanN
 
             _pipeline_parent_map.push(op->node_id(), cur_pipe);
             _pipeline_parent_map.push(op->node_id(), build_side_pipe);
+            sink->set_followed_by_shuffled_join(sink->is_shuffled_hash_join());
+            op->set_followed_by_shuffled_join(op->is_shuffled_hash_join());
         }
         _require_bucket_distribution =
                 _require_bucket_distribution || op->require_data_distribution();