update

Author: sollhui

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

@@ -97,6 +97,7 @@
 import org.apache.doris.thrift.PaloInternalServiceVersion;
 import org.apache.doris.thrift.TAIResource;
 import org.apache.doris.thrift.TBrokerScanRange;
+import org.apache.doris.thrift.TDataSink;
 import org.apache.doris.thrift.TDataSinkType;
 import org.apache.doris.thrift.TDescriptorTable;
 import org.apache.doris.thrift.TErrorTabletInfo;
@@ -565,12 +566,19 @@ private static void assignAdaptiveRandomBucketForFragment(
             if (partitionAssignments == null) {
                 continue;
             }
+            TOlapTableSink copiedSink = deepCopyOlapTableSinkForCurrentBackend(sinkParam);
             OlapTableSink.applyAdaptiveRandomBucketAssignments(
-                    sinkParam.getFragment().getOutputSink().getOlapTableSink().getPartition().getPartitions(),
+                    copiedSink.getPartition().getPartitions(),
                     partitionAssignments);
         }
     }
 
+    private static TOlapTableSink deepCopyOlapTableSinkForCurrentBackend(TPipelineFragmentParams sinkParam) {
+        TDataSink copiedOutputSink = sinkParam.getFragment().getOutputSink().deepCopy();
+        sinkParam.getFragment().setOutputSink(copiedOutputSink);
+        return copiedOutputSink.getOlapTableSink();
+    }
+
     // Initialize
     protected void prepare() throws UserException {
         for (PlanFragment fragment : fragments) {

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

@@ -295,12 +295,19 @@ private static void assignAdaptiveRandomBucketForSinkParams(List<TPipelineFragme
             if (partitionAssignments == null) {
                 continue;
             }
+            TOlapTableSink copiedSink = deepCopyOlapTableSinkForCurrentBackend(sinkParam);
             OlapTableSink.applyAdaptiveRandomBucketAssignments(
-                    sinkParam.getFragment().getOutputSink().getOlapTableSink().getPartition().getPartitions(),
+                    copiedSink.getPartition().getPartitions(),
                     partitionAssignments);
         }
     }
 
+    private static TOlapTableSink deepCopyOlapTableSinkForCurrentBackend(TPipelineFragmentParams sinkParam) {
+        TDataSink copiedOutputSink = sinkParam.getFragment().getOutputSink().deepCopy();
+        sinkParam.getFragment().setOutputSink(copiedOutputSink);
+        return copiedOutputSink.getOlapTableSink();
+    }
+
     private static Multiset<DistributedPlanWorker> computeInstanceNumPerWorker(
             List<PipelineDistributedPlan> distributedPlans) {
         Multiset<DistributedPlanWorker> workerCounter = LinkedHashMultiset.create();