adjust some local shuffle rules

Author: BiteTheDDDDt

be/src/pipeline/exec/distinct_streaming_aggregation_operator.h

@@ -118,7 +118,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; }