[pipelineX](local shuffle) Fix bucket hash shuffle (apache#28202)

Author: Gabriel39

fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java

@@ -3596,6 +3596,7 @@ Map<TNetworkAddress, TPipelineFragmentParams> toTPipelineParams(int backendNum)
             }
 
             Map<TNetworkAddress, TPipelineFragmentParams> res = new HashMap();
+            Map<TNetworkAddress, Integer> instanceIdx = new HashMap();
             for (int i = 0; i < instanceExecParams.size(); ++i) {
                 final FInstanceExecParam instanceExecParam = instanceExecParams.get(i);
                 if (!res.containsKey(instanceExecParam.host)) {
@@ -3625,10 +3626,16 @@ Map<TNetworkAddress, TPipelineFragmentParams> toTPipelineParams(int backendNum)
                     params.setNumBuckets(fragment.getBucketNum());
                     res.put(instanceExecParam.host, params);
                     res.get(instanceExecParam.host).setBucketSeqToInstanceIdx(new HashMap<Integer, Integer>());
+                    instanceIdx.put(instanceExecParam.host, 0);
                 }
+                // Set each bucket belongs to which instance on this BE.
+                // This is used for LocalExchange(BUCKET_HASH_SHUFFLE).
+                int instanceId = instanceIdx.get(instanceExecParam.host);
                 for (int bucket : instanceExecParam.bucketSeqSet) {
-                    res.get(instanceExecParam.host).getBucketSeqToInstanceIdx().put(bucket, i);
+                    res.get(instanceExecParam.host).getBucketSeqToInstanceIdx().put(bucket, instanceId);
+
                 }
+                instanceIdx.replace(instanceExecParam.host, ++instanceId);
                 TPipelineFragmentParams params = res.get(instanceExecParam.host);
                 TPipelineInstanceParams localParams = new TPipelineInstanceParams();