[Chore](shuffle) adjust some local shuffle rules (#59366)

1. make distinct streaming agg always shuffle(Improve parallelism)
2. make broadcast join probe do not shuffle

This pull request updates the logic for determining required data
distributions in the aggregation and join pipeline operators. The main
focus is on improving the handling of exchange types, especially for
passthrough and broadcast scenarios.

Key changes include:

**Data distribution logic updates:**

* In `DistinctStreamingAggOperatorX`, the method now always returns
`ExchangeType::PASSTHROUGH` instead of delegating to the base class,
simplifying the distribution requirement when colocation is not needed.
* In `HashJoinProbeOperatorX`, the logic for broadcast joins is refined:
if the child is a serial operator, it returns
`ExchangeType::PASSTHROUGH`; otherwise, it returns `ExchangeType::NOOP`.
The handling of bucket shuffle and colocate join distributions is also
clarified.

Author: BiteTheDDDDt

be/src/pipeline/exec/distinct_streaming_aggregation_operator.h

@@ -120,7 +120,7 @@ class DistinctStreamingAggOperatorX final
                            ? DataDistribution(ExchangeType::BUCKET_HASH_SHUFFLE, _partition_exprs)
                            : DataDistribution(ExchangeType::HASH_SHUFFLE, _partition_exprs);
         }
-        return StatefulOperatorX<DistinctStreamingAggLocalState>::required_data_distribution(state);
+        return {ExchangeType::PASSTHROUGH};
     }
 
     bool require_data_distribution() const override { return _is_colocate; }

be/src/pipeline/exec/hashjoin_probe_operator.h

@@ -133,14 +133,16 @@ class HashJoinProbeOperatorX MOCK_REMOVE(final)
     DataDistribution required_data_distribution(RuntimeState* /*state*/) const override {
         if (_join_op == TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN) {
             return {ExchangeType::NOOP};
+        } else if (_is_broadcast_join) {
+            return _child && _child->is_serial_operator()
+                           ? DataDistribution(ExchangeType::PASSTHROUGH)
+                           : DataDistribution(ExchangeType::NOOP);
         }
-        return _is_broadcast_join
-                       ? DataDistribution(ExchangeType::PASSTHROUGH)
-                       : (_join_distribution == TJoinDistributionType::BUCKET_SHUFFLE ||
-                                          _join_distribution == TJoinDistributionType::COLOCATE
-                                  ? DataDistribution(ExchangeType::BUCKET_HASH_SHUFFLE,
-                                                     _partition_exprs)
-                                  : DataDistribution(ExchangeType::HASH_SHUFFLE, _partition_exprs));
+
+        return (_join_distribution == TJoinDistributionType::BUCKET_SHUFFLE ||
+                                _join_distribution == TJoinDistributionType::COLOCATE
+                        ? DataDistribution(ExchangeType::BUCKET_HASH_SHUFFLE, _partition_exprs)
+                        : DataDistribution(ExchangeType::HASH_SHUFFLE, _partition_exprs));
     }
     bool is_broadcast_join() const { return _is_broadcast_join; }