[fix](profile) Fix execution profile leak when dispatch fails before profile registration (#68658)

### What problem does this PR solve?

Issue Number: N/A

Related PR: N/A

Problem Summary:

An ExecutionProfile is registered before query dispatch. If dispatch
fails before the first `updateProfile(false)` call, the corresponding
Profile has not been added to `profileIdToProfileMap`.

When cleanup is invoked, the previous `removeProfile` implementation
removed execution profiles only after finding a historical Profile
entry. If no such entry existed, the ExecutionProfile remained in
`queryIdToExecutionProfiles` and continued to retain query state in FE
memory. This was reproduced with a UDF JAR download failure: three
failed queries left three ExecutionProfiles and no historical Profile
entries.

Pass the current Profile to `removeProfile` and remove its
ExecutionProfiles directly, regardless of whether a historical entry
exists. Add tests covering a missing historical entry, an existing
entry, and multiple ExecutionProfiles.

### Release note

Fix FE memory growth caused by execution profiles retained when query
dispatch fails before profile registration.

fe/fe-core/src/main/java/org/apache/doris/common/profile/Profile.java

@@ -281,6 +281,33 @@ public class Profile {
         return this.executionProfiles;
     }
 
+    public synchronized void clearExecutionProfiles() {
+        ProfileManager.getInstance().removeProfile(this);
+        this.executionProfiles.clear();
+        summaryProfile.clearExecutionDetails();
+        // Clear failed-attempt rows before reusing the plan.
+        rowsProducedMap.clear();
+        resetActualRowCounts(physicalPlan);
+    }
+
+    private void resetActualRowCounts(Plan plan) {
+        if (plan == null) {
+            return;
+        }
+        ((AbstractPlan) plan).updateActualRowCount(-1L);
+        for (Plan child : plan.children()) {
+            resetActualRowCounts(child);
+        }
+    }
+
+    public synchronized void clearPlan() {
+        summaryProfile.clearPlanDetails();
+        physicalPlan = null;
+        physicalRelations.clear();
+        planNodeMap = Maps.newHashMap();
+        rowsProducedMap.clear();
+    }
+
     // This API will also add the profile to ProfileManager, so that we could get the profile from ProfileManager.
     // isFinished ONLY means the coordinator or stmt executor is finished.
     public synchronized void updateSummary(Map<String, String> summaryInfo, boolean isFinished,
@@ -314,7 +341,7 @@ public class Profile {
                 long durationThreshold = executionProfiles.isEmpty()
                                     ? autoProfileDurationMs : executionProfiles.size() * autoProfileDurationMs;
                 if (this.queryFinishTimestamp != Long.MAX_VALUE && durationMs < durationThreshold) {
-                    ProfileManager.getInstance().removeProfile(this.getId());
+                    ProfileManager.getInstance().removeProfile(this);
                     if (LOG.isDebugEnabled()) {
                         LOG.debug("Removed profile {} because it's costs {} is less than {}", this.getId(),
                                 durationMs, autoProfileDurationMs * this.executionProfiles.size());

fe/fe-core/src/main/java/org/apache/doris/common/profile/ProfileManager.java

@@ -132,14 +132,13 @@ public class ProfileManager extends MasterDaemon {
     final AtomicReference<ProfileLoadStatus> profileLoadStatus =
             new AtomicReference<>(ProfileLoadStatus.UNLOADED);
 
-    // only protect queryIdDeque; queryIdToProfileMap is concurrent, no need to protect
     private ReentrantReadWriteLock lock;
     private ReadLock readLock;
     private WriteLock writeLock;
 
     // profile id is long string for broker load
     // is TUniqueId for others.
-    final Map<String, ProfileElement> queryIdToProfileMap;
+    final Map<String, ProfileElement> profileIdToProfileMap;
     // Sometimes one Profile is related with multiple execution profiles(Broker-load), so that
     // execution profile's query id is not related with Profile's query id.
     final Map<TUniqueId, ExecutionProfile> queryIdToExecutionProfiles;
@@ -164,7 +163,7 @@ public class ProfileManager extends MasterDaemon {
         lock = new ReentrantReadWriteLock(true);
         readLock = lock.readLock();
         writeLock = lock.writeLock();
-        queryIdToProfileMap = Maps.newHashMap();
+        profileIdToProfileMap = Maps.newHashMap();
         queryIdToExecutionProfiles = Maps.newHashMap();
         fetchRealTimeProfileExecutor = ThreadPoolManager.newDaemonFixedThreadPool(
                 10, 100, "fetch-realtime-profile-pool", true);
@@ -218,23 +217,20 @@ public class ProfileManager extends MasterDaemon {
 
         writeLock.lock();
         try {
-            if (!queryIdToProfileMap.containsKey(profile.getId())) {
+            if (!profileIdToProfileMap.containsKey(profile.getId())) {
                 deleteOutdatedProfilesFromMemory(1);
             }
 
             ProfileElement element = createElement(profile);
-            // 'insert into' does have job_id, put all profiles key with query_id
-            String key = profile.getSummaryProfile().getProfileId();
-            // check when push in, which can ensure every element in the list has QUERY_ID column,
-            // so there is no need to check when remove element from list.
-            if (Strings.isNullOrEmpty(key)) {
+            // Queries use query IDs; Broker Load uses job IDs.
+            String profileId = profile.getId();
+            if (Strings.isNullOrEmpty(profileId)) {
                 LOG.warn("the key or value of Map is null, "
                         + "may be forget to insert 'QUERY_ID' or 'JOB_ID' column into infoStrings");
             }
 
-            // a profile may be updated multiple times in queryIdToProfileMap,
-            // and only needs to be inserted into the queryIdDeque for the first time.
-            queryIdToProfileMap.put(key, element);
+            // Replace the previous entry for this profile.
+            profileIdToProfileMap.put(profileId, element);
         } finally {
             writeLock.unlock();
         }
@@ -255,9 +251,9 @@ public class ProfileManager extends MasterDaemon {
         List<List<String>> result = Lists.newArrayList();
         readLock.lock();
         try {
-            PriorityQueue<ProfileElement> queueIdDeque = getProfileOrderByQueryFinishTimeDesc();
-            while (!queueIdDeque.isEmpty()) {
-                ProfileElement profileElement = queueIdDeque.poll();
+            PriorityQueue<ProfileElement> profileQueue = getProfileOrderByQueryFinishTimeDesc();
+            while (!profileQueue.isEmpty()) {
+                ProfileElement profileElement = profileQueue.poll();
                 List<String> row = Lists.newArrayList();
                 for (String str : columnNameList) {
                     row.add(getProfileInfoString(profileElement, str));
@@ -322,7 +318,7 @@ public class ProfileManager extends MasterDaemon {
     }
 
     private List<Future<TGetRealtimeExecStatusResponse>> createFetchRealTimeProfileTasks(String id, String reqType) {
-        // For query, id is queryId, for load, id is LoadLoadingTaskId
+        // Broker Load profile IDs are job IDs.
         class QueryIdAndAddress {
             public TUniqueId id;
             public TNetworkAddress beAddress;
@@ -450,8 +446,8 @@ public class ProfileManager extends MasterDaemon {
         return summary;
     }
 
-    public String getProfile(String id) {
-        List<Future<TGetRealtimeExecStatusResponse>> futures = createFetchRealTimeProfileTasks(id, "profile");
+    public String getProfile(String profileId) {
+        List<Future<TGetRealtimeExecStatusResponse>> futures = createFetchRealTimeProfileTasks(profileId, "profile");
         // beAddr of reportExecStatus of QeProcessorImpl is meaningless, so assign a dummy address
         // to avoid compile failing.
         TNetworkAddress dummyAddr = new TNetworkAddress();
@@ -462,17 +458,17 @@ public class ProfileManager extends MasterDaemon {
                     QeProcessorImpl.INSTANCE.reportExecStatus(resp.getReportExecStatusParams(), dummyAddr);
                 }
             } catch (Exception e) {
-                LOG.warn("Failed to get real-time profile, id {}, error: {}", id, e.getMessage(), e);
+                LOG.warn("Failed to get real-time profile, id {}, error: {}", profileId, e.getMessage(), e);
             }
         }
 
         if (!futures.isEmpty()) {
-            LOG.info("Get real-time exec status finished, id {}", id);
+            LOG.info("Get real-time exec status finished, id {}", profileId);
         }
 
         readLock.lock();
         try {
-            ProfileElement element = queryIdToProfileMap.get(id);
+            ProfileElement element = profileIdToProfileMap.get(profileId);
             if (element == null) {
                 return null;
             }
@@ -483,10 +479,10 @@ public class ProfileManager extends MasterDaemon {
         }
     }
 
-    public String getProfileBrief(String queryID) {
+    public String getProfileBrief(String profileId) {
         readLock.lock();
         try {
-            ProfileElement element = queryIdToProfileMap.get(queryID);
+            ProfileElement element = profileIdToProfileMap.get(profileId);
             if (element == null) {
                 return null;
             }
@@ -496,22 +492,22 @@ public class ProfileManager extends MasterDaemon {
         }
     }
 
-    public ProfileElement findProfileElementObject(String queryId) {
-        return queryIdToProfileMap.get(queryId);
+    public ProfileElement findProfileElementObject(String profileId) {
+        return profileIdToProfileMap.get(profileId);
     }
 
     /**
-     * Check if the query with specific query id is queried by specific user.
+     * Whether this profile belongs to the user.
      */
-    public void checkAuthByUserAndQueryId(String user, String queryId) throws AuthenticationException {
+    public void checkAuthByUserAndProfileId(String user, String profileId) throws AuthenticationException {
         readLock.lock();
         try {
-            ProfileElement element = queryIdToProfileMap.get(queryId);
+            ProfileElement element = profileIdToProfileMap.get(profileId);
             if (element == null) {
-                throw new AuthenticationException("query with id " + queryId + " not found");
+                throw new AuthenticationException("query with id " + profileId + " not found");
             }
             if (!element.infoStrings.get(SummaryProfile.USER).equals(user)) {
-                throw new AuthenticationException("Access deny to view query with id: " + queryId);
+                throw new AuthenticationException("Access deny to view query with id: " + profileId);
             }
         } finally {
             readLock.unlock();
@@ -521,7 +517,7 @@ public class ProfileManager extends MasterDaemon {
     public String getQueryIdByTraceId(String traceId) {
         readLock.lock();
         try {
-            for (Map.Entry<String, ProfileElement> entry : queryIdToProfileMap.entrySet()) {
+            for (Map.Entry<String, ProfileElement> entry : profileIdToProfileMap.entrySet()) {
                 if (entry.getValue().infoStrings.getOrDefault(SummaryProfile.TRACE_ID, "").equals(traceId)) {
                     return entry.getKey();
                 }
@@ -542,7 +538,7 @@ public class ProfileManager extends MasterDaemon {
     public void cleanProfile() {
         writeLock.lock();
         try {
-            queryIdToProfileMap.clear();
+            profileIdToProfileMap.clear();
             queryIdToExecutionProfiles.clear();
         } finally {
             writeLock.unlock();
@@ -694,7 +690,7 @@ public class ProfileManager extends MasterDaemon {
     protected List<ProfileElement> getProfilesNeedStore() {
         List<ProfileElement> profilesToBeStored = Lists.newArrayList();
 
-        queryIdToProfileMap.forEach((queryId, profileElement) -> {
+        profileIdToProfileMap.forEach((profileId, profileElement) -> {
             if (profileElement.profile.shouldStoreToStorage()) {
                 profilesToBeStored.add(profileElement);
             }
@@ -769,7 +765,7 @@ public class ProfileManager extends MasterDaemon {
         long totalProfileSize = 0;
 
         // Collect all profiles that has been stored to storage
-        for (ProfileElement profileElement : queryIdToProfileMap.values()) {
+        for (ProfileElement profileElement : profileIdToProfileMap.values()) {
             if (profileElement.profile.profileHasBeenStored()) {
                 totalProfileSize += profileElement.profile.getProfileSize();
                 profileDeque.add(profileElement);
@@ -778,16 +774,16 @@ public class ProfileManager extends MasterDaemon {
 
         final int maxSpilledProfileNum = Config.max_spilled_profile_num;
         final long spilledProfileLimitBytes = Config.spilled_profile_storage_limit_bytes;
-        List<ProfileElement> queryIdToBeRemoved = Lists.newArrayList();
+        List<ProfileElement> profilesToBeRemoved = Lists.newArrayList();
 
         while (profileDeque.size() > maxSpilledProfileNum || totalProfileSize >= spilledProfileLimitBytes) {
             // First profile is the oldest profile
             ProfileElement profileElement = profileDeque.poll();
             totalProfileSize -= profileElement.profile.getProfileSize();
-            queryIdToBeRemoved.add(profileElement);
+            profilesToBeRemoved.add(profileElement);
         }
 
-        return queryIdToBeRemoved;
+        return profilesToBeRemoved;
     }
 
     // We can not store all profiles on storage, because the storage space is limited
@@ -798,43 +794,43 @@ public class ProfileManager extends MasterDaemon {
         }
 
         try {
-            List<ProfileElement> queryIdToBeRemoved = Lists.newArrayList();
+            List<ProfileElement> profilesToBeRemoved = Lists.newArrayList();
             readLock.lock();
             try {
-                queryIdToBeRemoved = getProfilesToBeRemoved();
+                profilesToBeRemoved = getProfilesToBeRemoved();
             } finally {
                 readLock.unlock();
             }
 
-            if (queryIdToBeRemoved.isEmpty()) {
+            if (profilesToBeRemoved.isEmpty()) {
                 return;
             }
 
             // Archive or delete profiles based on configuration
             if (Config.enable_profile_archive) {
                 // Move profiles to pending directory for archiving
-                moveProfilesToArchivePending(queryIdToBeRemoved);
+                moveProfilesToArchivePending(profilesToBeRemoved);
             } else {
                 // Directly delete profiles if archiving is disabled
-                deleteProfilesFromStorage(queryIdToBeRemoved);
+                deleteProfilesFromStorage(profilesToBeRemoved);
             }
 
             // Remove profile references from memory
             writeLock.lock();
             try {
-                for (ProfileElement profileElement : queryIdToBeRemoved) {
-                    queryIdToProfileMap.remove(profileElement.profile.getSummaryProfile().getProfileId());
-                    TUniqueId thriftQueryId = DebugUtil.parseTUniqueIdFromString(
-                            profileElement.profile.getSummaryProfile().getProfileId());
-                    queryIdToExecutionProfiles.remove(thriftQueryId);
+                for (ProfileElement profileElement : profilesToBeRemoved) {
+                    profileIdToProfileMap.remove(profileElement.profile.getSummaryProfile().getProfileId());
+                    for (ExecutionProfile executionProfile : profileElement.profile.getExecutionProfiles()) {
+                        queryIdToExecutionProfiles.remove(executionProfile.getQueryId());
+                    }
                 }
             } finally {
                 writeLock.unlock();
             }
 
-            if (queryIdToBeRemoved.size() != 0 && LOG.isDebugEnabled()) {
+            if (profilesToBeRemoved.size() != 0 && LOG.isDebugEnabled()) {
                 StringBuilder builder = new StringBuilder();
-                for (ProfileElement profileElement : queryIdToBeRemoved) {
+                for (ProfileElement profileElement : profilesToBeRemoved) {
                     builder.append(profileElement.profile.getSummaryProfile().getProfileId()).append(",");
                 }
                 LOG.debug("Remove outdated profile: {}", builder.toString());
@@ -876,7 +872,7 @@ public class ProfileManager extends MasterDaemon {
 
             readLock.lock();
             try {
-                if (!queryIdToProfileMap.containsKey(profileId)) {
+                if (!profileIdToProfileMap.containsKey(profileId)) {
                     LOG.debug("Wild profile {}, need to be removed.", profileDirAbsPath);
                     brokenProfiles.add(profileDirAbsPath);
                 }
@@ -928,14 +924,14 @@ public class ProfileManager extends MasterDaemon {
     protected PriorityQueue<ProfileElement> getProfileOrderByQueryFinishTimeDesc() {
         readLock.lock();
         try {
-            PriorityQueue<ProfileElement> queryIdDeque = new PriorityQueue<>(Comparator.comparingLong(
+            PriorityQueue<ProfileElement> profileQueue = new PriorityQueue<>(Comparator.comparingLong(
                     (ProfileElement profileElement) -> profileElement.profile.getQueryFinishTimestamp()).reversed());
 
-            queryIdToProfileMap.forEach((queryId, profileElement) -> {
-                queryIdDeque.add(profileElement);
+            profileIdToProfileMap.forEach((profileId, profileElement) -> {
+                profileQueue.add(profileElement);
             });
 
-            return queryIdDeque;
+            return profileQueue;
         } finally {
             readLock.unlock();
         }
@@ -946,14 +942,14 @@ public class ProfileManager extends MasterDaemon {
     protected PriorityQueue<ProfileElement> getProfileOrderByQueryFinishTime() {
         readLock.lock();
         try {
-            PriorityQueue<ProfileElement> queryIdDeque = new PriorityQueue<>(Comparator.comparingLong(
+            PriorityQueue<ProfileElement> profileQueue = new PriorityQueue<>(Comparator.comparingLong(
                     (ProfileElement profileElement) -> profileElement.profile.getQueryFinishTimestamp()));
 
-            queryIdToProfileMap.forEach((queryId, profileElement) -> {
-                queryIdDeque.add(profileElement);
+            profileIdToProfileMap.forEach((profileId, profileElement) -> {
+                profileQueue.add(profileElement);
             });
 
-            return queryIdDeque;
+            return profileQueue;
         } finally {
             readLock.unlock();
         }
@@ -963,14 +959,14 @@ public class ProfileManager extends MasterDaemon {
     protected PriorityQueue<ProfileElement> getProfileOrderByQueryStartTime() {
         readLock.lock();
         try {
-            PriorityQueue<ProfileElement> queryIdDeque = new PriorityQueue<>(Comparator.comparingLong(
+            PriorityQueue<ProfileElement> profileQueue = new PriorityQueue<>(Comparator.comparingLong(
                     (ProfileElement profileElement) -> profileElement.profile.getSummaryProfile().getQueryBeginTime()));
 
-            queryIdToProfileMap.forEach((queryId, profileElement) -> {
-                queryIdDeque.add(profileElement);
+            profileIdToProfileMap.forEach((profileId, profileElement) -> {
+                profileQueue.add(profileElement);
             });
 
-            return queryIdDeque;
+            return profileQueue;
         } finally {
             readLock.unlock();
         }
@@ -996,8 +992,8 @@ public class ProfileManager extends MasterDaemon {
     }
 
     public String getLastProfileId() {
-        PriorityQueue<ProfileElement> queueIdDeque = getProfileOrderByQueryFinishTimeDesc();
-        ProfileElement profileElement = queueIdDeque.poll();
+        PriorityQueue<ProfileElement> profileQueue = getProfileOrderByQueryFinishTimeDesc();
+        ProfileElement profileElement = profileQueue.poll();
         return profileElement.profile.getSummaryProfile().getProfileId();
     }
 
@@ -1045,20 +1041,20 @@ public class ProfileManager extends MasterDaemon {
         writeLock.lock();
 
         try {
-            if (this.queryIdToProfileMap.size() + numOfNewProfiles <= Config.max_query_profile_num) {
+            if (this.profileIdToProfileMap.size() + numOfNewProfiles <= Config.max_query_profile_num) {
                 return;
             }
 
             // profile is ordered by query finish time
             // query finished earlier will be on the top of heap
             // query finished time of unfinished query is INT_MAX, so they will be on the bottom of the heap.
-            PriorityQueue<ProfileElement> queueIdDeque = getProfileOrderByQueryFinishTime();
+            PriorityQueue<ProfileElement> profileQueue = getProfileOrderByQueryFinishTime();
 
-            while (queueIdDeque.size() + numOfNewProfiles > Config.max_query_profile_num && !queueIdDeque.isEmpty()) {
-                ProfileElement profileElement = queueIdDeque.poll();
+            while (profileQueue.size() + numOfNewProfiles > Config.max_query_profile_num && !profileQueue.isEmpty()) {
+                ProfileElement profileElement = profileQueue.poll();
                 String profileId = profileElement.profile.getSummaryProfile().getProfileId();
                 stringBuilder.append(profileId).append(",");
-                queryIdToProfileMap.remove(profileId);
+                profileIdToProfileMap.remove(profileId);
                 for (ExecutionProfile executionProfile : profileElement.profile.getExecutionProfiles()) {
                     queryIdToExecutionProfiles.remove(executionProfile.getQueryId());
                 }
@@ -1069,7 +1065,7 @@ public class ProfileManager extends MasterDaemon {
                 }
             }
         } finally {
-            int profileNum = queryIdToProfileMap.size();
+            int profileNum = profileIdToProfileMap.size();
             writeLock.unlock();
 
             if (stringBuilder.length() != 0) {
@@ -1083,7 +1079,7 @@ public class ProfileManager extends MasterDaemon {
         StringBuilder stringBuilder = new StringBuilder();
         readLock.lock();
         try {
-            for (ProfileElement profileElement : queryIdToProfileMap.values()) {
+            for (ProfileElement profileElement : profileIdToProfileMap.values()) {
                 stringBuilder.append(profileElement.profile.debugInfo()).append("\n");
             }
         } finally {
@@ -1097,9 +1093,9 @@ public class ProfileManager extends MasterDaemon {
         readLock.lock();
 
         try {
-            PriorityQueue<ProfileElement> queueIdDeque = getProfileOrderByQueryFinishTimeDesc();
-            while (!queueIdDeque.isEmpty() && limit > 0) {
-                ProfileElement profileElement = queueIdDeque.poll();
+            PriorityQueue<ProfileElement> profileQueue = getProfileOrderByQueryFinishTimeDesc();
+            while (!profileQueue.isEmpty() && limit > 0) {
+                ProfileElement profileElement = profileQueue.poll();
                 Map<String, String> infoStrings = profileElement.infoStrings;
                 if (infoStrings.get(SummaryProfile.TASK_TYPE).equals(profileType.toString())) {
                     List<String> row = Lists.newArrayList();
@@ -1121,21 +1117,18 @@ public class ProfileManager extends MasterDaemon {
         return profileLoadStatus.get() == ProfileLoadStatus.LOADED;
     }
 
-    public void removeProfile(String profileId) {
+    public void removeProfile(Profile profile) {
         writeLock.lock();
         try {
-            ProfileElement profileToRemove = this.queryIdToProfileMap.remove(profileId);
-            if (profileToRemove != null) {
-                for (ExecutionProfile executionProfile : profileToRemove.profile.getExecutionProfiles()) {
-                    queryIdToExecutionProfiles.remove(executionProfile.getQueryId());
-                }
+            profileIdToProfileMap.remove(profile.getId());
+            for (ExecutionProfile executionProfile : profile.getExecutionProfiles()) {
+                queryIdToExecutionProfiles.remove(executionProfile.getQueryId());
             }
         } finally {
             writeLock.unlock();
         }
     }
 
-
     /**
      * Moves profiles to the archive pending directory.
      * Files in pending will be archived when batch size is reached or timeout occurs.

fe/fe-core/src/main/java/org/apache/doris/common/profile/SummaryProfile.java

@@ -19,6 +19,7 @@ package org.apache.doris.common.profile;
 
 import org.apache.doris.common.Config;
 import org.apache.doris.common.io.Text;
+import org.apache.doris.common.util.DebugUtil;
 import org.apache.doris.common.util.SafeStringBuilder;
 import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.persist.gson.GsonUtils;
@@ -26,6 +27,7 @@ import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.StmtExecutor;
 import org.apache.doris.system.Backend;
 import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TUniqueId;
 import org.apache.doris.thrift.TUnit;
 import org.apache.doris.transaction.TransactionType;
 
@@ -40,11 +42,18 @@ import com.google.gson.annotations.SerializedName;
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import java.lang.reflect.Field;
+import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
@@ -56,6 +65,8 @@ public class SummaryProfile {
     // Summary
     public static final String SUMMARY_PROFILE_NAME = "Summary";
     public static final String PROFILE_ID = "Profile ID";
+    public static final String QUERY_RETRY_TIMES = "QueryRetryTimes";
+    public static final String QUERY_RETRY_DETAILS = "QueryRetryDetails";
     public static final String DORIS_VERSION = "Doris Version";
     public static final String TASK_TYPE = "Task Type";
     public static final String START_TIME = "Start Time";
@@ -180,6 +191,8 @@ public class SummaryProfile {
 
     // The display order of execution summary items.
     public static final ImmutableList<String> EXECUTION_SUMMARY_KEYS = ImmutableList.of(
+            QUERY_RETRY_TIMES,
+            QUERY_RETRY_DETAILS,
             WORKLOAD_GROUP,
             MAX_CONCURRENCY,
             MAX_QUEUE_SIZE,
@@ -308,6 +321,7 @@ public class SummaryProfile {
             .put(READ_BYTES_PER_SECOND, 1)
             .put(TAG, 1)
             .build();
+
     public boolean parsedByConnectionProcess = false;
     @SerializedName(value = "summaryProfile")
     private RuntimeProfile summaryProfile = new RuntimeProfile(SUMMARY_PROFILE_NAME);
@@ -318,145 +332,218 @@ public class SummaryProfile {
     @SerializedName(value = "parseSqlFinishTime")
     private long parseSqlFinishTime = -1;
     @SerializedName(value = "nereidsPreloadExternalMetadataTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsPreloadExternalMetadataTime = 0;
     @SerializedName(value = "nereidsLockTableStartTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsLockTableStartTime = -1;
     @SerializedName(value = "nereidsLockTableFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsLockTableFinishTime = -1;
     @SerializedName(value = "nereidsCollectTablePartitionFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsCollectTablePartitionFinishTime = -1;
     @SerializedName(value = "nereidsCollectTablePartitionTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsCollectTablePartitionTime = 0;
     @SerializedName(value = "nereidsAnalysisFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsAnalysisFinishTime = -1;
     @SerializedName(value = "nereidsRewriteFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsRewriteFinishTime = -1;
     @SerializedName(value = "nereidsPreRewriteByMvFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsPreRewriteByMvFinishTime = -1;
     @SerializedName(value = "nereidsOptimizeFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsOptimizeFinishTime = -1;
     @SerializedName(value = "nereidsTranslateFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsTranslateFinishTime = -1;
     @SerializedName(value = "nereidsGarbageCollectionTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsGarbageCollectionTime = -1;
     @SerializedName(value = "nereidsBeFoldConstTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsBeFoldConstTime = 0;
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsDistributeFinishTime = -1;
     // timestamp of query begin
     @SerializedName(value = "queryBeginTime")
     private long queryBeginTime = -1;
     @SerializedName(value = "initScanNodeStartTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long initScanNodeStartTime = -1;
     @SerializedName(value = "initScanNodeFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long initScanNodeFinishTime = -1;
     @SerializedName(value = "finalizeScanNodeStartTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long finalizeScanNodeStartTime = -1;
     @SerializedName(value = "finalizeScanNodeFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long finalizeScanNodeFinishTime = -1;
     @SerializedName(value = "getSplitsStartTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long getSplitsStartTime = -1;
     @SerializedName(value = "getPartitionsFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long getPartitionsFinishTime = -1;
     @SerializedName(value = "getPartitionFilesFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long getPartitionFilesFinishTime = -1;
     @SerializedName(value = "sinkSetPartitionValuesStartTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long sinkSetPartitionValuesStartTime = -1;
     @SerializedName(value = "sinkSetPartitionValuesFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long sinkSetPartitionValuesFinishTime = -1;
     @SerializedName(value = "getSplitsFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long getSplitsFinishTime = -1;
     @SerializedName(value = "createScanRangeFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long createScanRangeFinishTime = -1;
     // Plan end time
     @SerializedName(value = "queryPlanFinishTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long queryPlanFinishTime = -1;
     @SerializedName(value = "assignFragmentTime")
+    @ResetOnRetry(ResetScope.EXECUTION)
     private long assignFragmentTime = -1;
     @SerializedName(value = "fragmentSerializeTime")
+    @ResetOnRetry(ResetScope.EXECUTION)
     private long fragmentSerializeTime = -1;
     @SerializedName(value = "fragmentSendPhase1Time")
+    @ResetOnRetry(ResetScope.EXECUTION)
     private long fragmentSendPhase1Time = -1;
     @SerializedName(value = "fragmentSendPhase2Time")
+    @ResetOnRetry(ResetScope.EXECUTION)
     private long fragmentSendPhase2Time = -1;
     @SerializedName(value = "fragmentCompressedSize")
+    @ResetOnRetry(ResetScope.EXECUTION)
     private long fragmentCompressedSize = 0;
     @SerializedName(value = "fragmentRpcCount")
+    @ResetOnRetry(ResetScope.EXECUTION)
     private long fragmentRpcCount = 0;
+    @SerializedName(value = "queryScheduleStartTime")
+    @ResetOnRetry(ResetScope.EXECUTION)
+    private long queryScheduleStartTime = -1;
+    @SerializedName(value = "failedAttempts")
+    private int failedAttempts = 0;
+    private TUniqueId lastFailedAttemptId;
+    @SerializedName(value = "queryRetryDetails")
+    private Map<String, QueryAttempt> queryRetryDetails = Maps.newLinkedHashMap();
+    private TUniqueId queryAttemptId;
+    private long queryAttemptStartTimeMs = -1;
     // Fragment schedule and send end time
     @SerializedName(value = "queryScheduleFinishTime")
+    @ResetOnRetry(ResetScope.EXECUTION)
     private long queryScheduleFinishTime = -1;
     // Query result fetch end time
     @SerializedName(value = "queryFetchResultFinishTime")
+    @ResetOnRetry(ResetScope.EXECUTION)
     private long queryFetchResultFinishTime = -1;
     @SerializedName(value = "tempStarTime")
+    @ResetOnRetry(ResetScope.EXECUTION)
     private long tempStarTime = -1;
     @SerializedName(value = "queryFetchResultConsumeTime")
+    @ResetOnRetry(ResetScope.EXECUTION)
     private long queryFetchResultConsumeTime = 0;
     @SerializedName(value = "queryWriteResultConsumeTime")
+    @ResetOnRetry(ResetScope.EXECUTION)
     private long queryWriteResultConsumeTime = 0;
     @SerializedName(value = "getPartitionVersionTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long getPartitionVersionTime = 0;
     @SerializedName(value = "getPartitionVersionCount")
+    @ResetOnRetry(ResetScope.PLAN)
     private long getPartitionVersionCount = 0;
     @SerializedName(value = "getPartitionVersionByHasDataCount")
+    @ResetOnRetry(ResetScope.PLAN)
     private long getPartitionVersionByHasDataCount = 0;
     @SerializedName(value = "getTableVersionTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long getTableVersionTime = 0;
     @SerializedName(value = "getTableVersionCount")
+    @ResetOnRetry(ResetScope.PLAN)
     private long getTableVersionCount = 0;
     @SerializedName(value = "getMetaVersionRateLimitWaitTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long getMetaVersionRateLimitWaitTime = 0;
     @SerializedName(value = "transactionCommitBeginTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long transactionCommitBeginTime = -1;
     @SerializedName(value = "transactionCommitEndTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long transactionCommitEndTime = -1;
     @SerializedName(value = "filesystemOptTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long filesystemOptTime = -1;
     @SerializedName(value = "hmsAddPartitionTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long hmsAddPartitionTime = -1;
     @SerializedName(value = "hmsAddPartitionCnt")
+    @ResetOnRetry(ResetScope.PLAN)
     private long hmsAddPartitionCnt = 0;
     @SerializedName(value = "hmsUpdatePartitionTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long hmsUpdatePartitionTime = -1;
     @SerializedName(value = "hmsUpdatePartitionCnt")
+    @ResetOnRetry(ResetScope.PLAN)
     private long hmsUpdatePartitionCnt = 0;
     @SerializedName(value = "filesystemRenameFileCnt")
+    @ResetOnRetry(ResetScope.PLAN)
     private long filesystemRenameFileCnt = 0;
     @SerializedName(value = "filesystemRenameDirCnt")
+    @ResetOnRetry(ResetScope.PLAN)
     private long filesystemRenameDirCnt = 0;
     @SerializedName(value = "filesystemDeleteDirCnt")
+    @ResetOnRetry(ResetScope.PLAN)
     private long filesystemDeleteDirCnt = 0;
     @SerializedName(value = "filesystemDeleteFileCnt")
+    @ResetOnRetry(ResetScope.PLAN)
     private long filesystemDeleteFileCnt = 0;
     @SerializedName(value = "transactionType")
+    @ResetOnRetry(ResetScope.PLAN)
     private TransactionType transactionType = TransactionType.UNKNOWN;
     @SerializedName(value = "nereidsMvRewriteTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsMvRewriteTime = 0;
     @SerializedName(value = "externalCatalogMetaTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long externalCatalogMetaTime = 0;
     // Total time to get table meta, including time to get table meta from external catalog and time to do some
     // process based on the meta, such as partition prune.
     @SerializedName(value = "externalTableGetTableMetaTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long externalTableGetTableMetaTime = 0;
     // Total time to get partition values, including time to get partition values from external catalog and time to do
     // some process based on the partition values, such as partition prune.
     @SerializedName(value = "externalTableGetPartitionValuesTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long externalTableGetPartitionValuesTime = 0;
     // Total time to get partitions, including time to get partitions from external catalog and time to do some
     // process based on the partitions, such as partition prune.
     @SerializedName(value = "externalTableGetPartitionsTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long externalTableGetPartitionsTime = 0;
     // Total time to get partition files, including time to get partition files from external catalog and time to do
     // some process based on the partition files, such as creating scan range.
     @SerializedName(value = "externalTableGetPartitionFilesTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long externalTableGetPartitionFilesTime = 0;
     // Total time to get file scan tasks, including time to get file scan tasks from external catalog and time to do
     // some process based on the file scan tasks, such as creating scan range.
     @SerializedName(value = "externalTableGetFileScanTasksTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long externalTableGetFileScanTasksTime = 0;
     @SerializedName(value = "externalTvfInitTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long externalTvfInitTime = 0;
     @SerializedName(value = "nereidsPartitiionPruneTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long nereidsPartitiionPruneTime = 0;
     @SerializedName("maxConcurrency")
     private int maxConcurrency = 0;
@@ -482,14 +569,51 @@ public class SummaryProfile {
     private long remoteReadBytesPerSecond = -1L;
     // BE -> (RPC latency from FE to BE, Execution latency on bthread, Duration of doing work, RPC latency from BE
     // to FE)
+    @ResetOnRetry(ResetScope.EXECUTION)
     private Map<TNetworkAddress, List<Long>> rpcPhase1Latency;
+    @ResetOnRetry(ResetScope.EXECUTION)
     private Map<TNetworkAddress, List<Long>> rpcPhase2Latency;
+    @ResetOnRetry(ResetScope.PLAN)
     private Map<Backend, Long> assignedWeightPerBackend;
     @SerializedName("waitChangeVisibleStartTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long waitChangeVisibleStartTime = -1L;
     @SerializedName("waitChangeVisibleEndTime")
+    @ResetOnRetry(ResetScope.PLAN)
     private long waitChangeVisibleEndTime = -1L;
 
+    private enum ResetScope {
+        PLAN,
+        EXECUTION
+    }
+
+    @Retention(RetentionPolicy.RUNTIME)
+    @Target(ElementType.FIELD)
+    private @interface ResetOnRetry {
+        ResetScope value();
+    }
+
+    private static final Map<ResetScope, Map<Field, Object>> RETRY_FIELD_DEFAULTS = buildRetryFieldDefaults();
+
+    private static Map<ResetScope, Map<Field, Object>> buildRetryFieldDefaults() {
+        Map<ResetScope, Map<Field, Object>> defaults = new EnumMap<>(ResetScope.class);
+        for (ResetScope scope : ResetScope.values()) {
+            defaults.put(scope, new LinkedHashMap<>());
+        }
+        SummaryProfile initial = new SummaryProfile(false);
+        try {
+            for (Field field : SummaryProfile.class.getDeclaredFields()) {
+                ResetOnRetry reset = field.getAnnotation(ResetOnRetry.class);
+                if (reset != null) {
+                    defaults.get(reset.value()).put(field, field.get(initial));
+                }
+            }
+        } catch (IllegalAccessException e) {
+            throw new IllegalStateException("Cannot read summary profile defaults", e);
+        }
+        return defaults;
+    }
+
     public SummaryProfile() {
         this(true);
     }
@@ -570,6 +694,83 @@ public class SummaryProfile {
         updateExecutionSummaryProfile();
     }
 
+    public synchronized void startQueryAttempt(TUniqueId queryId, long startTimeMs) {
+        if (queryAttemptStartTimeMs >= 0 && Objects.equals(queryAttemptId, queryId)) {
+            return;
+        }
+        queryAttemptId = queryId;
+        queryAttemptStartTimeMs = startTimeMs;
+    }
+
+    public synchronized void recordQueryAttempt(TUniqueId queryId, boolean failed) {
+        String id = DebugUtil.printId(queryId);
+        QueryAttempt attempt = queryRetryDetails.get(id);
+        if (attempt == null) {
+            long startTimeMs = queryAttemptStartTimeMs >= 0 && Objects.equals(queryAttemptId, queryId)
+                    ? queryAttemptStartTimeMs : queryBeginTime;
+            if (startTimeMs < 0) {
+                return;
+            }
+            attempt = new QueryAttempt(TimeUtils.getStartTimeMs() - startTimeMs, failed);
+            queryRetryDetails.put(id, attempt);
+        } else if (failed) {
+            attempt.state = "FAILED";
+        }
+        executionSummaryProfile.addInfoString(QUERY_RETRY_DETAILS, GsonUtils.GSON.toJson(queryRetryDetails));
+    }
+
+    public synchronized void recordFailedAttempt(TUniqueId queryId) {
+        recordQueryAttempt(queryId, true);
+        if (lastFailedAttemptId != null && lastFailedAttemptId.equals(queryId)) {
+            return;
+        }
+        lastFailedAttemptId = queryId;
+        failedAttempts++;
+        executionSummaryProfile.addInfoString(QUERY_RETRY_TIMES, Integer.toString(failedAttempts));
+    }
+
+    public void setQueryScheduleStartTime(long startTime) {
+        queryScheduleStartTime = startTime;
+    }
+
+    private long getQueryScheduleStartTime() {
+        return queryScheduleStartTime > 0 ? queryScheduleStartTime : queryPlanFinishTime;
+    }
+
+    private long getQueryPlanStartTime() {
+        return Math.max(queryBeginTime, parseSqlFinishTime);
+    }
+
+    public void clearExecutionDetails() {
+        resetFields(ResetScope.EXECUTION);
+        executionSummaryProfile.addInfoString(SCHEDULE_TIME_PER_BE, "{}");
+        updateExecutionSummaryProfile();
+    }
+
+    public void clearPlanDetails() {
+        resetFields(ResetScope.PLAN);
+        for (String key : ImmutableList.of(FILESYSTEM_OPT_TIME, FILESYSTEM_OPT_RENAME_FILE_CNT,
+                FILESYSTEM_OPT_RENAME_DIR_CNT, FILESYSTEM_OPT_DELETE_FILE_CNT, FILESYSTEM_OPT_DELETE_DIR_CNT,
+                HMS_ADD_PARTITION_TIME, HMS_ADD_PARTITION_CNT, HMS_UPDATE_PARTITION_TIME, HMS_UPDATE_PARTITION_CNT)) {
+            executionSummaryProfile.addInfoString(key, "N/A");
+        }
+        updateExecutionSummaryProfile();
+        summaryProfile.addInfoString(DISTRIBUTED_PLAN, "N/A");
+        executionSummaryProfile.addInfoString(QUERY_BACKEND_SELECTION, "N/A");
+        executionSummaryProfile.addInfoString(LOAD_BACKEND_SELECTION, "N/A");
+        executionSummaryProfile.addInfoString(SPLITS_ASSIGNMENT_WEIGHT, "N/A");
+    }
+
+    private void resetFields(ResetScope scope) {
+        try {
+            for (Entry<Field, Object> entry : RETRY_FIELD_DEFAULTS.get(scope).entrySet()) {
+                entry.getKey().set(this, entry.getValue());
+            }
+        } catch (IllegalAccessException e) {
+            throw new IllegalStateException("Cannot reset summary profile fields", e);
+        }
+    }
+
     // This method is used to display the final data status when the overall query ends.
     // This can avoid recalculating some strings and so on every time during the update process.
     public void queryFinished() {
@@ -601,9 +802,11 @@ public class SummaryProfile {
     }
 
     private void updateExecutionSummaryProfile() {
+        executionSummaryProfile.addInfoString(QUERY_RETRY_TIMES, Integer.toString(failedAttempts));
+        executionSummaryProfile.addInfoString(QUERY_RETRY_DETAILS, GsonUtils.GSON.toJson(queryRetryDetails));
         executionSummaryProfile.addInfoString(PARSE_SQL_TIME, getPrettyParseSqlTime());
         executionSummaryProfile.addInfoString(PLAN_TIME,
-                getPrettyTime(queryPlanFinishTime, parseSqlFinishTime, TUnit.TIME_MS));
+                getPrettyTime(queryPlanFinishTime, getQueryPlanStartTime(), TUnit.TIME_MS));
         executionSummaryProfile.addInfoString(NEREIDS_PRELOAD_EXTERNAL_METADATA_TIME,
                 getPrettyNereidsPreloadExternalMetadataTime());
         executionSummaryProfile.addInfoString(WAIT_CHANGE_VISIBLE_TIME, getPrettyWaitChangeVisibleEndTime());
@@ -642,10 +845,10 @@ public class SummaryProfile {
         executionSummaryProfile.addInfoString(CREATE_SCAN_RANGE_TIME,
                 getPrettyTime(createScanRangeFinishTime, getSplitsFinishTime, TUnit.TIME_MS));
         executionSummaryProfile.addInfoString(SCHEDULE_TIME,
-                getPrettyTime(queryScheduleFinishTime, queryPlanFinishTime, TUnit.TIME_MS));
+                getPrettyTime(queryScheduleFinishTime, getQueryScheduleStartTime(), TUnit.TIME_MS));
         executionSummaryProfile.addInfoString(SCHEDULE_TIME_PER_BE, getRpcLatency());
         executionSummaryProfile.addInfoString(ASSIGN_FRAGMENT_TIME,
-                getPrettyTime(assignFragmentTime, queryPlanFinishTime, TUnit.TIME_MS));
+                getPrettyTime(assignFragmentTime, getQueryScheduleStartTime(), TUnit.TIME_MS));
         executionSummaryProfile.addInfoString(FRAGMENT_SERIALIZE_TIME,
                 getPrettyTime(fragmentSerializeTime, assignFragmentTime, TUnit.TIME_MS));
         executionSummaryProfile.addInfoString(SEND_FRAGMENT_PHASE1_TIME,
@@ -916,7 +1119,7 @@ public class SummaryProfile {
     }
 
     public int getPlanTimeMs() {
-        return getTimeMs(queryPlanFinishTime, parseSqlFinishTime);
+        return getTimeMs(queryPlanFinishTime, getQueryPlanStartTime());
     }
 
     public int getNereidsLockTableTimeMs() {
@@ -980,11 +1183,11 @@ public class SummaryProfile {
     }
 
     public int getScheduleTimeMs() {
-        return getTimeMs(queryScheduleFinishTime, queryPlanFinishTime);
+        return getTimeMs(queryScheduleFinishTime, getQueryScheduleStartTime());
     }
 
     public int getFragmentAssignTimsMs() {
-        return getTimeMs(assignFragmentTime, queryPlanFinishTime);
+        return getTimeMs(assignFragmentTime, getQueryScheduleStartTime());
     }
 
     public int getFragmentSerializeTimeMs() {
@@ -1436,6 +1639,18 @@ public class SummaryProfile {
                 + "}";
     }
 
+    private static class QueryAttempt {
+        @SerializedName(value = "durationMs")
+        private final long durationMs;
+        @SerializedName(value = "state")
+        private String state;
+
+        private QueryAttempt(long durationMs, boolean failed) {
+            this.durationMs = durationMs;
+            this.state = failed ? "FAILED" : "SUCCEEDED";
+        }
+    }
+
     public static class SummaryBuilder {
         private Map<String, String> map = Maps.newHashMap();
 

fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/QueryProfileAction.java

@@ -453,7 +453,7 @@ public class QueryProfileAction extends RestBaseController {
     private void checkAuthByUserAndQueryId(String queryId) throws AuthenticationException {
         String user = ConnectContext.get().getCurrentUserIdentity().getQualifiedUser();
         if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
-            ProfileManager.getInstance().checkAuthByUserAndQueryId(user, queryId);
+            ProfileManager.getInstance().checkAuthByUserAndProfileId(user, queryId);
         }
     }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java

@@ -312,6 +312,7 @@ public abstract class AbstractInsertExecutor {
                 }
             }
         } catch (Throwable t) {
+            executor.getSummaryProfile().recordFailedAttempt(ctx.queryId());
             onFail(t);
             if (ctx.getStatementContext().isIvmMTMVRewrite()) {
                 for (Throwable cause : ExceptionUtils.getThrowableList(t)) {

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

@@ -242,6 +242,10 @@ public class StmtExecutor {
     private boolean isHandleQueryInFe = false;
     // The profile of this execution
     private final Profile profile;
+    // Direct execute() calls finish their own profiles.
+    private boolean isInQueryRetry;
+    // queryRetry must finish the profile.
+    private boolean finishProfileInQueryRetry;
     private Boolean isForwardedToMaster = null;
     // Flag for execute prepare statement, need to use binary protocol resultset
     private boolean isComStmtExecute = false;
@@ -662,6 +666,21 @@ public class StmtExecutor {
     }
 
     public void queryRetry(TUniqueId queryId) throws Exception {
+        isInQueryRetry = true;
+        try {
+            executeWithReplanRetries(queryId);
+        } finally {
+            isInQueryRetry = false;
+            // Finish after the last attempt, including planning failures.
+            if (finishProfileInQueryRetry && context.isReturnResultFromLocal()
+                    && profile.getQueryFinishTimestamp() == Long.MAX_VALUE) {
+                updateProfile(true);
+            }
+            finishProfileInQueryRetry = false;
+        }
+    }
+
+    private void executeWithReplanRetries(TUniqueId queryId) throws Exception {
         TUniqueId firstQueryId = queryId;
         int retryTime = Config.max_query_retry_time;
         retryTime = retryTime <= 0 ? 1 : retryTime + 1;
@@ -672,6 +691,7 @@ public class StmtExecutor {
             retryTime = ((Queriable) parsedStmt).hasOutFileClause() ? 1 : retryTime;
         }
         for (int i = 1; i <= retryTime; i++) {
+            profile.getSummaryProfile().startQueryAttempt(queryId, TimeUtils.getStartTimeMs());
             try {
                 if (disableCloudVersionCacheOnRetry) {
                     executeWithVersionCacheDisabled(queryId);
@@ -680,6 +700,7 @@ public class StmtExecutor {
                 }
                 return;
             } catch (UserException e) {
+                profile.getSummaryProfile().recordFailedAttempt(context.queryId());
                 if (!SystemInfoService.needRetryWithReplan(e.getMessage()) || i == retryTime) {
                     // We have retried internally(in handleQueryWithRetry()) for other kinds of exceptions.
                     // And for error in SystemInfoService.NEED_REPLAN_ERRORS, they are not handled internally but here
@@ -704,7 +725,15 @@ public class StmtExecutor {
                         DebugUtil.printId(queryId), randomMillis);
                 Thread.sleep(randomMillis);
                 context.getState().reset();
+                if (finishProfileInQueryRetry) {
+                    profile.clearExecutionProfiles();
+                }
+                // Drop the old execution state before replanning.
+                planner = null;
+                setCoord(null);
+                profile.clearPlan();
             } catch (Exception e) {
+                profile.getSummaryProfile().recordFailedAttempt(context.queryId());
                 throw e;
             }
         }
@@ -739,6 +768,7 @@ public class StmtExecutor {
     }
 
     public void execute(TUniqueId queryId) throws Exception {
+        profile.getSummaryProfile().startQueryAttempt(queryId, TimeUtils.getStartTimeMs());
         SessionVariable sessionVariable = context.getSessionVariable();
         context.setEffectiveCloudCluster(null);
         externalDmlAuditCoordinator = null;
@@ -753,6 +783,7 @@ public class StmtExecutor {
             } catch (NereidsException | ParseException e) {
                 // COMPUTE_GROUPS_NO_ALIVE_BE, planner can't get alive be, need retry
                 if (Config.isCloudMode() && SystemInfoService.needRetryWithReplan(e.getMessage())) {
+                    profile.getSummaryProfile().recordFailedAttempt(context.queryId());
                     LOG.debug("planner failed with cloud compute group error, need retry. {}",
                             context.getQueryIdentifier(), e);
                     throw new UserException(e.getMessage());
@@ -775,6 +806,13 @@ public class StmtExecutor {
                 throw e;
             }
         } finally {
+            if (context.getState().getStateType() == MysqlStateType.ERR) {
+                profile.getSummaryProfile().recordFailedAttempt(context.queryId());
+            }
+            if (context.getSessionVariable().enableProfile() && !deferredForArrowFlight) {
+                profile.getSummaryProfile().recordQueryAttempt(context.queryId(),
+                        context.getState().getStateType() == MysqlStateType.ERR);
+            }
             // Preserve the effective per-query compute group before SET_VAR values are reverted.
             // Audit logging runs after this method returns and otherwise sees the session value.
             if (Config.isCloudMode()) {
@@ -791,6 +829,12 @@ public class StmtExecutor {
                             context.getQueryIdentifier(), t);
                 }
             }
+            // Record this attempt before restoring SET_VAR.
+            if (finishProfileInQueryRetry && parsedStmt instanceof LogicalPlanAdapter
+                    && ((LogicalPlanAdapter) parsedStmt).getLogicalPlan() instanceof Command
+                    && !DebugUtil.printId(context.queryId()).equals(profile.getId())) {
+                updateProfile(false);
+            }
             // revert Session Value
             try {
                 VariableMgr.revertSessionValue(sessionVariable);
@@ -1141,10 +1185,17 @@ public class StmtExecutor {
     }
 
     public void finalizeQuery() {
-        // The final profile report occurs after be returns the query data, and the profile cannot be
-        // received after unregisterQuery(), causing the instance profile to be lost, so we should wait
-        // for the profile before unregisterQuery().
-        updateProfile(true);
+        finalizeQuery(false);
+    }
+
+    void finalizeQuery(boolean willRetry) {
+        if (willRetry) {
+            // Discard the failed attempt; keep the statement open.
+            profile.clearExecutionProfiles();
+        } else {
+            // Update before unregistering to keep the final BE reports.
+            updateProfile(true);
+        }
         QeProcessorImpl.INSTANCE.unregisterQuery(queryId());
     }
 
@@ -1210,6 +1261,8 @@ public class StmtExecutor {
         int retryTime = Config.max_query_retry_time;
         retryTime = retryTime <= 0 ? 1 : retryTime + 1;
         for (int i = 0; i < retryTime; i++) {
+            boolean willRetry = false;
+            boolean letQueryRetryFinishProfile = false;
             try {
                 // reset query id for each retry
                 if (i > 0) {
@@ -1243,13 +1296,19 @@ public class StmtExecutor {
                         }
                     }
                 }
+                if (i > 0) {
+                    profile.getSummaryProfile().setQueryScheduleStartTime(TimeUtils.getStartTimeMs());
+                }
+                profile.getSummaryProfile().startQueryAttempt(context.queryId(), TimeUtils.getStartTimeMs());
                 handleQueryStmt();
                 LOG.info("Query {} finished", DebugUtil.printId(context.queryId));
                 break;
             } catch (RpcException | UserException e) {
+                profile.getSummaryProfile().recordFailedAttempt(context.queryId());
                 if (Config.isCloudMode() && SystemInfoService.needRetryWithReplan(e.getMessage())) {
                     // For errors in SystemInfoService.NEED_REPLAN_ERRORS,
                     // throw exception directly to trigger a replan retry outside(in StmtExecutor.queryRetry())
+                    letQueryRetryFinishProfile = isInQueryRetry && e instanceof UserException;
                     throw e;
                 }
                 // If the previous try is timeout or cancelled, then do not need try again.
@@ -1305,12 +1364,20 @@ public class StmtExecutor {
                 }
                 if (i != retryTime - 1 && isNeedRetry && context.getProtocolAdapter().canRetryQuery(context)) {
                     LOG.warn("retry {} times. stmt: {}", (i + 1), parsedStmt.getOrigStmt().originStmt);
+                    willRetry = true;
                 } else {
                     throw e;
                 }
+            } catch (Exception e) {
+                profile.getSummaryProfile().recordFailedAttempt(context.queryId());
+                throw e;
             } finally {
-                if (context.isReturnResultFromLocal()) {
-                    finalizeQuery();
+                if (letQueryRetryFinishProfile) {
+                    // Unregister now; the outer retry loop will finish the profile.
+                    finishProfileInQueryRetry = true;
+                    QeProcessorImpl.INSTANCE.unregisterQuery(queryId());
+                } else if (context.isReturnResultFromLocal()) {
+                    finalizeQuery(willRetry);
                 }
                 LOG.debug("Finalize query {}", DebugUtil.printId(context.queryId()));
             }
@@ -1424,7 +1491,34 @@ public class StmtExecutor {
         // and ensure the sql is finished normally. For example, if update profile
         // failed, the insert stmt should be success
         try {
-            profile.updateSummary(getSummaryInfo(isFinished), isFinished, this.planner);
+            if (isFinished && ((coord != null && !coord.getExecStatus().ok())
+                    || (!deferredForArrowFlight && context.getState().getStateType() == MysqlStateType.ERR))) {
+                profile.getSummaryProfile().recordFailedAttempt(queryId());
+            }
+            if (isFinished) {
+                profile.getSummaryProfile().recordQueryAttempt(queryId(), false);
+            }
+            boolean isCommand = parsedStmt instanceof LogicalPlanAdapter
+                    && ((LogicalPlanAdapter) parsedStmt).getLogicalPlan() instanceof Command;
+            // Finish command profiles after the outer retry loop.
+            if (isFinished && isInQueryRetry && isCommand) {
+                finishProfileInQueryRetry = true;
+                isFinished = false;
+            }
+            Map<String, String> summaryInfo;
+            if (isFinished && finishProfileInQueryRetry && isCommand
+                    && DebugUtil.printId(context.queryId()).equals(profile.getId())) {
+                // Keep this attempt's settings after SET_VAR is restored.
+                long now = System.currentTimeMillis();
+                SummaryBuilder completion = new SummaryBuilder().endTime(TimeUtils.longToTimeString(now))
+                        .taskState(coord == null ? context.getState().toString()
+                                : coord.getExecStatus().getErrorCode().name());
+                addTotalTime(completion, now - context.getStartTime());
+                summaryInfo = completion.build();
+            } else {
+                summaryInfo = getSummaryInfo(isFinished);
+            }
+            profile.updateSummary(summaryInfo, isFinished, this.planner);
             if (planner instanceof NereidsPlanner) {
                 NereidsPlanner nereidsPlanner = ((NereidsPlanner) planner);
                 profile.setPhysicalPlan(nereidsPlanner.getPhysicalPlan());

fe/fe-core/src/test/java/org/apache/doris/common/profile/AutoProfileTest.java

@@ -62,13 +62,13 @@ public class AutoProfileTest {
         profile.autoProfileDurationMs = 10000;
         Thread.sleep(100);
         profile.updateSummary(summaryInfo, true, null);
-        Assertions.assertNull(ProfileManager.getInstance().queryIdToProfileMap.get(profile.getId()));
+        Assertions.assertNull(ProfileManager.getInstance().profileIdToProfileMap.get(profile.getId()));
 
         profile = createProfile();
         profile.setSummaryProfile(summaryProfile);
         profile.autoProfileDurationMs = 50;
         Thread.sleep(200);
         profile.updateSummary(summaryInfo, true, null);
-        Assertions.assertNotNull(ProfileManager.getInstance().queryIdToProfileMap.get(profile.getId()));
+        Assertions.assertNotNull(ProfileManager.getInstance().profileIdToProfileMap.get(profile.getId()));
     }
 }

fe/fe-core/src/test/java/org/apache/doris/common/profile/ProfileManagerTest.java

@@ -43,6 +43,8 @@ import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.parallel.ResourceLock;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.Mockito;
 
 import java.io.File;
@@ -471,7 +473,7 @@ class ProfileManagerTest {
         // Clean profile
         profileManager.cleanProfile();
         // Make sure map is cleaned.
-        Assertions.assertTrue(profileManager.queryIdToProfileMap.isEmpty());
+        Assertions.assertTrue(profileManager.profileIdToProfileMap.isEmpty());
         Assertions.assertTrue(profileManager.queryIdToExecutionProfiles.isEmpty());
     }
 
@@ -534,7 +536,7 @@ class ProfileManagerTest {
 
             profileManager.loadProfilesFromStorageIfFirstTime(true);
             Assertions.assertEquals(ProfileLoadStatus.LOADED, profileManager.profileLoadStatus.get());
-            Assertions.assertEquals(30, profileManager.queryIdToProfileMap.size());
+            Assertions.assertEquals(30, profileManager.profileIdToProfileMap.size());
             Assertions.assertEquals(0, profileManager.queryIdToExecutionProfiles.size());
         } catch (InterruptedException e) {
             throw new RuntimeException(e);
@@ -606,7 +608,7 @@ class ProfileManagerTest {
         File[] files = tempDir.listFiles();
         assert files != null;
         Assertions.assertEquals(30, files.length);
-        Assertions.assertEquals(30, profileManager.queryIdToProfileMap.size());
+        Assertions.assertEquals(30, profileManager.profileIdToProfileMap.size());
         Assertions.assertEquals(0, profileManager.queryIdToExecutionProfiles.size());
     }
 
@@ -690,7 +692,7 @@ class ProfileManagerTest {
             profileManager.deleteOutdatedProfilesFromStorage();
 
             // Verify correct profiles were deleted
-            Assertions.assertEquals(10, profileManager.queryIdToProfileMap.size());
+            Assertions.assertEquals(10, profileManager.profileIdToProfileMap.size());
             for (int j = 0; j < spyProfiles.size(); j++) {
                 Mockito.verify(spyProfiles.get(j), Mockito.times(j < 20 ? 1 : 0)).deleteFromStorage();
             }
@@ -703,6 +705,82 @@ class ProfileManagerTest {
         }
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void testPruneMixedQueryAndBrokerLoadProfiles(boolean archiveEnabled) throws Exception {
+        boolean originalArchiveEnabled = Config.enable_profile_archive;
+        String originalArchivePath = Config.profile_archive_path;
+        int originalMaxSpilled = Config.max_spilled_profile_num;
+        long originalStorageLimit = Config.spilled_profile_storage_limit_bytes;
+        int originalBatchSize = Config.profile_archive_batch_size;
+        int originalPendingTimeout = Config.profile_archive_pending_timeout_seconds;
+        List<Profile> profiles = new ArrayList<>();
+        List<File> profileFiles = new ArrayList<>();
+        try {
+            Config.enable_profile_archive = archiveEnabled;
+            Config.profile_archive_path = "";
+            Config.max_spilled_profile_num = 1;
+            Config.max_query_profile_num = 3;
+            Config.spilled_profile_storage_limit_bytes = Long.MAX_VALUE;
+            Config.profile_archive_batch_size = 100;
+            Config.profile_archive_pending_timeout_seconds = 3600;
+            long finishTime = System.currentTimeMillis();
+            for (int i = 0; i < 3; i++) {
+                boolean brokerLoad = i != 1;
+                TUniqueId queryId = new TUniqueId(0x22040L, i * 2 + 1);
+                Profile profile = ProfilePersistentTest.constructRandomProfile(0);
+                profile.getSummaryProfile().getSummary().getInfoStrings().put(SummaryProfile.PROFILE_ID,
+                        brokerLoad ? String.valueOf(123456 + i) : DebugUtil.printId(queryId));
+                profile.isQueryFinished = true;
+                profile.setQueryFinishTimestamp(finishTime + i);
+                profile.addExecutionProfile(new ExecutionProfile(queryId, Lists.newArrayList(0)));
+                if (brokerLoad) {
+                    profile.addExecutionProfile(new ExecutionProfile(
+                            new TUniqueId(queryId.hi, queryId.lo + 1), Lists.newArrayList(1)));
+                }
+                for (ExecutionProfile execution : profile.getExecutionProfiles()) {
+                    profileManager.addExecutionProfile(execution);
+                }
+                profileManager.pushProfile(profile);
+                profile.writeToStorage(ProfileManager.PROFILE_STORAGE_PATH);
+                File profileFile = new File(profile.getProfileStoragePath());
+                Assertions.assertTrue(profileFile.exists());
+                profileFiles.add(profileFile);
+                profiles.add(profile);
+            }
+            List<ProfileElement> selected = profileManager.getProfilesToBeRemoved();
+            Assertions.assertEquals(2, selected.size());
+            Assertions.assertSame(profiles.get(0), selected.get(0).profile);
+            Assertions.assertSame(profiles.get(1), selected.get(1).profile);
+
+            profileManager.profileLoadStatus.set(ProfileLoadStatus.LOADED);
+            profileManager.deleteOutdatedProfilesFromStorage();
+
+            Assertions.assertEquals(1, profileManager.profileIdToProfileMap.size());
+            Assertions.assertEquals(2, profileManager.queryIdToExecutionProfiles.size());
+            for (int i = 0; i < profiles.size(); i++) {
+                Profile profile = profiles.get(i);
+                boolean retained = i == 2;
+                Assertions.assertEquals(retained, profileManager.profileIdToProfileMap.containsKey(profile.getId()));
+                Assertions.assertEquals(retained, profileFiles.get(i).exists());
+                for (ExecutionProfile execution : profile.getExecutionProfiles()) {
+                    Assertions.assertEquals(retained,
+                            profileManager.queryIdToExecutionProfiles.containsKey(execution.getQueryId()));
+                }
+                if (!retained && archiveEnabled) {
+                    Assertions.assertTrue(new File(tempDir, "archive/pending/" + profileFiles.get(i).getName()).exists());
+                }
+            }
+        } finally {
+            Config.enable_profile_archive = originalArchiveEnabled;
+            Config.profile_archive_path = originalArchivePath;
+            Config.max_spilled_profile_num = originalMaxSpilled;
+            Config.spilled_profile_storage_limit_bytes = originalStorageLimit;
+            Config.profile_archive_batch_size = originalBatchSize;
+            Config.profile_archive_pending_timeout_seconds = originalPendingTimeout;
+        }
+    }
+
     @Test
     void testGetBrokenProfiles() throws IOException {
         // Create normal profiles
@@ -800,7 +878,7 @@ class ProfileManagerTest {
         profileManager.loadProfilesFromStorageIfFirstTime(true);
 
         // Verify all profiles are loaded
-        Assertions.assertEquals(numProfiles, profileManager.queryIdToProfileMap.size());
+        Assertions.assertEquals(numProfiles, profileManager.profileIdToProfileMap.size());
     }
 
     @Test
@@ -938,11 +1016,11 @@ class ProfileManagerTest {
         profileManager.deleteOutdatedProfilesFromMemory(0);
 
         // Verify memory profile count is within limit
-        Assertions.assertEquals(Config.max_query_profile_num, profileManager.queryIdToProfileMap.size());
+        Assertions.assertEquals(Config.max_query_profile_num, profileManager.profileIdToProfileMap.size());
 
         // Verify newest profiles are kept
         for (String profileId : profilesToKeep) {
-            Assertions.assertTrue(profileManager.queryIdToProfileMap.containsKey(profileId));
+            Assertions.assertTrue(profileManager.profileIdToProfileMap.containsKey(profileId));
         }
     }
 
@@ -964,13 +1042,13 @@ class ProfileManagerTest {
             profileManager.pushProfile(profile);
         }
         // Verify profile count
-        Assertions.assertEquals(3, profileManager.queryIdToProfileMap.size());
+        Assertions.assertEquals(3, profileManager.profileIdToProfileMap.size());
 
         // Try to delete outdated profiles
         profileManager.deleteOutdatedProfilesFromMemory(0);
 
         // Verify no profiles were deleted
-        Assertions.assertEquals(3, profileManager.queryIdToProfileMap.size());
+        Assertions.assertEquals(3, profileManager.profileIdToProfileMap.size());
 
         for (int i = 3; i < 5; i++) {
             Profile profile = profiles.get(i);
@@ -980,17 +1058,17 @@ class ProfileManagerTest {
         for (int i = 0; i < 5; i++) {
             Profile profile = profiles.get(i);
             if (i <= 1) {
-                Assertions.assertFalse(profileManager.queryIdToProfileMap.containsKey(profile.getId()));
+                Assertions.assertFalse(profileManager.profileIdToProfileMap.containsKey(profile.getId()));
             } else {
-                Assertions.assertTrue(profileManager.queryIdToProfileMap.containsKey(profile.getId()));
+                Assertions.assertTrue(profileManager.profileIdToProfileMap.containsKey(profile.getId()));
             }
         }
 
         profiles.clear();
 
-        Assertions.assertEquals(3, profileManager.queryIdToProfileMap.size());
+        Assertions.assertEquals(3, profileManager.profileIdToProfileMap.size());
         for (Profile profile : profiles) {
-            Assertions.assertTrue(profileManager.queryIdToProfileMap.containsKey(profile.getId()));
+            Assertions.assertTrue(profileManager.profileIdToProfileMap.containsKey(profile.getId()));
         }
     }
 
@@ -1023,10 +1101,10 @@ class ProfileManagerTest {
 
         // Verify unfinished profile was not deleted
         for (Profile profile : profileUnfinished) {
-            Assertions.assertTrue(profileManager.queryIdToProfileMap.containsKey(profile.getId()));
+            Assertions.assertTrue(profileManager.profileIdToProfileMap.containsKey(profile.getId()));
         }
 
-        Assertions.assertEquals(5, profileManager.queryIdToProfileMap.size());
+        Assertions.assertEquals(5, profileManager.profileIdToProfileMap.size());
 
         profileFinished.clear();
         for (int i = 0; i < 5; i++) {
@@ -1041,7 +1119,7 @@ class ProfileManagerTest {
 
         // Verify unfinished profile was not deleted
         for (Profile profile : profileUnfinished) {
-            Assertions.assertTrue(profileManager.queryIdToProfileMap.containsKey(profile.getId()));
+            Assertions.assertTrue(profileManager.profileIdToProfileMap.containsKey(profile.getId()));
         }
     }
 

fe/fe-core/src/test/java/org/apache/doris/common/profile/ProfileTest.java

@@ -20,19 +20,24 @@ package org.apache.doris.common.profile;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.util.DebugUtil;
 import org.apache.doris.common.util.SafeStringBuilder;
+import org.apache.doris.system.Backend;
+import org.apache.doris.thrift.TNetworkAddress;
 import org.apache.doris.thrift.TUniqueId;
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.ResourceLock;
 import org.mockito.Mockito;
 
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.nio.file.Files;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 
@@ -54,7 +59,44 @@ public class ProfileTest {
 
     @AfterEach
     public void tearDown() {
-        ProfileManager.getInstance().removeProfile(profile.getId());
+        ProfileManager.getInstance().removeProfile(profile);
+    }
+
+    @Test
+    public void testClearAttemptSummaryState() {
+        SummaryProfile summary = profile.getSummaryProfile();
+        summary.setQueryBeginTime(123);
+        summary.update(Map.of(SummaryProfile.DISTRIBUTED_PLAN, "abandoned-plan",
+                SummaryProfile.QUERY_BACKEND_SELECTION, "abandoned-query-selection",
+                SummaryProfile.LOAD_BACKEND_SELECTION, "abandoned-load-selection"));
+        TNetworkAddress address = new TNetworkAddress("abandoned-backend", 9060);
+        summary.setRpcPhase1Latency(Map.of(address, List.of(1L, 2L, 3L, 4L)));
+        summary.setRpcPhase2Latency(Map.of(address, List.of(5L, 6L, 7L, 8L)));
+        Backend backend = new Backend(1L, "abandoned-backend", 9050);
+        summary.setAssignedWeightPerBackend(Map.of(backend, 100L));
+        summary.queryFinished();
+        summary.update(Collections.emptyMap());
+        Assertions.assertTrue(summary.getExecutionSummary().getInfoString(SummaryProfile.SCHEDULE_TIME_PER_BE)
+                .contains("abandoned-backend"));
+
+        profile.clearExecutionProfiles();
+        summary.update(Collections.emptyMap());
+        Assertions.assertEquals("{}", summary.getExecutionSummary()
+                .getInfoString(SummaryProfile.SCHEDULE_TIME_PER_BE));
+        // Inner retries reuse scan assignments.
+        Assertions.assertEquals("abandoned-plan", summary.getSummary().getInfoString(SummaryProfile.DISTRIBUTED_PLAN));
+        Assertions.assertTrue(summary.getExecutionSummary().getInfoString(SummaryProfile.SPLITS_ASSIGNMENT_WEIGHT)
+                .contains("abandoned-backend"));
+
+        profile.clearPlan();
+        summary.queryFinished();
+        summary.update(Collections.emptyMap());
+        Assertions.assertEquals("N/A", summary.getSummary().getInfoString(SummaryProfile.DISTRIBUTED_PLAN));
+        for (String key : List.of(SummaryProfile.QUERY_BACKEND_SELECTION, SummaryProfile.LOAD_BACKEND_SELECTION,
+                SummaryProfile.SPLITS_ASSIGNMENT_WEIGHT)) {
+            Assertions.assertEquals("N/A", summary.getExecutionSummary().getInfoString(key));
+        }
+        Assertions.assertEquals(123, summary.getQueryBeginTime());
     }
 
     @Test
@@ -87,6 +129,77 @@ public class ProfileTest {
         Assertions.assertTrue(Long.MAX_VALUE != profile.getQueryFinishTimestamp());
     }
 
+    @Test
+    @ResourceLock("global")
+    public void testShortFailedQueryRemovesExecutionProfileWithoutSummary() {
+        ProfileManager manager = ProfileManager.getInstance();
+        TUniqueId queryId = executionProfile.getQueryId();
+        profile.autoProfileDurationMs = 10_000;
+        profile.getSummaryProfile().setQueryBeginTime(System.currentTimeMillis());
+        manager.addExecutionProfile(executionProfile);
+
+        try {
+            Assertions.assertSame(executionProfile, manager.getExecutionProfile(queryId));
+            Assertions.assertNull(manager.findProfileElementObject(profile.getId()));
+
+            profile.updateSummary(new HashMap<>(), true, null);
+
+            Assertions.assertNull(manager.getExecutionProfile(queryId));
+            Assertions.assertNull(manager.findProfileElementObject(profile.getId()));
+        } finally {
+            manager.cleanProfile();
+        }
+    }
+
+    @Test
+    @ResourceLock("global")
+    public void testShortCompletedQueryRemovesStoredProfileAndExecutionProfile() {
+        ProfileManager manager = ProfileManager.getInstance();
+        TUniqueId queryId = executionProfile.getQueryId();
+        profile.autoProfileDurationMs = 10_000;
+        profile.getSummaryProfile().setQueryBeginTime(System.currentTimeMillis());
+        manager.addExecutionProfile(executionProfile);
+
+        try {
+            profile.updateSummary(new HashMap<>(), false, null);
+            Assertions.assertNotNull(manager.findProfileElementObject(profile.getId()));
+
+            profile.updateSummary(new HashMap<>(), true, null);
+
+            Assertions.assertNull(manager.getExecutionProfile(queryId));
+            Assertions.assertNull(manager.findProfileElementObject(profile.getId()));
+        } finally {
+            manager.cleanProfile();
+        }
+    }
+
+    @Test
+    @ResourceLock("global")
+    public void testJobBelowCombinedThresholdRemovesMultipleExecutionProfiles() {
+        ProfileManager manager = ProfileManager.getInstance();
+        TUniqueId firstQueryId = executionProfile.getQueryId();
+        UUID secondUuid = UUID.randomUUID();
+        TUniqueId secondQueryId = new TUniqueId(secondUuid.getMostSignificantBits(),
+                secondUuid.getLeastSignificantBits());
+        ExecutionProfile secondExecutionProfile = new ExecutionProfile(secondQueryId, Collections.emptyList());
+        profile.addExecutionProfile(secondExecutionProfile);
+        profile.autoProfileDurationMs = 10_000;
+        // Broker Load keeps its per-task threshold.
+        profile.getSummaryProfile().setQueryBeginTime(System.currentTimeMillis() - 15_000);
+        manager.addExecutionProfile(executionProfile);
+        manager.addExecutionProfile(secondExecutionProfile);
+
+        try {
+            profile.updateSummary(new HashMap<>(), true, null);
+
+            Assertions.assertNull(manager.getExecutionProfile(firstQueryId));
+            Assertions.assertNull(manager.getExecutionProfile(secondQueryId));
+            Assertions.assertNull(manager.findProfileElementObject(profile.getId()));
+        } finally {
+            manager.cleanProfile();
+        }
+    }
+
     @Test
     public void testShouldStoreToStorage() {
         // Initially not finished, should not store

fe/fe-core/src/test/java/org/apache/doris/common/profile/SummaryProfileTest.java

@@ -18,10 +18,23 @@
 package org.apache.doris.common.profile;
 
 import org.apache.doris.common.Config;
+import org.apache.doris.common.util.DebugUtil;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TUniqueId;
+import org.apache.doris.transaction.TransactionType;
 
 import com.google.common.collect.ImmutableMap;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.lang.reflect.Field;
+import java.util.Arrays;
 
 public class SummaryProfileTest {
 
@@ -177,4 +190,193 @@ public class SummaryProfileTest {
         Assertions.assertEquals("2ms", executionSummary.getInfoString(SummaryProfile.NEREIDS_OPTIMIZE_TIME));
         Assertions.assertEquals(32, profile.getNereidsOptimizeTimeMs());
     }
+
+    @Test
+    public void testReplanUsesFinalAttemptPhaseTimes() {
+        SummaryProfile profile = new SummaryProfile();
+        profile.setQueryBeginTime(1);
+        profile.setParseSqlFinishTime(6);
+        profile.setNereidsLockTableStartTime(8);
+        profile.setNereidsLockTableFinishTime(10);
+        profile.setNereidsAnalysisTime(15);
+        profile.setQueryPlanFinishTime(66);
+        profile.setQueryScheduleFinishTime(78);
+        profile.addNereidsPreloadExternalMetadataTime(11);
+        profile.addExternalTableGetTableMetaTime(22);
+        profile.addGetPartitionVersionTime(33_000_000);
+        profile.setTransactionBeginTime(TransactionType.HMS);
+        profile.addHmsAddPartitionCnt(7);
+        profile.update(ImmutableMap.of());
+        Assertions.assertEquals("7", profile.getExecutionSummary().getInfoString(SummaryProfile.HMS_ADD_PARTITION_CNT));
+        profile.clearExecutionDetails();
+        profile.clearPlanDetails();
+        profile.setQueryBeginTime(1000);
+        profile.setQueryPlanFinishTime(1010);
+        profile.setQueryScheduleFinishTime(1020);
+        profile.update(ImmutableMap.of());
+        Assertions.assertEquals("10ms", profile.getExecutionSummary().getInfoString(SummaryProfile.PLAN_TIME));
+        Assertions.assertEquals("10ms", profile.getExecutionSummary().getInfoString(SummaryProfile.SCHEDULE_TIME));
+        Assertions.assertEquals("N/A", profile.getExecutionSummary().getInfoString(SummaryProfile.NEREIDS_ANALYSIS_TIME));
+        Assertions.assertEquals(10, profile.getScheduleTimeMs());
+        Assertions.assertEquals(0, profile.getNereidsPreloadExternalMetadataTimeMs());
+        Assertions.assertEquals(0, profile.getExternalCatalogMetaTimeMs());
+        Assertions.assertEquals(0, profile.getGetPartitionVersionTimeMs());
+        Assertions.assertEquals("N/A", profile.getExecutionSummary().getInfoString(SummaryProfile.HMS_ADD_PARTITION_CNT));
+    }
+
+    @Test
+    public void testRedispatchExcludesFailedAttemptAndBackoff() throws Exception {
+        SummaryProfile profile = new SummaryProfile();
+        profile.setQueryBeginTime(1);
+        profile.setParseSqlFinishTime(6);
+        profile.setQueryPlanFinishTime(66);
+        profile.setQueryScheduleFinishTime(78);
+        profile.setQueryFetchResultFinishTime(91);
+        profile.updateFragmentCompressedSize(1234);
+        profile.updateFragmentRpcCount(7);
+        profile.clearExecutionDetails();
+        profile.setQueryScheduleStartTime(1000);
+        Field assignTime = SummaryProfile.class.getDeclaredField("assignFragmentTime");
+        assignTime.setAccessible(true);
+        assignTime.setLong(profile, 1010);
+        profile.setQueryScheduleFinishTime(1012);
+        profile.update(ImmutableMap.of());
+        Assertions.assertEquals(1, profile.getQueryBeginTime());
+        Assertions.assertEquals("60ms", profile.getExecutionSummary().getInfoString(SummaryProfile.PLAN_TIME));
+        Assertions.assertEquals("12ms", profile.getExecutionSummary().getInfoString(SummaryProfile.SCHEDULE_TIME));
+        Assertions.assertEquals("N/A", profile.getExecutionSummary().getInfoString(SummaryProfile.WAIT_FETCH_RESULT_TIME));
+        Assertions.assertEquals(12, profile.getScheduleTimeMs());
+        Assertions.assertEquals(10, profile.getFragmentAssignTimsMs());
+        Assertions.assertEquals("10ms", profile.getExecutionSummary().getInfoString(SummaryProfile.ASSIGN_FRAGMENT_TIME));
+        Assertions.assertEquals(0, profile.getFragmentCompressedSizeByte());
+        Assertions.assertEquals(0, profile.getFragmentRPCCount());
+        Assertions.assertEquals("0", profile.getExecutionSummary().getInfoString(SummaryProfile.FRAGMENT_RPC_COUNT));
+    }
+
+    @Test
+    public void testRetryCleanupAfterSerialization() {
+        SummaryProfile profile = new SummaryProfile();
+        profile.setQueryBeginTime(1);
+        profile.setParseSqlFinishTime(6);
+        profile.setQueryPlanFinishTime(66);
+        profile.setQueryScheduleFinishTime(78);
+        profile.updateFragmentRpcCount(7);
+        profile.setTransactionBeginTime(TransactionType.HMS);
+        profile.addHmsAddPartitionCnt(9);
+        SummaryProfile restored = GsonUtils.GSON.fromJson(GsonUtils.GSON.toJson(profile), SummaryProfile.class);
+        restored.setRpcPhase1Latency(ImmutableMap.of(new TNetworkAddress("backend", 1234),
+                Arrays.asList(1L, 2L, 3L, 4L)));
+        restored.update(ImmutableMap.of());
+        Assertions.assertNotEquals("{}", restored.getExecutionSummary().getInfoString(
+                SummaryProfile.SCHEDULE_TIME_PER_BE));
+        restored.clearExecutionDetails();
+        Assertions.assertEquals("{}", restored.getExecutionSummary().getInfoString(SummaryProfile.SCHEDULE_TIME_PER_BE));
+        Assertions.assertEquals(0, restored.getFragmentRPCCount());
+        Assertions.assertEquals("60ms", restored.getExecutionSummary().getInfoString(SummaryProfile.PLAN_TIME));
+        Assertions.assertEquals("9", restored.getExecutionSummary().getInfoString(SummaryProfile.HMS_ADD_PARTITION_CNT));
+        restored.clearPlanDetails();
+        Assertions.assertEquals(1, restored.getQueryBeginTime());
+        Assertions.assertEquals("N/A", restored.getExecutionSummary().getInfoString(SummaryProfile.PLAN_TIME));
+        Assertions.assertEquals("N/A", restored.getExecutionSummary().getInfoString(SummaryProfile.HMS_ADD_PARTITION_CNT));
+        Assertions.assertEquals(7, profile.getFragmentRPCCount());
+        SummaryProfile clean = new SummaryProfile();
+        clean.clearExecutionDetails();
+        clean.clearPlanDetails();
+        Assertions.assertEquals(0, clean.getFragmentRPCCount());
+    }
+
+    @Test
+    public void testFailedAttemptsSurviveCleanupWithoutDoubleCounting() {
+        SummaryProfile profile = new SummaryProfile();
+        TUniqueId first = new TUniqueId(1, 1);
+        profile.recordFailedAttempt(first);
+        profile.recordFailedAttempt(new TUniqueId(1, 1));
+        profile.clearExecutionDetails();
+        profile.clearPlanDetails();
+        profile.recordFailedAttempt(new TUniqueId(1, 2));
+        profile.update(ImmutableMap.of());
+        Assertions.assertEquals("2", profile.getExecutionSummary().getInfoString(SummaryProfile.QUERY_RETRY_TIMES));
+    }
+
+    @Test
+    public void testFailedAttemptDurationIsRecordedOnce() {
+        SummaryProfile profile = new SummaryProfile();
+        TUniqueId firstId = new TUniqueId(1, 1);
+        profile.setQueryBeginTime(1000);
+        try (MockedStatic<TimeUtils> clock = Mockito.mockStatic(TimeUtils.class)) {
+            clock.when(TimeUtils::getStartTimeMs).thenReturn(1100L);
+            profile.recordFailedAttempt(firstId);
+            clock.when(TimeUtils::getStartTimeMs).thenReturn(1500L);
+            profile.recordFailedAttempt(firstId);
+            profile.clearExecutionDetails();
+            profile.clearPlanDetails();
+            profile.update(ImmutableMap.of());
+            String text = profile.getExecutionSummary().getInfoString("QueryRetryDetails");
+            Assertions.assertNotNull(text);
+            Assertions.assertNotEquals("null", text);
+            JsonObject attempts = JsonParser.parseString(text).getAsJsonObject();
+            Assertions.assertEquals(1, attempts.size());
+            JsonObject first = attempts.getAsJsonObject(DebugUtil.printId(firstId));
+            Assertions.assertEquals(100, first.get("durationMs").getAsLong());
+            Assertions.assertEquals("FAILED", first.get("state").getAsString());
+        }
+    }
+
+    @Test
+    public void testAttemptDurationsExcludeWaitAndSurviveStorage() {
+        SummaryProfile profile = new SummaryProfile();
+        TUniqueId firstId = new TUniqueId(1, 1);
+        TUniqueId finalId = new TUniqueId(1, 2);
+        try (MockedStatic<TimeUtils> clock = Mockito.mockStatic(TimeUtils.class)) {
+            profile.startQueryAttempt(firstId, 1000);
+            profile.startQueryAttempt(firstId, 1050);
+            clock.when(TimeUtils::getStartTimeMs).thenReturn(1100L);
+            profile.recordFailedAttempt(firstId);
+            profile.clearExecutionDetails();
+            profile.clearPlanDetails();
+            profile.startQueryAttempt(finalId, 5000);
+            clock.when(TimeUtils::getStartTimeMs).thenReturn(5300L);
+            profile.recordQueryAttempt(finalId, false);
+            clock.when(TimeUtils::getStartTimeMs).thenReturn(6000L);
+            profile.recordQueryAttempt(finalId, false);
+            SummaryProfile stored = GsonUtils.GSON.fromJson(GsonUtils.GSON.toJson(profile), SummaryProfile.class);
+            stored.update(ImmutableMap.of());
+            JsonObject attempts = JsonParser.parseString(stored.getExecutionSummary()
+                    .getInfoString(SummaryProfile.QUERY_RETRY_DETAILS)).getAsJsonObject();
+            Assertions.assertEquals(2, attempts.size());
+            Assertions.assertEquals(100, attempts.getAsJsonObject(DebugUtil.printId(firstId))
+                    .get("durationMs").getAsLong());
+            JsonObject last = attempts.getAsJsonObject(DebugUtil.printId(finalId));
+            Assertions.assertEquals(300, last.get("durationMs").getAsLong());
+            Assertions.assertEquals("SUCCEEDED", last.get("state").getAsString());
+            Assertions.assertEquals("1", stored.getExecutionSummary().getInfoString(SummaryProfile.QUERY_RETRY_TIMES));
+        }
+    }
+
+    @Test
+    public void testUnstartedAttemptHasNoDuration() {
+        SummaryProfile profile = new SummaryProfile();
+        profile.recordFailedAttempt(new TUniqueId(1, 1));
+        profile.update(ImmutableMap.of());
+        Assertions.assertEquals("{}", profile.getExecutionSummary().getInfoString(SummaryProfile.QUERY_RETRY_DETAILS));
+    }
+
+    @Test
+    public void testLateFailureUpdatesStateWithoutExtendingDuration() {
+        SummaryProfile profile = new SummaryProfile();
+        TUniqueId id = new TUniqueId(1, 1);
+        profile.startQueryAttempt(id, 1000);
+        try (MockedStatic<TimeUtils> clock = Mockito.mockStatic(TimeUtils.class)) {
+            clock.when(TimeUtils::getStartTimeMs).thenReturn(1100L);
+            profile.recordQueryAttempt(id, false);
+            clock.when(TimeUtils::getStartTimeMs).thenReturn(1500L);
+            profile.recordFailedAttempt(id);
+            JsonObject attempt = JsonParser.parseString(profile.getExecutionSummary()
+                    .getInfoString(SummaryProfile.QUERY_RETRY_DETAILS)).getAsJsonObject()
+                    .getAsJsonObject(DebugUtil.printId(id));
+            Assertions.assertEquals(100, attempt.get("durationMs").getAsLong());
+            Assertions.assertEquals("FAILED", attempt.get("state").getAsString());
+        }
+    }
+
 }

fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/CloudInsertProfileRetryTest.java

@@ -0,0 +1,424 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.trees.plans.commands.insert;
+
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.EnvFactory;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.Status;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.profile.ExecutionProfile;
+import org.apache.doris.common.profile.Profile;
+import org.apache.doris.common.profile.ProfileManager;
+import org.apache.doris.common.profile.ProfileManager.ProfileType;
+import org.apache.doris.common.profile.SummaryProfile;
+import org.apache.doris.common.util.DebugUtil;
+import org.apache.doris.nereids.NereidsPlanner;
+import org.apache.doris.nereids.glue.LogicalPlanAdapter;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.plans.RelationId;
+import org.apache.doris.nereids.trees.plans.commands.Command;
+import org.apache.doris.nereids.trees.plans.commands.merge.MergeIntoCommand;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalEmptyRelation;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalSink;
+import org.apache.doris.planner.DataSink;
+import org.apache.doris.planner.PlanFragment;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.Coordinator;
+import org.apache.doris.qe.QeProcessorImpl;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.rpc.RpcException;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.TDetailedReportParams;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TQueryOptions;
+import org.apache.doris.thrift.TQueryProfile;
+import org.apache.doris.thrift.TRuntimeProfileNode;
+import org.apache.doris.thrift.TRuntimeProfileTree;
+import org.apache.doris.thrift.TStatusCode;
+import org.apache.doris.thrift.TUniqueId;
+import org.apache.doris.utframe.TestWithFeService;
+
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.api.parallel.ResourceLock;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import org.mockito.stubbing.Answer;
+
+import java.lang.reflect.Method;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class CloudInsertProfileRetryTest extends TestWithFeService {
+    @ParameterizedTest
+    @ValueSource(strings = {"SUCCESS", "SUCCESS_WITH_SESSION_REVERT", "SUCCESS_EMPTY_WITH_SESSION_REVERT",
+            "EMPTY_WITH_SESSION_REVERT", "PLANNING_FAILURE", "EXHAUSTED",
+            "CANCELLED", "DIRECT", "RPC", "NON_REPLAN", "MERGE_SUCCESS",
+            "MERGE_PLANNING_FAILURE", "MERGE_SUCCESS_WITH_SESSION_REVERT",
+            "MERGE_SUCCESS_EMPTY_WITH_SESSION_REVERT"})
+    @ResourceLock("global")
+    public void testInsertProfileRemainsOpenUntilOuterRetryDecision(String caseName, @TempDir Path storage) throws Exception {
+        boolean merge = caseName.startsWith("MERGE_");
+        String outcome = merge ? caseName.substring("MERGE_".length()) : caseName;
+        boolean realExecution = merge && "SUCCESS_EMPTY_WITH_SESSION_REVERT".equals(outcome);
+        String oldCloudUniqueId = Config.cloud_unique_id;
+        int oldRetryTime = Config.max_query_retry_time;
+        SessionVariable oldSession = connectContext.getSessionVariable();
+        SessionVariable session = new SessionVariable();
+        session.enableProfile = true;
+        session.autoProfileThresholdMs = 0;
+        session.parallelPipelineTaskNum = 1;
+        session.cloudCluster = "profile-insert-retry-test";
+        ConnectContext context = Mockito.spy(connectContext);
+        context.setSessionVariable(session);
+        Mockito.doReturn(connectContext.getComputeGroup()).when(context).getComputeGroup();
+        TUniqueId firstId = new TUniqueId(0x22040L, 31L);
+        context.setQueryId(firstId);
+        context.setStartTime();
+        String sql = merge
+                ? "merge into profile_retry_db.profile_retry_target t using (select 1 as id) s on t.id = s.id "
+                    + "when not matched then insert (id) values (s.id)"
+                : "insert into profile_retry_db.profile_retry_target select 1";
+        StmtExecutor executor = Mockito.spy(new StmtExecutor(context, sql));
+        Method parse = StmtExecutor.class.getDeclaredMethod("parseByNereids");
+        parse.setAccessible(true);
+        parse.invoke(executor);
+        Assertions.assertTrue(executor.isProfileSafeStmt());
+        if (merge) {
+            Assertions.assertInstanceOf(MergeIntoCommand.class,
+                    ((LogicalPlanAdapter) executor.getParsedStmt()).getLogicalPlan());
+        } else {
+            // Match the profile type set by InsertIntoTableCommand.
+            executor.setProfileType(ProfileType.LOAD);
+        }
+        LogicalPlanAdapter adapter = Mockito.spy((LogicalPlanAdapter) executor.getParsedStmt());
+        Command command = Mockito.spy((Command) adapter.getLogicalPlan());
+        if (realExecution) {
+            Mockito.doReturn(command).when(adapter).getLogicalPlan();
+            executor.setParsedStmt(adapter);
+        }
+        Profile profile = executor.getProfile();
+        ProfileManager manager = ProfileManager.getInstance();
+        NereidsPlanner planner = Mockito.mock(NereidsPlanner.class);
+        Mockito.when(planner.getPhysicalRelations()).thenReturn(Collections.emptyList());
+        OlapTable table = Mockito.mock(OlapTable.class);
+        Database database = new Database(12345L, "profile_retry_db");
+        Mockito.when(table.getDatabase()).thenReturn(database);
+        EnvFactory factory = Mockito.mock(EnvFactory.class);
+        List<ExecutionProfile> executions = new ArrayList<>();
+        AtomicInteger attempts = new AtomicInteger();
+        AtomicInteger callbacks = new AtomicInteger();
+        AtomicInteger spillChecks = new AtomicInteger();
+        ExecutorService backgroundStorage = Executors.newSingleThreadExecutor();
+        Mockito.when(factory.createCoordinator(Mockito.eq(context), Mockito.eq(planner), Mockito.any(),
+                Mockito.anyLong())).thenAnswer(invocation -> {
+                    TUniqueId queryId = context.queryId();
+                    ExecutionProfile execution = new ExecutionProfile(queryId, Collections.singletonList(0));
+                    boolean empty = "EMPTY_WITH_SESSION_REVERT".equals(outcome)
+                            || ("SUCCESS_EMPTY_WITH_SESSION_REVERT".equals(outcome) && attempts.get() > 1);
+                    if (!empty) {
+                        executions.add(execution);
+                    }
+                    Coordinator coordinator = Mockito.mock(Coordinator.class);
+                    TQueryOptions options = new TQueryOptions();
+                    options.enable_profile = true;
+                    Mockito.when(coordinator.getQueryOptions()).thenReturn(options);
+                    Mockito.when(coordinator.getExecutionProfile()).thenReturn(execution);
+                    Mockito.when(coordinator.getExecStatus()).thenReturn(Status.OK);
+                    Mockito.when(coordinator.getBeToInstancesNum())
+                            .thenReturn(empty ? Collections.emptyMap() : Collections.singletonMap(
+                                    attempts.get() == 1 ? "abandoned-be:9060" : "final-be:9060",
+                                    attempts.get() == 1 ? 7 : 2));
+                    Mockito.when(coordinator.isDone()).thenReturn(true);
+                    Mockito.when(coordinator.isQueryCancelled()).thenReturn("CANCELLED".equals(outcome));
+                    Mockito.doAnswer(dispatch -> {
+                        QeProcessorImpl.INSTANCE.registerQueryFinishCallback(DebugUtil.printId(queryId),
+                                callbacks::incrementAndGet);
+                        execution.addFragmentBackend(0, 1L);
+                        Assertions.assertTrue(execution.updateProfile(completeBackendReport(queryId),
+                                new TNetworkAddress("127.0.0.1", 9060), true).ok());
+                        Assertions.assertTrue(execution.isCompleted());
+                        if (attempts.get() == 1) {
+                            if ("RPC".equals(outcome)) {
+                                throw new RpcException("test-be", SystemInfoService.ERROR_E230);
+                            }
+                            throw new UserException("NON_REPLAN".equals(outcome)
+                                    ? "terminal insert failure" : SystemInfoService.ERROR_E230);
+                        }
+                        return null;
+                    }).when(coordinator).exec();
+                    return coordinator;
+                });
+        Answer<Void> executeAttempt = invocation -> {
+            TUniqueId queryId = realExecution ? context.queryId() : invocation.getArgument(0);
+            if (!realExecution) {
+                context.setQueryId(queryId);
+                context.setStartTime();
+            }
+            if (outcome.endsWith("WITH_SESSION_REVERT")) {
+                if (realExecution) {
+                    Assertions.assertTrue(session.setVarOnce("parallel_pipeline_task_num", "3"));
+                } else {
+                    session.parallelPipelineTaskNum = 3;
+                }
+            }
+            profile.getSummaryProfile().setQueryBeginTime(context.getStartTime());
+            if (attempts.incrementAndGet() > 1) {
+                Assertions.assertEquals(Long.MAX_VALUE, profile.getQueryFinishTimestamp());
+                Assertions.assertNull(manager.getExecutionProfile(firstId));
+                Assertions.assertNull(manager.findProfileElementObject(DebugUtil.printId(firstId)));
+                Assertions.assertTrue(profile.getExecutionProfiles().isEmpty());
+                Assertions.assertNull(profile.getPhysicalPlan());
+                Assertions.assertNull(executor.getCoord());
+                if ("PLANNING_FAILURE".equals(outcome)) {
+                    context.getState().setError("terminal planning failure");
+                    throw new UserException("terminal planning failure");
+                }
+            }
+            PhysicalEmptyRelation plan = new PhysicalEmptyRelation(
+                    new RelationId(attempts.get()), Collections.emptyList(), Optional.empty(), null,
+                    PhysicalProperties.ANY, null);
+            Mockito.when(planner.getPhysicalPlan()).thenReturn(plan);
+            executor.setPlanner(planner);
+            if (realExecution && attempts.get() > 1) {
+                // RowLevelDmlCommand returns before executeSingleInsert for empty input.
+                context.getState().setOk();
+                return null;
+            }
+            AbstractInsertExecutor insert = new AbstractInsertExecutor(context, table, "retry_test", planner,
+                    Optional.empty(), "EMPTY_WITH_SESSION_REVERT".equals(outcome)
+                            || ("SUCCESS_EMPTY_WITH_SESSION_REVERT".equals(outcome) && attempts.get() > 1),
+                    attempts.get()) {
+                @Override
+                public void beginTransaction() {
+                }
+
+                @Override
+                protected void finalizeSink(PlanFragment fragment, DataSink sink, PhysicalSink physicalSink) {
+                }
+
+                @Override
+                protected void beforeExec() {
+                }
+
+                @Override
+                protected void onComplete() {
+                    context.getState().setOk();
+                }
+
+                @Override
+                protected void onFail(Throwable error) {
+                    context.getState().setError(error.getMessage());
+                }
+
+                @Override
+                protected void afterExec(StmtExecutor statementExecutor) {
+                }
+            };
+            try {
+                insert.executeSingleInsert(executor);
+            } catch (UserException error) {
+                if (!"DIRECT".equals(outcome)) {
+                    // The attempt ended; the retry decision is still pending.
+                    backgroundStorage.submit(() -> {
+                        spillChecks.incrementAndGet();
+                        if (profile.shouldStoreToStorage()) {
+                            profile.writeToStorage(storage.toString());
+                            profile.releaseMemory();
+                        }
+                    }).get(10, TimeUnit.SECONDS);
+                    Assertions.assertFalse(profile.profileHasBeenStored(),
+                            "a retryable INSERT must not become eligible for background spill");
+                    Assertions.assertEquals(Long.MAX_VALUE, profile.getQueryFinishTimestamp());
+                    Assertions.assertNotNull(profile.rowsProducedMap);
+                }
+                if ("CANCELLED".equals(outcome)) {
+                    // Cancel before the outer retry decision.
+                    Mockito.when(insert.getCoordinator().getExecStatus())
+                            .thenReturn(new Status(TStatusCode.CANCELLED, "cancelled before outer decision"));
+                }
+                throw error;
+            }
+            if (outcome.endsWith("WITH_SESSION_REVERT") && !realExecution) {
+                // Restore SET_VAR before the outer finalizer.
+                session.parallelPipelineTaskNum = 1;
+            }
+            return null;
+        };
+        if (realExecution) {
+            Mockito.doAnswer(executeAttempt).when(command).run(Mockito.eq(context), Mockito.eq(executor));
+        } else {
+            Mockito.doAnswer(executeAttempt).when(executor).execute(Mockito.any(TUniqueId.class));
+        }
+        try (MockedStatic<EnvFactory> factories = Mockito.mockStatic(EnvFactory.class)) {
+            factories.when(EnvFactory::getInstance).thenReturn(factory);
+            Config.cloud_unique_id = "profile-insert-retry-test";
+            Config.max_query_retry_time = "EXHAUSTED".equals(outcome) ? 0 : 1;
+            context.setThreadLocalInfo();
+            if (outcome.startsWith("SUCCESS") || "EMPTY_WITH_SESSION_REVERT".equals(outcome)
+                    || "NON_REPLAN".equals(outcome)) {
+                executor.queryRetry(firstId);
+            } else if ("DIRECT".equals(outcome)) {
+                Assertions.assertThrows(UserException.class, () -> executor.execute(firstId));
+            } else {
+                Class<? extends Exception> errorType = "RPC".equals(outcome) ? RpcException.class : UserException.class;
+                Assertions.assertThrows(errorType,
+                        () -> executor.queryRetry(firstId));
+            }
+            int expectedAttempts = outcome.startsWith("SUCCESS") || "PLANNING_FAILURE".equals(outcome) ? 2 : 1;
+            Assertions.assertEquals(expectedAttempts, attempts.get());
+            String details = profile.getSummaryProfile().getExecutionSummary()
+                    .getInfoString(SummaryProfile.QUERY_RETRY_DETAILS);
+            JsonObject attemptDetails = JsonParser.parseString(details).getAsJsonObject();
+            Assertions.assertEquals(expectedAttempts, attemptDetails.size());
+            JsonObject finalAttempt = attemptDetails.getAsJsonObject(DebugUtil.printId(context.queryId()));
+            Assertions.assertEquals(outcome.startsWith("SUCCESS") || outcome.contains("EMPTY") ? "SUCCEEDED" : "FAILED",
+                    finalAttempt.get("state").getAsString());
+            int failedAttempts = "PLANNING_FAILURE".equals(outcome) ? 2
+                    : "EMPTY_WITH_SESSION_REVERT".equals(outcome) ? 0 : 1;
+            Assertions.assertEquals(Integer.toString(failedAttempts), profile.getSummaryProfile().getExecutionSummary()
+                    .getInfoString(SummaryProfile.QUERY_RETRY_TIMES));
+            if (realExecution) {
+                Assertions.assertEquals(1, session.parallelPipelineTaskNum);
+            }
+            Assertions.assertNotEquals(Long.MAX_VALUE, profile.getQueryFinishTimestamp());
+            Assertions.assertEquals(DebugUtil.printId(context.queryId()), profile.getId());
+            Assertions.assertEquals(merge ? "QUERY" : "LOAD", profile.getSummaryProfile().getSummary()
+                    .getInfoString(SummaryProfile.TASK_TYPE));
+            Assertions.assertNotNull(manager.findProfileElementObject(profile.getId()));
+            Assertions.assertEquals(executions.size(), callbacks.get());
+            if (outcome.contains("EMPTY")) {
+                Assertions.assertTrue(profile.getExecutionProfiles().isEmpty());
+                if (realExecution) {
+                    Assertions.assertNull(executor.getCoord());
+                    Assertions.assertEquals(1, executions.size());
+                    Mockito.verify(command, Mockito.times(2)).run(context, executor);
+                } else {
+                    Mockito.verify(executor.getCoord(), Mockito.never()).exec();
+                }
+            } else if (outcome.startsWith("SUCCESS")) {
+                Assertions.assertEquals(Collections.singletonList(executions.get(1)), profile.getExecutionProfiles());
+            } else if ("PLANNING_FAILURE".equals(outcome)) {
+                Assertions.assertTrue(profile.getExecutionProfiles().isEmpty());
+                Assertions.assertNull(profile.getPhysicalPlan());
+                Assertions.assertEquals("ERR", profile.getSummaryProfile().getSummary()
+                        .getInfoString(SummaryProfile.TASK_STATE));
+                Assertions.assertNull(executor.getCoord());
+                Assertions.assertEquals("0", profile.getSummaryProfile().getExecutionSummary()
+                        .getInfoString(SummaryProfile.TOTAL_INSTANCES_NUM));
+                Assertions.assertEquals("", profile.getSummaryProfile().getExecutionSummary()
+                        .getInfoString(SummaryProfile.INSTANCES_NUM_PER_BE));
+                Assertions.assertFalse(profile.getProfileByLevel().contains("abandoned-be:9060"));
+            }
+            Assertions.assertEquals("DIRECT".equals(outcome) || "RPC".equals(outcome)
+                    || "NON_REPLAN".equals(outcome) || "EMPTY_WITH_SESSION_REVERT".equals(outcome)
+                    ? 0 : 1, spillChecks.get());
+            for (ExecutionProfile execution : executions) {
+                Assertions.assertNull(QeProcessorImpl.INSTANCE.getCoordinator(execution.getQueryId()));
+            }
+            if ("CANCELLED".equals(outcome)) {
+                Assertions.assertEquals("CANCELLED", profile.getSummaryProfile().getSummary()
+                        .getInfoString(SummaryProfile.TASK_STATE));
+            }
+            if (outcome.endsWith("WITH_SESSION_REVERT")) {
+                Assertions.assertEquals("3", profile.getSummaryProfile().getExecutionSummary()
+                        .getInfoString(SummaryProfile.PARALLEL_FRAGMENT_EXEC_INSTANCE));
+            }
+            if (outcome.startsWith("SUCCESS") || outcome.contains("EMPTY")) {
+                Assertions.assertEquals("OK", profile.getSummaryProfile().getSummary()
+                        .getInfoString(SummaryProfile.TASK_STATE));
+                Assertions.assertNotNull(profile.getSummaryProfile().getSummary()
+                        .getInfoString(SummaryProfile.END_TIME));
+                Assertions.assertNotNull(profile.getSummaryProfile().getSummary()
+                        .getInfoString(SummaryProfile.TOTAL_TIME));
+            }
+            if (outcome.startsWith("SUCCESS")) {
+                Assertions.assertFalse(profile.getProfileByLevel().contains("abandoned-be:9060"));
+            }
+            // Store only after the statement finishes.
+            Assertions.assertTrue(profile.shouldStoreToStorage());
+            profile.writeToStorage(storage.toString());
+            Assertions.assertTrue(profile.profileHasBeenStored());
+            Profile stored = Profile.read(profile.getProfileStoragePath());
+            Assertions.assertNotNull(stored);
+            Assertions.assertEquals(details, stored.getSummaryProfile().getExecutionSummary()
+                    .getInfoString(SummaryProfile.QUERY_RETRY_DETAILS));
+            if ("PLANNING_FAILURE".equals(outcome)) {
+                Assertions.assertEquals("ERR", stored.getSummaryProfile().getSummary()
+                        .getInfoString(SummaryProfile.TASK_STATE));
+            }
+            if ("PLANNING_FAILURE".equals(outcome) || outcome.startsWith("SUCCESS")) {
+                Assertions.assertFalse(stored.getProfileByLevel().contains("abandoned-be:9060"));
+            }
+            if (outcome.endsWith("WITH_SESSION_REVERT")) {
+                Assertions.assertEquals("3", stored.getSummaryProfile().getExecutionSummary()
+                        .getInfoString(SummaryProfile.PARALLEL_FRAGMENT_EXEC_INSTANCE));
+            }
+            profile.releaseMemory();
+        } finally {
+            backgroundStorage.shutdownNow();
+            for (ExecutionProfile execution : executions) {
+                QeProcessorImpl.INSTANCE.unregisterQuery(execution.getQueryId());
+            }
+            Profile cleanup = new Profile(true, 1, -1);
+            executions.forEach(cleanup::addExecutionProfile);
+            manager.removeProfile(cleanup);
+            manager.removeProfile(profile);
+            Config.cloud_unique_id = oldCloudUniqueId;
+            Config.max_query_retry_time = oldRetryTime;
+            connectContext.setSessionVariable(oldSession);
+            connectContext.setThreadLocalInfo();
+        }
+    }
+
+    private static TQueryProfile completeBackendReport(TUniqueId queryId) {
+        TRuntimeProfileNode pipeline = new TRuntimeProfileNode();
+        pipeline.setName("Pipeline 0");
+        pipeline.setNumChildren(0);
+        pipeline.setCounters(Collections.emptyList());
+        pipeline.setMetadata(0);
+        pipeline.setIndent(true);
+        pipeline.setInfoStrings(Collections.emptyMap());
+        pipeline.setInfoStringsDisplayOrder(Collections.emptyList());
+        pipeline.setChildCountersMap(Collections.emptyMap());
+        pipeline.setTimestamp(0);
+        TRuntimeProfileTree tree = new TRuntimeProfileTree();
+        tree.setNodes(Collections.singletonList(pipeline));
+        TDetailedReportParams report = new TDetailedReportParams();
+        report.setProfile(tree);
+        report.setIsFragmentLevel(false);
+        TQueryProfile queryProfile = new TQueryProfile();
+        queryProfile.setQueryId(queryId);
+        queryProfile.putToFragmentIdToProfile(0, Collections.singletonList(report));
+        return queryProfile;
+    }
+}

fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java

@@ -20,6 +20,7 @@ package org.apache.doris.qe;
 import org.apache.doris.analysis.StatementBase;
 import org.apache.doris.arrowflight.protocol.FlightProtocolAdapter;
 import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.EnvFactory;
 import org.apache.doris.catalog.InternalSchemaInitializer;
 import org.apache.doris.catalog.ResourceMgr;
 import org.apache.doris.common.Config;
@@ -28,31 +29,68 @@ import org.apache.doris.common.FeConstants;
 import org.apache.doris.common.IncrWindowNotReadyException;
 import org.apache.doris.common.NereidsException;
 import org.apache.doris.common.Status;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.profile.ExecutionProfile;
+import org.apache.doris.common.profile.Profile;
+import org.apache.doris.common.profile.ProfileManager;
 import org.apache.doris.common.profile.RuntimeProfile;
 import org.apache.doris.common.profile.SummaryProfile;
 import org.apache.doris.common.util.DebugUtil;
+import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.mysql.authenticate.TestLogAppender;
 import org.apache.doris.nereids.NereidsPlanner;
+import org.apache.doris.nereids.glue.LogicalPlanAdapter;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.plans.LimitPhase;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.RelationId;
+import org.apache.doris.nereids.trees.plans.distribute.DistributedPlan;
+import org.apache.doris.nereids.trees.plans.distribute.FragmentIdMapping;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalEmptyRelation;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalLimit;
 import org.apache.doris.planner.PlanFragment;
+import org.apache.doris.planner.PlanFragmentId;
 import org.apache.doris.planner.Planner;
 import org.apache.doris.planner.ResultFileSink;
+import org.apache.doris.qe.protocol.ResultSender;
+import org.apache.doris.resource.BackendSelection;
+import org.apache.doris.rpc.RpcException;
+import org.apache.doris.statistics.model.Statistics;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.TCounter;
+import org.apache.doris.thrift.TDetailedReportParams;
+import org.apache.doris.thrift.TNetworkAddress;
 import org.apache.doris.thrift.TQueryOptions;
+import org.apache.doris.thrift.TQueryProfile;
+import org.apache.doris.thrift.TRuntimeProfileNode;
+import org.apache.doris.thrift.TRuntimeProfileTree;
 import org.apache.doris.thrift.TStatusCode;
 import org.apache.doris.thrift.TUniqueId;
+import org.apache.doris.thrift.TUnit;
 import org.apache.doris.utframe.TestWithFeService;
 
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.api.parallel.ResourceLock;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 
 import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.nio.file.Path;
+import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
@@ -351,6 +389,658 @@ public class StmtExecutorTest extends TestWithFeService {
         Assertions.assertNull(QeProcessorImpl.INSTANCE.getCoordinator(queryId));
     }
 
+    @Test
+    @ResourceLock("global")
+    public void testFailedDispatchRetryFinalizesProfileAfterLastAttempt() throws Exception {
+        ProfileManager manager = ProfileManager.getInstance();
+        connectContext.getSessionVariable().enableProfile = true;
+        connectContext.getSessionVariable().autoProfileThresholdMs = 10_000;
+        TUniqueId firstQueryId = new TUniqueId(0x22040L, 1L);
+        TUniqueId retryQueryId = new TUniqueId(0x22040L, 2L);
+        connectContext.setQueryId(firstQueryId);
+        connectContext.setStartTime();
+        StmtExecutor executor = new StmtExecutor(connectContext,
+                analyzeAndGetStmtByNereids("select 1", connectContext));
+        executor.getSummaryProfile().setQueryBeginTime(System.currentTimeMillis());
+        ExecutionProfile firstExecutionProfile = new ExecutionProfile(firstQueryId, Collections.emptyList());
+        ExecutionProfile retryExecutionProfile = new ExecutionProfile(retryQueryId, Collections.emptyList());
+        TQueryOptions queryOptions = new TQueryOptions();
+        queryOptions.enable_profile = true;
+
+        try {
+            Coordinator firstCoord = Mockito.mock(Coordinator.class);
+            Mockito.when(firstCoord.getQueryOptions()).thenReturn(queryOptions);
+            Mockito.when(firstCoord.getExecutionProfile()).thenReturn(firstExecutionProfile);
+            executor.getProfile().addExecutionProfile(firstExecutionProfile);
+            QeProcessorImpl.INSTANCE.registerQuery(firstQueryId, new QeProcessorImpl.QueryInfo(firstCoord));
+
+            // Dispatch failed before publishing history.
+            executor.finalizeQuery(true);
+            Assertions.assertEquals(Long.MAX_VALUE, executor.getProfile().getQueryFinishTimestamp());
+            Assertions.assertNull(manager.getExecutionProfile(firstQueryId));
+            Assertions.assertTrue(executor.getProfile().getExecutionProfiles().isEmpty());
+
+            connectContext.setQueryId(retryQueryId);
+            Coordinator retryCoord = Mockito.mock(Coordinator.class);
+            Mockito.when(retryCoord.getQueryOptions()).thenReturn(queryOptions);
+            Mockito.when(retryCoord.getExecutionProfile()).thenReturn(retryExecutionProfile);
+            executor.getProfile().addExecutionProfile(retryExecutionProfile);
+            QeProcessorImpl.INSTANCE.registerQuery(retryQueryId, new QeProcessorImpl.QueryInfo(retryCoord));
+
+            // Publish the successful retry.
+            executor.updateProfile(false);
+            Assertions.assertNotNull(manager.findProfileElementObject(executor.getProfile().getId()));
+            executor.finalizeQuery();
+
+            Assertions.assertNull(manager.getExecutionProfile(firstQueryId));
+            Assertions.assertNull(manager.getExecutionProfile(retryQueryId));
+            Assertions.assertNull(manager.findProfileElementObject(executor.getProfile().getId()));
+        } finally {
+            QeProcessorImpl.INSTANCE.unregisterQuery(firstQueryId);
+            QeProcessorImpl.INSTANCE.unregisterQuery(retryQueryId);
+            manager.cleanProfile();
+        }
+    }
+
+    @ParameterizedTest
+    @EnumSource(value = TStatusCode.class, names = {"OK", "INTERNAL_ERROR"})
+    @ResourceLock("global")
+    public void testRetryRetainsFinalProfileAboveStatementThreshold(TStatusCode finalStatus) throws Exception {
+        ProfileManager manager = ProfileManager.getInstance();
+        connectContext.getSessionVariable().enableProfile = true;
+        connectContext.getSessionVariable().autoProfileThresholdMs = 5_000;
+        TUniqueId firstQueryId = new TUniqueId(0x22040L, 3L);
+        TUniqueId retryQueryId = new TUniqueId(0x22040L, 4L);
+        connectContext.setQueryId(firstQueryId);
+        connectContext.setStartTime();
+        StmtExecutor executor = new StmtExecutor(connectContext,
+                analyzeAndGetStmtByNereids("select 1", connectContext));
+        long queryBeginTime = System.currentTimeMillis() - 6_100;
+        executor.getSummaryProfile().setQueryBeginTime(queryBeginTime);
+        ExecutionProfile firstExecutionProfile = new ExecutionProfile(firstQueryId, Collections.emptyList());
+        ExecutionProfile retryExecutionProfile = new ExecutionProfile(retryQueryId, Collections.emptyList());
+        TQueryOptions queryOptions = new TQueryOptions();
+        queryOptions.enable_profile = true;
+
+        try {
+            Coordinator firstCoord = Mockito.mock(Coordinator.class);
+            Mockito.when(firstCoord.getQueryOptions()).thenReturn(queryOptions);
+            Mockito.when(firstCoord.getExecutionProfile()).thenReturn(firstExecutionProfile);
+            executor.getProfile().addExecutionProfile(firstExecutionProfile);
+            QeProcessorImpl.INSTANCE.registerQuery(firstQueryId, new QeProcessorImpl.QueryInfo(firstCoord));
+
+            // Fetch failed after publishing history.
+            executor.updateProfile(false);
+            Assertions.assertNotNull(manager.findProfileElementObject(executor.getProfile().getId()));
+            executor.finalizeQuery(true);
+            Assertions.assertNull(manager.findProfileElementObject(executor.getProfile().getId()));
+            Assertions.assertNull(manager.getExecutionProfile(firstQueryId));
+            Assertions.assertTrue(executor.getProfile().getExecutionProfiles().isEmpty());
+            Assertions.assertEquals(Long.MAX_VALUE, executor.getProfile().getQueryFinishTimestamp());
+            Assertions.assertEquals(queryBeginTime, executor.getSummaryProfile().getQueryBeginTime());
+
+            connectContext.setQueryId(retryQueryId);
+            Coordinator retryCoord = Mockito.mock(Coordinator.class);
+            Mockito.when(retryCoord.getQueryOptions()).thenReturn(queryOptions);
+            Mockito.when(retryCoord.getExecutionProfile()).thenReturn(retryExecutionProfile);
+            Mockito.when(retryCoord.getExecStatus()).thenReturn(new Status(finalStatus, "final attempt"));
+            executor.setCoord(retryCoord);
+            executor.getProfile().addExecutionProfile(retryExecutionProfile);
+            QeProcessorImpl.INSTANCE.registerQuery(retryQueryId, new QeProcessorImpl.QueryInfo(retryCoord));
+            executor.updateProfile(false);
+            Assertions.assertNull(manager.findProfileElementObject(DebugUtil.printId(firstQueryId)));
+            Assertions.assertNotNull(manager.findProfileElementObject(executor.getProfile().getId()));
+            executor.finalizeQuery();
+            Assertions.assertNull(manager.findProfileElementObject(DebugUtil.printId(firstQueryId)));
+            Assertions.assertNotNull(manager.findProfileElementObject(DebugUtil.printId(retryQueryId)));
+            Assertions.assertNotEquals(Long.MAX_VALUE, executor.getProfile().getQueryFinishTimestamp());
+            Assertions.assertEquals(Collections.singletonList(retryExecutionProfile),
+                    executor.getProfile().getExecutionProfiles());
+            // Total duration is between one and two thresholds.
+            long durationMs = executor.getProfile().getQueryFinishTimestamp() - queryBeginTime;
+            Assertions.assertTrue(durationMs >= 5_000 && durationMs < 10_000);
+        } finally {
+            QeProcessorImpl.INSTANCE.unregisterQuery(firstQueryId);
+            QeProcessorImpl.INSTANCE.unregisterQuery(retryQueryId);
+            manager.cleanProfile();
+        }
+    }
+
+    @Test
+    @ResourceLock("global")
+    public void testSuccessfulRetryRendersMergedProfileAndActualRows(@TempDir Path profileDirectory) throws Exception {
+        ProfileManager manager = ProfileManager.getInstance();
+        connectContext.getSessionVariable().enableProfile = true;
+        connectContext.getSessionVariable().autoProfileThresholdMs = 0;
+        TUniqueId firstQueryId = new TUniqueId(0x22040L, 5L);
+        TUniqueId retryQueryId = new TUniqueId(0x22040L, 6L);
+        connectContext.setQueryId(firstQueryId);
+        connectContext.setStartTime();
+        StmtExecutor executor = new StmtExecutor(connectContext,
+                analyzeAndGetStmtByNereids("select 1", connectContext));
+        executor.getSummaryProfile().setQueryBeginTime(System.currentTimeMillis());
+        Profile profile = executor.getProfile();
+        Statistics statistics = new Statistics(0, Collections.emptyMap());
+        PhysicalEmptyRelation physicalPlan = new PhysicalEmptyRelation(new RelationId(0), Collections.emptyList(),
+                Optional.empty(), null, PhysicalProperties.ANY, statistics);
+        int planNodeId = physicalPlan.getId();
+        profile.setPhysicalPlan(physicalPlan);
+        ExecutionProfile failedExecution = new ExecutionProfile(firstQueryId, Collections.singletonList(0));
+        ExecutionProfile successfulExecution = new ExecutionProfile(retryQueryId, Collections.singletonList(0));
+        TQueryOptions queryOptions = new TQueryOptions();
+        queryOptions.enable_profile = true;
+
+        try {
+            Coordinator firstCoord = Mockito.mock(Coordinator.class);
+            Mockito.when(firstCoord.getQueryOptions()).thenReturn(queryOptions);
+            Mockito.when(firstCoord.getExecutionProfile()).thenReturn(failedExecution);
+            profile.addExecutionProfile(failedExecution);
+            QeProcessorImpl.INSTANCE.registerQuery(firstQueryId, new QeProcessorImpl.QueryInfo(firstCoord));
+            Assertions.assertTrue(failedExecution.updateProfile(
+                    createBackendRowsProfile(firstQueryId, planNodeId, 1000),
+                    new TNetworkAddress("127.0.0.1", 9060), true).ok());
+            executor.updateProfile(false);
+            executor.finalizeQuery(true);
+
+            connectContext.setQueryId(retryQueryId);
+            Coordinator retryCoord = Mockito.mock(Coordinator.class);
+            Mockito.when(retryCoord.getQueryOptions()).thenReturn(queryOptions);
+            Mockito.when(retryCoord.getExecutionProfile()).thenReturn(successfulExecution);
+            Mockito.when(retryCoord.getExecStatus()).thenReturn(Status.OK);
+            executor.setCoord(retryCoord);
+            profile.addExecutionProfile(successfulExecution);
+            QeProcessorImpl.INSTANCE.registerQuery(retryQueryId, new QeProcessorImpl.QueryInfo(retryCoord));
+            successfulExecution.addFragmentBackend(0, 1L);
+            successfulExecution.addFragmentBackend(0, 2L);
+            Assertions.assertTrue(successfulExecution.updateProfile(
+                    createBackendRowsProfile(retryQueryId, planNodeId, 17),
+                    new TNetworkAddress("127.0.0.1", 9060), true).ok());
+            Assertions.assertTrue(successfulExecution.updateProfile(
+                    createBackendRowsProfile(retryQueryId, planNodeId, 25),
+                    new TNetworkAddress("127.0.0.2", 9060), true).ok());
+            Assertions.assertTrue(successfulExecution.isCompleted());
+            executor.updateProfile(false);
+            executor.finalizeQuery();
+            Assertions.assertNotNull(manager.findProfileElementObject(profile.getId()));
+
+            String profileText = profile.getProfileByLevel();
+            Assertions.assertTrue(profileText.contains("MergedProfile:"));
+            Assertions.assertTrue(profileText.contains("Pipeline 0(instance_num=2)"));
+            Assertions.assertTrue(profileText.contains("RowsProduced: sum 42"));
+            Assertions.assertFalse(profileText.contains("build merged simple profile failed"));
+            Assertions.assertFalse(profileText.contains("DetailProfile(" + DebugUtil.printId(firstQueryId) + ")"));
+            Assertions.assertTrue(profileText.contains("DetailProfile(" + DebugUtil.printId(retryQueryId) + ")"));
+            Assertions.assertEquals(42L, profile.rowsProducedMap.get(String.valueOf(planNodeId)));
+            Assertions.assertEquals(42L, statistics.getActualRowCount());
+
+            profile.writeToStorage(profileDirectory.toString());
+            Profile storedProfile = Profile.read(profile.getProfileStoragePath());
+            Assertions.assertNotNull(storedProfile);
+            String storedText = storedProfile.getProfileByLevel();
+            Assertions.assertTrue(storedText.contains("MergedProfile:"));
+            Assertions.assertTrue(storedText.contains("RowsProduced: sum 42"));
+            Assertions.assertFalse(storedText.contains("DetailProfile(" + DebugUtil.printId(firstQueryId) + ")"));
+        } finally {
+            QeProcessorImpl.INSTANCE.unregisterQuery(firstQueryId);
+            QeProcessorImpl.INSTANCE.unregisterQuery(retryQueryId);
+            manager.removeProfile(profile);
+        }
+    }
+
+    private interface CoordinatorConfigurer {
+        void configure(Coordinator coordinator) throws Exception;
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"SUCCESS", "EXHAUSTED", "PLANNING_FAILURE", "RPC", "DIRECT",
+            "FLIGHT_SUCCESS", "FLIGHT_DEFERRED", "FETCH_FAILURE_SAME_ID", "FETCH_FAILURE_NEW_ID", "FETCH_FAILURE_PLANNING"})
+    @ResourceLock("global")
+    public void testCloudReplanDispatchProfileLifecycle(String outcome) throws Exception {
+        String oldCloudUniqueId = Config.cloud_unique_id;
+        int oldRetryTime = Config.max_query_retry_time;
+        SessionVariable originalSession = connectContext.getSessionVariable();
+        SessionVariable session = new SessionVariable();
+        session.enableProfile = true;
+        session.profileLevel = 3;
+        session.autoProfileThresholdMs = 0;
+        session.enableSqlCache = false;
+        session.cloudCluster = "profile-replan-test";
+        ConnectContext context = Mockito.spy(connectContext);
+        boolean flight = outcome.startsWith("FLIGHT_");
+        FlightProtocolAdapter flightAdapter = flight ? new FlightProtocolAdapter("profile-replan-test") : null;
+        if (flight) {
+            Mockito.doReturn(flightAdapter).when(context).getProtocolAdapter();
+            Mockito.doAnswer(invocation -> flightAdapter.returnsResultFromLocal(context))
+                    .when(context).isReturnResultFromLocal();
+        }
+        context.setSessionVariable(session);
+        Mockito.doReturn(Mockito.mock(ResultSender.class)).when(context).getResultSender();
+        Mockito.doReturn(connectContext.getComputeGroup()).when(context).getComputeGroup();
+        context.setStartTime();
+        TUniqueId firstQueryId = new TUniqueId(0x22040L, 7L);
+        context.setQueryId(firstQueryId);
+        StmtExecutor executor = Mockito.spy(new StmtExecutor(context, "select 1"));
+        Method parse = StmtExecutor.class.getDeclaredMethod("parseByNereids");
+        parse.setAccessible(true);
+        parse.invoke(executor);
+        LogicalPlanAdapter statement = (LogicalPlanAdapter) executor.getParsedStmt();
+        statement.setResultExprs(Collections.emptyList());
+        statement.setColLabels(Lists.newArrayList());
+        Assertions.assertTrue(executor.isProfileSafeStmt());
+        Profile profile = executor.getProfile();
+        profile.getSummaryProfile().setQueryBeginTime(context.getStartTime());
+        ProfileManager manager = ProfileManager.getInstance();
+        boolean fetchFailure = outcome.startsWith("FETCH_FAILURE_");
+        Planner planner = fetchFailure ? Mockito.mock(NereidsPlanner.class) : Mockito.mock(Planner.class);
+        Statistics failedStatistics = Mockito.mock(Statistics.class);
+        Statistics successfulStatistics = Mockito.mock(Statistics.class);
+        Mockito.when(failedStatistics.printColumnStats()).thenReturn("abandoned-plan-column-statistics\n");
+        Mockito.when(successfulStatistics.printColumnStats()).thenReturn("successful-plan-column-statistics\n");
+        Field plannerField = StmtExecutor.class.getDeclaredField("planner");
+        plannerField.setAccessible(true);
+        Method handle = StmtExecutor.class.getDeclaredMethod("handleQueryWithRetry", TUniqueId.class);
+        handle.setAccessible(true);
+        List<ExecutionProfile> executions = Lists.newArrayList();
+        List<PhysicalEmptyRelation> plans = Lists.newArrayList();
+        AtomicInteger attempts = new AtomicInteger();
+        AtomicInteger failedAttemptCallbacks = new AtomicInteger();
+        Mockito.doAnswer(invocation -> {
+            TUniqueId queryId = invocation.getArgument(0);
+            context.setQueryId(queryId);
+            context.getProtocolAdapter().beforeAttempt(context);
+            context.setStartTime();
+            profile.getSummaryProfile().setQueryBeginTime(context.getStartTime());
+            if (attempts.incrementAndGet() > 1) {
+                Assertions.assertEquals(Long.MAX_VALUE, profile.getQueryFinishTimestamp());
+                Assertions.assertNull(manager.getExecutionProfile(firstQueryId));
+                Assertions.assertTrue(profile.getExecutionProfiles().isEmpty());
+                Assertions.assertNull(QeProcessorImpl.INSTANCE.getCoordinator(firstQueryId));
+                Assertions.assertEquals(1, failedAttemptCallbacks.get());
+                if (fetchFailure) {
+                    Assertions.assertNull(profile.getPhysicalPlan());
+                    Assertions.assertTrue(profile.rowsProducedMap.isEmpty());
+                }
+                if ("PLANNING_FAILURE".equals(outcome) || "FETCH_FAILURE_PLANNING".equals(outcome)) {
+                    context.getState().setError("terminal planning failure");
+                    throw new UserException("terminal planning failure");
+                }
+            }
+            if (fetchFailure) {
+                int relationId = attempts.get() == 1 || "FETCH_FAILURE_SAME_ID".equals(outcome) ? 1 : 2;
+                PhysicalEmptyRelation plan = new PhysicalEmptyRelation(new RelationId(relationId),
+                        Collections.emptyList(), Optional.empty(), null, PhysicalProperties.ANY,
+                        attempts.get() == 1 ? failedStatistics : successfulStatistics);
+                plans.add(plan);
+                FragmentIdMapping<DistributedPlan> distributedPlans = null;
+                if (attempts.get() == 1) {
+                    DistributedPlan distributedPlan = Mockito.mock(DistributedPlan.class);
+                    Mockito.when(distributedPlan.toString(Mockito.anyInt())).thenReturn("abandoned-distributed-plan");
+                    distributedPlans = new FragmentIdMapping<>();
+                    distributedPlans.put(new PlanFragmentId(0), distributedPlan);
+                    context.getBackendSelectionProfile().recordQuerySelection(
+                            new BackendSelection.SelectionHint("abandoned-backend-selection",
+                                    BackendSelection.Mode.PREFER, "test"),
+                            BackendSelection.QuerySelectionResult.PREFERRED_HIT);
+                }
+                Mockito.when(((NereidsPlanner) planner).getDistributedPlans()).thenReturn(distributedPlans);
+                Mockito.when(((NereidsPlanner) planner).getPhysicalPlan()).thenReturn(plan);
+                Mockito.when(((NereidsPlanner) planner).getPhysicalRelations())
+                        .thenReturn(Collections.singletonList(plan));
+            }
+            plannerField.set(executor, planner);
+            try {
+                handle.invoke(executor, queryId);
+            } catch (InvocationTargetException e) {
+                context.getState().setError(e.getCause().getMessage());
+                throw e.getCause();
+            }
+            return null;
+        }).when(executor).execute(Mockito.any(TUniqueId.class));
+        EnvFactory factory = Mockito.mock(EnvFactory.class);
+        CoordinatorConfigurer configureCoordinator = coordinator -> {
+            TUniqueId queryId = context.queryId();
+            ExecutionProfile execution = new ExecutionProfile(queryId, Collections.singletonList(0));
+            executions.add(execution);
+            TQueryOptions options = new TQueryOptions();
+            options.enable_profile = true;
+            Mockito.when(coordinator.getQueryOptions()).thenReturn(options);
+            Mockito.when(coordinator.getExecutionProfile()).thenReturn(execution);
+            Mockito.when(coordinator.getExecStatus()).thenReturn(Status.OK);
+            Mockito.when(coordinator.mustOutliveDispatch()).thenReturn("FLIGHT_DEFERRED".equals(outcome));
+            Mockito.when(coordinator.getNext()).thenAnswer(fetch -> {
+                if (fetchFailure && executions.size() == 1) {
+                    String failedText = profile.getProfileByLevel();
+                    Assertions.assertTrue(failedText.contains("abandoned-plan-column-statistics"));
+                    Assertions.assertTrue(failedText.contains("abandoned-distributed-plan"));
+                    Assertions.assertTrue(failedText.contains("abandoned-backend-selection"));
+                    Assertions.assertEquals(1000L,
+                            profile.rowsProducedMap.get(String.valueOf(plans.get(0).getId())));
+                    throw new UserException(SystemInfoService.ERROR_E230);
+                }
+                return new RowBatch();
+            });
+            Mockito.doAnswer(dispatch -> {
+                Assertions.assertSame(execution, manager.getExecutionProfile(queryId));
+                Assertions.assertSame(coordinator, QeProcessorImpl.INSTANCE.getCoordinator(queryId));
+                if (fetchFailure) {
+                    execution.addFragmentBackend(0, 1L);
+                    Assertions.assertTrue(execution.updateProfile(
+                            createBackendRowsProfile(queryId, plans.get(plans.size() - 1).getId(),
+                                    executions.size() == 1 ? 1000 : 42),
+                            new TNetworkAddress("127.0.0.1", 9060), true).ok());
+                }
+                if (executions.size() == 1) {
+                    QeProcessorImpl.INSTANCE.registerQueryFinishCallback(DebugUtil.printId(queryId),
+                            failedAttemptCallbacks::incrementAndGet);
+                    if (flight) {
+                        Assertions.assertFalse(context.isReturnResultFromLocal());
+                        Assertions.assertFalse(execution.isCompleted());
+                    }
+                    if (fetchFailure) {
+                        return null;
+                    }
+                    if ("RPC".equals(outcome)) {
+                        throw new RpcException("test-be", SystemInfoService.NO_SCAN_NODE_BACKEND_AVAILABLE_MSG);
+                    }
+                    throw new UserException(SystemInfoService.NO_SCAN_NODE_BACKEND_AVAILABLE_MSG);
+                }
+                return null;
+            }).when(coordinator).exec();
+        };
+        Mockito.when(factory.createCoordinator(Mockito.eq(context), Mockito.eq(planner), Mockito.any()))
+                .thenAnswer(invocation -> {
+                    Coordinator coordinator = Mockito.mock(Coordinator.class);
+                    configureCoordinator.configure(coordinator);
+                    return coordinator;
+                });
+
+        try (MockedStatic<EnvFactory> factories = Mockito.mockStatic(EnvFactory.class);
+                MockedConstruction<NereidsCoordinator> coordinators = Mockito.mockConstruction(NereidsCoordinator.class,
+                        (coordinator, construction) -> configureCoordinator.configure(coordinator))) {
+            factories.when(EnvFactory::getInstance).thenReturn(factory);
+            Config.cloud_unique_id = "profile-replan-test";
+            Config.max_query_retry_time = "EXHAUSTED".equals(outcome) ? 0 : 1;
+            context.setThreadLocalInfo();
+            if ("SUCCESS".equals(outcome) || flight
+                    || (fetchFailure && !"FETCH_FAILURE_PLANNING".equals(outcome))) {
+                executor.queryRetry(firstQueryId);
+                Assertions.assertEquals(2, attempts.get());
+                ExecutionProfile successful = executions.get(1);
+                Assertions.assertEquals(Collections.singletonList(successful), profile.getExecutionProfiles());
+                Assertions.assertEquals(DebugUtil.printId(successful.getQueryId()), profile.getId());
+                Assertions.assertNotNull(manager.findProfileElementObject(profile.getId()));
+                Assertions.assertNull(manager.findProfileElementObject(DebugUtil.printId(firstQueryId)));
+                Assertions.assertNull(manager.getExecutionProfile(firstQueryId));
+                if (flight) {
+                    Assertions.assertEquals(1, failedAttemptCallbacks.get());
+                    Assertions.assertEquals(Long.MAX_VALUE, profile.getQueryFinishTimestamp());
+                    Assertions.assertSame(successful, manager.getExecutionProfile(successful.getQueryId()));
+                    Assertions.assertSame(executor.getCoord(),
+                            QeProcessorImpl.INSTANCE.getCoordinator(successful.getQueryId()));
+                    Assertions.assertEquals("FLIGHT_DEFERRED".equals(outcome), executor.isDeferredForArrowFlight());
+                    if (executor.isDeferredForArrowFlight()) {
+                        JsonObject pending = JsonParser.parseString(profile.getSummaryProfile().getExecutionSummary()
+                                .getInfoString(SummaryProfile.QUERY_RETRY_DETAILS)).getAsJsonObject();
+                        Assertions.assertEquals(1, pending.size());
+                        Assertions.assertFalse(pending.has(DebugUtil.printId(successful.getQueryId())));
+                        flightAdapter.closeDeferredExecutors();
+                    } else {
+                        executor.finalizeQuery();
+                    }
+                    Assertions.assertEquals(1, failedAttemptCallbacks.get());
+                }
+                Assertions.assertTrue(successful.getQueryFinishTime() > 0);
+                Assertions.assertEquals(TimeUtils.longToTimeString(context.getStartTime()),
+                        profile.getSummaryProfile().getSummary().getInfoString(SummaryProfile.START_TIME));
+                if (fetchFailure) {
+                    String finalText = profile.getProfileByLevel();
+                    Assertions.assertTrue(finalText.contains("PhysicalPlan:"));
+                    Assertions.assertTrue(finalText.contains("successful-plan-column-statistics"));
+                    Assertions.assertFalse(finalText.contains("abandoned-plan-column-statistics"));
+                    Assertions.assertFalse(finalText.contains("abandoned-distributed-plan"));
+                    Assertions.assertFalse(finalText.contains("abandoned-backend-selection"));
+                    Assertions.assertFalse(profile.rowsProducedMap.containsKey(String.valueOf(plans.get(0).getId())));
+                    Assertions.assertEquals(42L, profile.rowsProducedMap.get(String.valueOf(plans.get(1).getId())));
+                    Mockito.verify(failedStatistics, Mockito.times(1)).printColumnStats();
+                }
+            } else {
+                Class<? extends Exception> errorType = "RPC".equals(outcome) ? RpcException.class : UserException.class;
+                if ("DIRECT".equals(outcome)) {
+                    Assertions.assertThrows(errorType, () -> executor.execute(firstQueryId));
+                } else {
+                    Assertions.assertThrows(errorType, () -> executor.queryRetry(firstQueryId));
+                }
+                if ("PLANNING_FAILURE".equals(outcome) || "FETCH_FAILURE_PLANNING".equals(outcome)) {
+                    Assertions.assertEquals(2, attempts.get());
+                    Assertions.assertTrue(profile.getExecutionProfiles().isEmpty());
+                    Assertions.assertNull(manager.getExecutionProfile(firstQueryId));
+                    Assertions.assertNull(profile.getPhysicalPlan());
+                    String finalText = profile.getProfileByLevel();
+                    Assertions.assertFalse(finalText.contains("abandoned-plan-column-statistics"));
+                    Assertions.assertFalse(finalText.contains("abandoned-distributed-plan"));
+                    Assertions.assertFalse(finalText.contains("abandoned-backend-selection"));
+                } else {
+                    Assertions.assertEquals(1, attempts.get());
+                    Assertions.assertEquals(Collections.singletonList(executions.get(0)), profile.getExecutionProfiles());
+                    Assertions.assertNotNull(manager.findProfileElementObject(DebugUtil.printId(firstQueryId)));
+                    Assertions.assertSame(executions.get(0), manager.getExecutionProfile(firstQueryId));
+                    Assertions.assertTrue(executions.get(0).getQueryFinishTime() > 0);
+                }
+            }
+            int failedAttempts = "PLANNING_FAILURE".equals(outcome)
+                    || "FETCH_FAILURE_PLANNING".equals(outcome) ? 2 : 1;
+            Assertions.assertEquals(Integer.toString(failedAttempts), profile.getSummaryProfile().getExecutionSummary()
+                    .getInfoString(SummaryProfile.QUERY_RETRY_TIMES));
+            JsonObject attemptDetails = JsonParser.parseString(profile.getSummaryProfile().getExecutionSummary()
+                    .getInfoString(SummaryProfile.QUERY_RETRY_DETAILS)).getAsJsonObject();
+            Assertions.assertEquals(attempts.get(), attemptDetails.size());
+            Assertions.assertEquals("FAILED", attemptDetails.getAsJsonObject(DebugUtil.printId(firstQueryId))
+                    .get("state").getAsString());
+            JsonObject finalAttempt = attemptDetails.getAsJsonObject(DebugUtil.printId(context.queryId()));
+            Assertions.assertEquals(context.getState().getStateType() == QueryState.MysqlStateType.ERR ? "FAILED" : "SUCCEEDED",
+                    finalAttempt.get("state").getAsString());
+            Assertions.assertEquals(fetchFailure ? 1 : 0, coordinators.constructed().size());
+            Assertions.assertNotEquals(Long.MAX_VALUE, profile.getQueryFinishTimestamp());
+            Assertions.assertEquals(context.getStartTime(), profile.getSummaryProfile().getQueryBeginTime());
+            for (ExecutionProfile execution : executions) {
+                Assertions.assertNull(QeProcessorImpl.INSTANCE.getCoordinator(execution.getQueryId()));
+            }
+        } finally {
+            Config.cloud_unique_id = oldCloudUniqueId;
+            Config.max_query_retry_time = oldRetryTime;
+            connectContext.setSessionVariable(originalSession);
+            connectContext.setThreadLocalInfo();
+            for (ExecutionProfile execution : executions) {
+                QeProcessorImpl.INSTANCE.unregisterQuery(execution.getQueryId());
+            }
+            manager.removeProfile(profile);
+        }
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    @ResourceLock("global")
+    public void testInnerRetryClearsRenderedActualRows(boolean terminalFailure,
+            @TempDir Path profileDirectory) throws Exception {
+        String oldCloudUniqueId = Config.cloud_unique_id;
+        int oldRetryTime = Config.max_query_retry_time;
+        SessionVariable originalSession = connectContext.getSessionVariable();
+        SessionVariable session = new SessionVariable();
+        session.enableProfile = true;
+        session.autoProfileThresholdMs = 0;
+        session.enableSqlCache = false;
+        ConnectContext context = Mockito.spy(connectContext);
+        context.setSessionVariable(session);
+        Mockito.doReturn(Mockito.mock(ResultSender.class)).when(context).getResultSender();
+        context.getMysqlChannel().reset();
+        context.setStartTime();
+        TUniqueId firstQueryId = new TUniqueId(0x22040L, 19L);
+        context.setQueryId(firstQueryId);
+        StmtExecutor executor = new StmtExecutor(context, "select 1");
+        Method parse = StmtExecutor.class.getDeclaredMethod("parseByNereids");
+        parse.setAccessible(true);
+        parse.invoke(executor);
+        LogicalPlanAdapter statement = (LogicalPlanAdapter) executor.getParsedStmt();
+        statement.setOrigStmt(new OriginStatement("select 1", 0));
+        statement.setResultExprs(Collections.emptyList());
+        statement.setColLabels(Lists.newArrayList());
+        Profile profile = executor.getProfile();
+        profile.getSummaryProfile().setQueryBeginTime(context.getStartTime());
+        Statistics rootStatistics = new Statistics(1, Collections.emptyMap());
+        Statistics childStatistics = new Statistics(1, Collections.emptyMap());
+        PhysicalEmptyRelation leaf = new PhysicalEmptyRelation(new RelationId(0), Collections.emptyList(),
+                Optional.empty(), null, PhysicalProperties.ANY, null);
+        PhysicalLimit<Plan> child = new PhysicalLimit<>(1, 0, LimitPhase.GLOBAL, Optional.empty(), null,
+                PhysicalProperties.ANY, childStatistics, leaf);
+        PhysicalLimit<Plan> plan = new PhysicalLimit<>(1, 0, LimitPhase.GLOBAL, Optional.empty(), null,
+                PhysicalProperties.ANY, rootStatistics, child);
+        NereidsPlanner planner = Mockito.mock(NereidsPlanner.class);
+        Mockito.when(planner.getPhysicalPlan()).thenReturn(plan);
+        Field plannerField = StmtExecutor.class.getDeclaredField("planner");
+        plannerField.setAccessible(true);
+        plannerField.set(executor, planner);
+        List<ExecutionProfile> executions = Lists.newArrayList();
+        ProfileManager manager = ProfileManager.getInstance();
+        EnvFactory factory = Mockito.mock(EnvFactory.class);
+        Mockito.when(factory.createCoordinator(Mockito.eq(context), Mockito.eq(planner), Mockito.any()))
+                .thenAnswer(invocation -> {
+                    TUniqueId queryId = context.queryId();
+                    boolean firstAttempt = executions.isEmpty();
+                    ExecutionProfile execution = new ExecutionProfile(queryId, Lists.newArrayList(0, 1));
+                    executions.add(execution);
+                    Coordinator coordinator = Mockito.mock(Coordinator.class);
+                    TQueryOptions options = new TQueryOptions();
+                    options.enable_profile = true;
+                    Mockito.when(coordinator.getQueryOptions()).thenReturn(options);
+                    Mockito.when(coordinator.getExecutionProfile()).thenReturn(execution);
+                    Mockito.when(coordinator.getExecStatus()).thenReturn(Status.OK);
+                    Mockito.doAnswer(dispatch -> {
+                        if (!firstAttempt) {
+                            Assertions.assertSame(plan, profile.getPhysicalPlan());
+                            Assertions.assertTrue(profile.rowsProducedMap.isEmpty());
+                            Assertions.assertEquals(-1L, rootStatistics.getActualRowCount());
+                            Assertions.assertEquals(-1L, childStatistics.getActualRowCount());
+                            Assertions.assertNull(manager.getExecutionProfile(firstQueryId));
+                            Assertions.assertNull(QeProcessorImpl.INSTANCE.getCoordinator(firstQueryId));
+                        }
+                        execution.addFragmentBackend(0, 1L);
+                        execution.addFragmentBackend(1, 1L);
+                        TNetworkAddress backend = new TNetworkAddress("127.0.0.1", 9060);
+                        Assertions.assertTrue(execution.updateProfile(
+                                createBackendRowsProfile(queryId, plan.getId(), firstAttempt ? 1000 : 42),
+                                backend, true).ok());
+                        if (firstAttempt) {
+                            TQueryProfile childReport = createBackendRowsProfile(queryId, child.getId(), 500);
+                            childReport.setFragmentIdToProfile(Collections.singletonMap(1,
+                                    childReport.getFragmentIdToProfile().get(0)));
+                            Assertions.assertTrue(execution.updateProfile(childReport, backend, true).ok());
+                        }
+                        return null;
+                    }).when(coordinator).exec();
+                    Mockito.when(coordinator.getNext()).thenAnswer(fetch -> {
+                        if (firstAttempt) {
+                            String firstText = profile.getProfileByLevel();
+                            Assertions.assertTrue(firstText.contains("actualRows=1000"));
+                            Assertions.assertTrue(firstText.contains("actualRows=500"));
+                            Assertions.assertEquals(500L, childStatistics.getActualRowCount());
+                            throw new RpcException("test-be", "fetch failed before sending results");
+                        }
+                        if (terminalFailure) {
+                            throw new RpcException("test-be", "terminal fetch failure");
+                        }
+                        return new RowBatch();
+                    });
+                    return coordinator;
+                });
+        try (MockedStatic<EnvFactory> factories = Mockito.mockStatic(EnvFactory.class)) {
+            factories.when(EnvFactory::getInstance).thenReturn(factory);
+            Config.cloud_unique_id = "";
+            Config.max_query_retry_time = 1;
+            context.setThreadLocalInfo();
+            Method handle = StmtExecutor.class.getDeclaredMethod("handleQueryWithRetry", TUniqueId.class);
+            handle.setAccessible(true);
+            if (terminalFailure) {
+                InvocationTargetException failure = Assertions.assertThrows(InvocationTargetException.class,
+                        () -> handle.invoke(executor, firstQueryId));
+                Assertions.assertInstanceOf(RpcException.class, failure.getCause());
+            } else {
+                handle.invoke(executor, firstQueryId);
+            }
+            Assertions.assertEquals(2, executions.size());
+            Assertions.assertEquals(Collections.singletonList(executions.get(1)), profile.getExecutionProfiles());
+            Assertions.assertFalse(executions.get(1).isCompleted());
+            String details = profile.getSummaryProfile().getExecutionSummary()
+                    .getInfoString(SummaryProfile.QUERY_RETRY_DETAILS);
+            JsonObject attempts = JsonParser.parseString(details).getAsJsonObject();
+            Assertions.assertEquals(2, attempts.size());
+            Assertions.assertEquals("FAILED", attempts.getAsJsonObject(DebugUtil.printId(firstQueryId))
+                    .get("state").getAsString());
+            JsonObject last = attempts.getAsJsonObject(DebugUtil.printId(context.queryId()));
+            Assertions.assertEquals(terminalFailure ? "FAILED" : "SUCCEEDED", last.get("state").getAsString());
+            Assertions.assertTrue(last.get("durationMs").getAsLong() >= 0);
+            String finalText = profile.getProfileByLevel();
+            Assertions.assertTrue(finalText.contains("QueryRetryTimes: " + (terminalFailure ? 2 : 1)));
+            Assertions.assertTrue(finalText.contains("actualRows=42"));
+            Assertions.assertFalse(finalText.contains("actualRows=1000"));
+            Assertions.assertFalse(finalText.contains("actualRows=500"));
+            Assertions.assertEquals(42L, rootStatistics.getActualRowCount());
+            Assertions.assertEquals(-1L, childStatistics.getActualRowCount());
+            Assertions.assertFalse(profile.rowsProducedMap.containsKey(String.valueOf(child.getId())));
+            profile.writeToStorage(profileDirectory.toString());
+            Profile stored = Profile.read(profile.getProfileStoragePath());
+            Assertions.assertNotNull(stored);
+            Assertions.assertTrue(stored.getProfileByLevel().contains("QueryRetryTimes: " + (terminalFailure ? 2 : 1)));
+            Assertions.assertEquals(details, stored.getSummaryProfile().getExecutionSummary()
+                    .getInfoString(SummaryProfile.QUERY_RETRY_DETAILS));
+            Assertions.assertFalse(stored.getProfileByLevel().contains("actualRows=500"));
+        } finally {
+            Config.cloud_unique_id = oldCloudUniqueId;
+            Config.max_query_retry_time = oldRetryTime;
+            connectContext.setSessionVariable(originalSession);
+            connectContext.setThreadLocalInfo();
+            for (ExecutionProfile execution : executions) {
+                QeProcessorImpl.INSTANCE.unregisterQuery(execution.getQueryId());
+            }
+            manager.removeProfile(profile);
+        }
+    }
+
+    private static TQueryProfile createBackendRowsProfile(TUniqueId queryId, int planNodeId, long rowsProduced) {
+        TRuntimeProfileNode pipeline = createProfileNode("Pipeline 0", 1);
+        TRuntimeProfileNode task = createProfileNode("PipelineTask 0", 1);
+        TRuntimeProfileNode operator = createProfileNode("AGGREGATION_OPERATOR (nereids_id=" + planNodeId + ")", 1);
+        operator.setMetadata(planNodeId);
+        TRuntimeProfileNode commonCounters = createProfileNode("CommonCounters", 0);
+        TCounter rows = new TCounter("RowsProduced", TUnit.UNIT, rowsProduced);
+        rows.setLevel(1);
+        commonCounters.setCounters(Collections.singletonList(rows));
+        commonCounters.setChildCountersMap(Collections.singletonMap(RuntimeProfile.ROOT_COUNTER,
+                Collections.singleton("RowsProduced")));
+        TRuntimeProfileTree tree = new TRuntimeProfileTree();
+        tree.setNodes(Lists.newArrayList(pipeline, task, operator, commonCounters));
+        TDetailedReportParams report = new TDetailedReportParams();
+        report.setProfile(tree);
+        report.setIsFragmentLevel(false);
+        TQueryProfile queryProfile = new TQueryProfile();
+        queryProfile.setQueryId(queryId);
+        queryProfile.putToFragmentIdToProfile(0, Collections.singletonList(report));
+        return queryProfile;
+    }
+
+    private static TRuntimeProfileNode createProfileNode(String name, int children) {
+        TRuntimeProfileNode node = new TRuntimeProfileNode();
+        node.setName(name);
+        node.setNumChildren(children);
+        node.setCounters(Collections.emptyList());
+        node.setMetadata(0);
+        node.setIndent(true);
+        node.setInfoStrings(Collections.emptyMap());
+        node.setInfoStringsDisplayOrder(Collections.emptyList());
+        node.setChildCountersMap(Collections.emptyMap());
+        node.setTimestamp(0);
+        return node;
+    }
+
     @Test
     public void testKill() throws Exception {
         StmtExecutor stmtExecutor = new StmtExecutor(connectContext, "");