From 04da0ca6624b31cf0cd0c4835b0ff27ba90520c7 Mon Sep 17 00:00:00 2001 From: linrrarity Date: Mon, 12 Oct 2026 10:38:54 +0800 Subject: [PATCH] [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. --- .../apache/doris/common/profile/Profile.java | 29 +- .../doris/common/profile/ProfileManager.java | 155 ++-- .../doris/common/profile/SummaryProfile.java | 227 +++++- .../rest/manager/QueryProfileAction.java | 2 +- .../insert/AbstractInsertExecutor.java | 1 + .../org/apache/doris/qe/StmtExecutor.java | 108 ++- .../doris/common/profile/AutoProfileTest.java | 4 +- .../common/profile/ProfileManagerTest.java | 110 ++- .../doris/common/profile/ProfileTest.java | 115 ++- .../common/profile/SummaryProfileTest.java | 202 +++++ .../insert/CloudInsertProfileRetryTest.java | 424 +++++++++++ .../org/apache/doris/qe/StmtExecutorTest.java | 690 ++++++++++++++++++ 12 files changed, 1952 insertions(+), 115 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/CloudInsertProfileRetryTest.java diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/profile/Profile.java b/fe/fe-core/src/main/java/org/apache/doris/common/profile/Profile.java index f2f35da51bf..b6da0447c31 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/common/profile/Profile.java +++ b/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 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()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/profile/ProfileManager.java b/fe/fe-core/src/main/java/org/apache/doris/common/profile/ProfileManager.java index 232096be42b..5717899de04 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/common/profile/ProfileManager.java +++ b/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 = 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 queryIdToProfileMap; + final Map 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 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> result = Lists.newArrayList(); readLock.lock(); try { - PriorityQueue queueIdDeque = getProfileOrderByQueryFinishTimeDesc(); - while (!queueIdDeque.isEmpty()) { - ProfileElement profileElement = queueIdDeque.poll(); + PriorityQueue profileQueue = getProfileOrderByQueryFinishTimeDesc(); + while (!profileQueue.isEmpty()) { + ProfileElement profileElement = profileQueue.poll(); List row = Lists.newArrayList(); for (String str : columnNameList) { row.add(getProfileInfoString(profileElement, str)); @@ -322,7 +318,7 @@ public class ProfileManager extends MasterDaemon { } private List> 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> futures = createFetchRealTimeProfileTasks(id, "profile"); + public String getProfile(String profileId) { + List> 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 entry : queryIdToProfileMap.entrySet()) { + for (Map.Entry 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 getProfilesNeedStore() { List 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 queryIdToBeRemoved = Lists.newArrayList(); + List 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 queryIdToBeRemoved = Lists.newArrayList(); + List 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 getProfileOrderByQueryFinishTimeDesc() { readLock.lock(); try { - PriorityQueue queryIdDeque = new PriorityQueue<>(Comparator.comparingLong( + PriorityQueue 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 getProfileOrderByQueryFinishTime() { readLock.lock(); try { - PriorityQueue queryIdDeque = new PriorityQueue<>(Comparator.comparingLong( + PriorityQueue 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 getProfileOrderByQueryStartTime() { readLock.lock(); try { - PriorityQueue queryIdDeque = new PriorityQueue<>(Comparator.comparingLong( + PriorityQueue 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 queueIdDeque = getProfileOrderByQueryFinishTimeDesc(); - ProfileElement profileElement = queueIdDeque.poll(); + PriorityQueue 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 queueIdDeque = getProfileOrderByQueryFinishTime(); + PriorityQueue 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 queueIdDeque = getProfileOrderByQueryFinishTimeDesc(); - while (!queueIdDeque.isEmpty() && limit > 0) { - ProfileElement profileElement = queueIdDeque.poll(); + PriorityQueue profileQueue = getProfileOrderByQueryFinishTimeDesc(); + while (!profileQueue.isEmpty() && limit > 0) { + ProfileElement profileElement = profileQueue.poll(); Map infoStrings = profileElement.infoStrings; if (infoStrings.get(SummaryProfile.TASK_TYPE).equals(profileType.toString())) { List 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. diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/profile/SummaryProfile.java b/fe/fe-core/src/main/java/org/apache/doris/common/profile/SummaryProfile.java index 0a595d0fbfe..67f5a42fb0f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/common/profile/SummaryProfile.java +++ b/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 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 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> rpcPhase1Latency; + @ResetOnRetry(ResetScope.EXECUTION) private Map> rpcPhase2Latency; + @ResetOnRetry(ResetScope.PLAN) private Map 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> RETRY_FIELD_DEFAULTS = buildRetryFieldDefaults(); + + private static Map> buildRetryFieldDefaults() { + Map> 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 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 map = Maps.newHashMap(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/QueryProfileAction.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/QueryProfileAction.java index d0aae7d1a0e..aace963b073 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/QueryProfileAction.java +++ b/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); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java index 78d6375aed3..c57b97ae39e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java +++ b/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)) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java index 3e100238f03..391f52495dd 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java +++ b/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 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()); diff --git a/fe/fe-core/src/test/java/org/apache/doris/common/profile/AutoProfileTest.java b/fe/fe-core/src/test/java/org/apache/doris/common/profile/AutoProfileTest.java index 6c310c820a6..ab4e2c1fb59 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/common/profile/AutoProfileTest.java +++ b/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())); } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/common/profile/ProfileManagerTest.java b/fe/fe-core/src/test/java/org/apache/doris/common/profile/ProfileManagerTest.java index 8ed48a2d63e..75a7c39703c 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/common/profile/ProfileManagerTest.java +++ b/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 profiles = new ArrayList<>(); + List 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 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())); } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/common/profile/ProfileTest.java b/fe/fe-core/src/test/java/org/apache/doris/common/profile/ProfileTest.java index ed231e65d30..639fc0844c8 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/common/profile/ProfileTest.java +++ b/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 diff --git a/fe/fe-core/src/test/java/org/apache/doris/common/profile/SummaryProfileTest.java b/fe/fe-core/src/test/java/org/apache/doris/common/profile/SummaryProfileTest.java index 001d63364fc..6c0e1c629c2 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/common/profile/SummaryProfileTest.java +++ b/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 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 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 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()); + } + } + } diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/CloudInsertProfileRetryTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/CloudInsertProfileRetryTest.java new file mode 100644 index 00000000000..56e0d267c85 --- /dev/null +++ b/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 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 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 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 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; + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java index 86f36224ad0..aed8110a7b3 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java +++ b/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 executions = Lists.newArrayList(); + List 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 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 factories = Mockito.mockStatic(EnvFactory.class); + MockedConstruction 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 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 child = new PhysicalLimit<>(1, 0, LimitPhase.GLOBAL, Optional.empty(), null, + PhysicalProperties.ANY, childStatistics, leaf); + PhysicalLimit 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 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 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, "");