github-actions[bot] commented on code in PR #69830:
URL: https://github.com/apache/doris/pull/69830#discussion_r4422813965
##########
fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java:
##########
@@ -18,747 +18,634 @@
package org.apache.doris.qe;
import org.apache.doris.analysis.DescriptorTable;
-import org.apache.doris.analysis.DescriptorToThriftConverter;
import org.apache.doris.analysis.StorageBackend;
import org.apache.doris.arrowflight.results.FlightSqlEndpointsLocation;
-import org.apache.doris.catalog.AIResource;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.FsBroker;
-import org.apache.doris.catalog.Resource;
+import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
-import org.apache.doris.common.MarkedCountDownLatch;
-import org.apache.doris.common.Pair;
-import org.apache.doris.common.Reference;
import org.apache.doris.common.Status;
import org.apache.doris.common.ThreadPoolManager;
import org.apache.doris.common.UserException;
import org.apache.doris.common.profile.ExecutionProfile;
-import org.apache.doris.common.profile.SummaryProfile;
import org.apache.doris.common.util.DebugUtil;
-import org.apache.doris.common.util.ListUtil;
-import org.apache.doris.datasource.scan.ExternalScanNode;
-import org.apache.doris.datasource.scan.FileQueryScanNode;
-import org.apache.doris.datasource.split.FileScanRangeCompactor;
-import org.apache.doris.load.loadv2.LoadJob;
-import org.apache.doris.metric.MetricRepo;
+import org.apache.doris.datasource.InternalCatalog;
import org.apache.doris.nereids.NereidsPlanner;
import org.apache.doris.nereids.stats.StatsErrorEstimator;
-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.TopnFilter;
-import org.apache.doris.planner.DataPartition;
+import org.apache.doris.nereids.trees.plans.distribute.PipelineDistributedPlan;
+import org.apache.doris.nereids.trees.plans.distribute.worker.BackendWorker;
+import
org.apache.doris.nereids.trees.plans.distribute.worker.DistributedPlanWorker;
+import org.apache.doris.nereids.trees.plans.distribute.worker.job.AssignedJob;
+import org.apache.doris.nereids.util.Utils;
import org.apache.doris.planner.DataSink;
-import org.apache.doris.planner.DataStreamSink;
-import org.apache.doris.planner.DictionarySink;
-import org.apache.doris.planner.ExceptNode;
-import org.apache.doris.planner.ExchangeNode;
-import org.apache.doris.planner.HashJoinNode;
-import org.apache.doris.planner.IntersectNode;
-import org.apache.doris.planner.MultiCastDataSink;
-import org.apache.doris.planner.MultiCastPlanFragment;
-import org.apache.doris.planner.OlapScanNode;
import org.apache.doris.planner.OlapTableSink;
import org.apache.doris.planner.PlanFragment;
-import org.apache.doris.planner.PlanFragmentId;
-import org.apache.doris.planner.PlanNode;
-import org.apache.doris.planner.PlanNodeId;
-import org.apache.doris.planner.Planner;
import org.apache.doris.planner.ResultFileSink;
import org.apache.doris.planner.ResultSink;
-import org.apache.doris.planner.RuntimeFilter;
-import org.apache.doris.planner.RuntimeFilterId;
import org.apache.doris.planner.ScanNode;
import org.apache.doris.planner.SchemaScanNode;
-import org.apache.doris.planner.SetOperationNode;
-import org.apache.doris.planner.SortNode;
-import org.apache.doris.planner.TVFTableSink;
-import org.apache.doris.planner.UnionNode;
-import org.apache.doris.proto.InternalService;
-import org.apache.doris.proto.InternalService.PExecPlanFragmentResult;
-import org.apache.doris.proto.InternalService.PExecPlanFragmentStartRequest;
-import org.apache.doris.proto.Types;
-import org.apache.doris.proto.Types.PUniqueId;
import org.apache.doris.qe.QueryStatisticsItem.FragmentInstanceInfo;
+import org.apache.doris.qe.runtime.LoadProcessor;
+import org.apache.doris.qe.runtime.MultiFragmentsPipelineTask;
+import org.apache.doris.qe.runtime.PipelineExecutionTask;
+import org.apache.doris.qe.runtime.PipelineExecutionTaskBuilder;
+import org.apache.doris.qe.runtime.QueryProcessor;
+import org.apache.doris.qe.runtime.SingleFragmentPipelineTask;
+import org.apache.doris.qe.runtime.ThriftPlansBuilder;
import org.apache.doris.resource.BackendSelection;
import org.apache.doris.resource.BackendSelectionManager;
import org.apache.doris.resource.workloadgroup.QueryQueue;
import org.apache.doris.resource.workloadgroup.QueueToken;
import org.apache.doris.resource.workloadgroup.WorkloadGroup;
-import org.apache.doris.rpc.BackendServiceProxy;
-import org.apache.doris.rpc.RpcException;
-import org.apache.doris.service.FrontendOptions;
import org.apache.doris.system.Backend;
-import org.apache.doris.system.SystemInfoService;
-import org.apache.doris.task.LoadEtlTask;
-import org.apache.doris.thrift.PaloInternalServiceVersion;
-import org.apache.doris.thrift.TAIResource;
-import org.apache.doris.thrift.TBrokerScanRange;
-import org.apache.doris.thrift.TDataSinkType;
-import org.apache.doris.thrift.TDescriptorTable;
import org.apache.doris.thrift.TErrorTabletInfo;
-import org.apache.doris.thrift.TEsScanRange;
-import org.apache.doris.thrift.TExternalScanRange;
-import org.apache.doris.thrift.TFileScanRange;
-import org.apache.doris.thrift.TFileScanRangeParams;
-import org.apache.doris.thrift.TFragmentInstanceReport;
import org.apache.doris.thrift.TNetworkAddress;
-import org.apache.doris.thrift.TOlapTableSink;
-import org.apache.doris.thrift.TPaloScanRange;
import org.apache.doris.thrift.TPipelineFragmentParams;
import org.apache.doris.thrift.TPipelineFragmentParamsList;
-import org.apache.doris.thrift.TPipelineInstanceParams;
import org.apache.doris.thrift.TPipelineWorkloadGroup;
-import org.apache.doris.thrift.TPlanFragment;
-import org.apache.doris.thrift.TPlanFragmentDestination;
-import org.apache.doris.thrift.TQueryGlobals;
import org.apache.doris.thrift.TQueryOptions;
import org.apache.doris.thrift.TQueryType;
import org.apache.doris.thrift.TReportExecStatusParams;
-import org.apache.doris.thrift.TResourceInfo;
-import org.apache.doris.thrift.TRuntimeFilterParams;
-import org.apache.doris.thrift.TRuntimeFilterTargetParamsV2;
-import org.apache.doris.thrift.TScanRange;
-import org.apache.doris.thrift.TScanRangeLocation;
-import org.apache.doris.thrift.TScanRangeLocations;
-import org.apache.doris.thrift.TScanRangeParams;
import org.apache.doris.thrift.TStatusCode;
import org.apache.doris.thrift.TTabletCommitInfo;
-import org.apache.doris.thrift.TTopnFilterDesc;
import org.apache.doris.thrift.TUniqueId;
-import org.apache.doris.transaction.CommitDataSerializer;
-import org.apache.doris.transaction.Transaction;
-import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import com.google.common.base.Strings;
-import com.google.common.collect.HashMultiset;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
-import com.google.common.collect.Multiset;
-import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.protobuf.ByteString;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.tuple.ImmutableTriple;
-import org.apache.commons.lang3.tuple.Triple;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
-import org.apache.thrift.TException;
-import org.apache.thrift.TSerializer;
-import org.apache.thrift.protocol.TCompactProtocol;
-import org.jetbrains.annotations.NotNull;
-import org.joda.time.DateTime;
-
-import java.security.SecureRandom;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
+
import java.util.Comparator;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Optional;
-import java.util.Random;
import java.util.Set;
-import java.util.TreeSet;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.function.Consumer;
import java.util.stream.Collectors;
+/** Coordinator */
public class Coordinator implements CoordInterface {
+ public static final ExecutorService backendRpcCallbackExecutor =
ThreadPoolManager.newDaemonProfileThreadPool(
+ 32, 100, "backend-rpc-callback", true);
+
private static final Logger LOG = LogManager.getLogger(Coordinator.class);
- public static final String localIP = FrontendOptions.getLocalHostAddress();
+ protected final CoordinatorContext coordinatorContext;
- // Random is used to shuffle instances of partitioned
- private static final Random instanceRandom = new SecureRandom();
+ protected volatile PipelineExecutionTask executionTask;
- public static ExecutorService backendRpcCallbackExecutor =
ThreadPoolManager.newDaemonProfileThreadPool(
- 32, 100, "backend-rpc-callback", true);
+ private final boolean needEnqueue;
+
+ // sql execution
+ public Coordinator(ConnectContext context,
+ NereidsPlanner planner, StatsErrorEstimator statsErrorEstimator) {
+ this.coordinatorContext = CoordinatorContext.buildForSql(planner,
this);
+ this.needEnqueue = true;
+
+ DataSink dataSink = coordinatorContext.dataSink;
+ // output to mysql or to file
+ if ((dataSink instanceof ResultSink || dataSink instanceof
ResultFileSink)) {
+ setForQuery();
+ } else {
+ setForInsert(-1L);
+ }
- // Overall status of the entire query; set to the first reported fragment
error
- // status or to CANCELLED, if Cancel() is called.
- private Status queryStatus = new Status();
-
- // save of related backends of this query
- private Map<TNetworkAddress, Long> addressToBackendID = Maps.newHashMap();
-
- protected ImmutableMap<Long, Backend> idToBackend = ImmutableMap.of();
-
- private final TDescriptorTable descTable;
- private FragmentIdMapping<DistributedPlan> distributedPlans;
-
- // scan node id -> TFileScanRangeParams
- private Map<Integer, TFileScanRangeParams> fileScanRangeParamsMap =
Maps.newHashMap();
-
- // Why do we use query global?
- // When `NOW()` function is in sql, we need only one now(),
- // but, we execute `NOW()` distributed.
- // So we make a query global value here to make one `now()` value in one
query process.
- private final TQueryGlobals queryGlobals = new TQueryGlobals();
- private TQueryOptions queryOptions;
- private TNetworkAddress coordAddress;
- // fe audit log in connected FE,if a query is forward
- // we should send the connected FE to be,
- // then be report query statistics to the connected FE
- private TNetworkAddress currentConnectFE;
-
- // protects all fields below
- private final Lock lock = new ReentrantLock();
-
- // If true, the query is done returning all results. It is possible that
the
- // coordinator still needs to wait for cleanup on remote fragments (e.g.
queries
- // with limit)
- // Once this is set to true, errors from remote fragments are ignored.
- private boolean returnedAllResults = false;
-
- // populated in computeFragmentExecParams()
- private final Map<PlanFragmentId, FragmentExecParams>
fragmentExecParamsMap = Maps.newHashMap();
-
- private final List<PlanFragment> fragments;
-
- private Map<Long, PipelineExecContexts> beToPipelineExecCtxs =
Maps.newHashMap();
- private final Set<Long> dispatchedBackendIdsForAudit =
Sets.newConcurrentHashSet();
-
- private final Map<Pair<Integer, Long>, PipelineExecContext>
pipelineExecContexts = new HashMap<>();
- private final List<PipelineExecContext> needCheckPipelineExecContexts =
Lists.newArrayList();
- private List<ResultReceiver> receivers = Lists.newArrayList();
- private ResultReceiverConsumer receiverConsumer;
- private final List<ScanNode> scanNodes;
- private int scanRangeNum = 0;
- // number of instances of this query, equals to
- // number of backends executing plan fragments on behalf of this query;
- // set in computeFragmentExecParams();
- // same as backend_exec_states_.size() after Exec()
- private final Set<TUniqueId> instanceIds = Sets.newHashSet();
-
- private int numReceivedRows = 0;
-
- private List<String> deltaUrls;
- private Map<String, String> loadCounters;
- private String trackingUrl;
- private String firstErrorMsg;
- // related txnId and label of group commit
- // Final reports race with status readers, so the transaction identity
must be safely published.
- private volatile long txnId;
- private String label;
-
- // for export
- private List<String> exportFiles;
-
- private final List<TTabletCommitInfo> commitInfos = Lists.newArrayList();
- private final List<TErrorTabletInfo> errorTabletInfos =
Lists.newArrayList();
-
- // Input parameter
- private long jobId = -1; // job which this task belongs to
- private TUniqueId queryId;
-
- // a timestamp represent the absolute timeout
- // eg, System.currentTimeMillis() + executeTimeoutS * 1000
- private long timeoutDeadline;
-
- private boolean enableShareHashTableForBroadcastJoin = false;
-
- private boolean useNereids = false;
-
- private Backend groupCommitBackend;
-
- // Runtime filter merge instance address and ID
- private TNetworkAddress runtimeFilterMergeAddr;
- private TUniqueId runtimeFilterMergeInstanceId;
- // Runtime filter ID to the target instance address of the fragment,
- // that is expected to use this runtime filter, the instance address is
not repeated
- private Map<RuntimeFilterId, List<FRuntimeFilterTargetParam>>
ridToTargetParam = Maps.newHashMap();
- // The runtime filter that expects the instance to be used
- private List<RuntimeFilter> assignedRuntimeFilters = new ArrayList<>();
- private List<TopnFilter> topnFilters = new ArrayList<>();
- // Runtime filter ID to the builder instance number
- private Map<RuntimeFilterId, Integer> ridToBuilderNum = Maps.newHashMap();
- private ConnectContext context;
-
- private StatsErrorEstimator statsErrorEstimator;
-
- // A countdown latch to mark the completion of each instance.
- // use for old pipeline
- // instance id -> dummy value
- private MarkedCountDownLatch<TUniqueId, Long> instancesDoneLatch = null;
-
- // A countdown latch to mark the completion of each fragment. use for
pipelineX
- // fragmentid -> backendid
- private MarkedCountDownLatch<Integer, Long> fragmentsDoneLatch = null;
-
- public void setGroupCommitBe(Backend backend) {
- this.groupCommitBackend = backend;
+ syncLocalShufflePlannerOption();
+ Preconditions.checkState(!planner.getFragments().isEmpty()
+ && coordinatorContext.instanceNum.get() > 0, "Fragment and
Instance can not be empty˚");
}
- public void setTWorkloadGroups(List<TPipelineWorkloadGroup>
tWorkloadGroups) {
- this.tWorkloadGroups = tWorkloadGroups;
+ // load, with JobId
+ public Coordinator(ConnectContext context,
+ NereidsPlanner planner, StatsErrorEstimator statsErrorEstimator,
long jobId) {
+ this.coordinatorContext = CoordinatorContext.buildForSql(planner,
this);
+ this.needEnqueue = true;
+
+ // we don't need to check the dataSink, Because setting jobId means
this must be a load operation
+ setForInsert(jobId);
+ syncLocalShufflePlannerOption();
+ Preconditions.checkState(!planner.getFragments().isEmpty()
+ && coordinatorContext.instanceNum.get() > 0, "Fragment and
Instance can not be empty˚");
}
- public long getNumReceivedRows() {
- return numReceivedRows;
+ // broker load
+ public Coordinator(Long jobId, TUniqueId queryId, DescriptorTable
descTable,
+ List<PlanFragment> fragments, List<PipelineDistributedPlan>
distributedPlans,
+ List<ScanNode> scanNodes, String timezone, boolean
loadZeroTolerance,
+ boolean enableProfile) {
+ this.coordinatorContext = CoordinatorContext.buildForLoad(
+ this, jobId, queryId, fragments, distributedPlans, scanNodes,
+ descTable, timezone, loadZeroTolerance, enableProfile
+ );
+ // same reason in `setForInsert`
+ this.coordinatorContext.queryOptions.setDisableFileCache(true);
+ this.needEnqueue = false;
+ syncLocalShufflePlannerOption();
+
+ Preconditions.checkState(!fragments.isEmpty()
+ && coordinatorContext.instanceNum.get() > 0, "Fragment and
Instance can not be empty˚");
}
- public List<TPipelineWorkloadGroup> getTWorkloadGroups() {
- return tWorkloadGroups;
+ private void syncLocalShufflePlannerOption() {
+ coordinatorContext.queryOptions.setEnableLocalShufflePlanner(
+ coordinatorContext.distributedPlans != null
+ && !coordinatorContext.distributedPlans.isEmpty()
+ && coordinatorContext.connectContext != null
+ &&
coordinatorContext.connectContext.getSessionVariable().isEnableLocalShufflePlanner());
}
- private List<TPipelineWorkloadGroup> tWorkloadGroups =
Lists.newArrayList();
+ @Override
+ public void exec() throws Exception {
+ Status status = getQueryStatus();
+ if (!status.ok()) {
+ throw new UserException(status.getErrorMsg());
+ }
+ enqueue(coordinatorContext.connectContext);
+
+ processTopSink(coordinatorContext,
coordinatorContext.topDistributedPlan);
+ status = getQueryStatus();
+ if (!status.ok()) {
+ throw new UserException(status.getErrorMsg());
+ }
+
+ QeProcessorImpl.INSTANCE.registerInstances(coordinatorContext.queryId,
coordinatorContext.instanceNum.get());
- private final ExecutionProfile executionProfile;
+ Map<DistributedPlanWorker, TPipelineFragmentParamsList>
workerToFragments
+ = ThriftPlansBuilder.plansToThrift(coordinatorContext);
+ executionTask = PipelineExecutionTaskBuilder.build(coordinatorContext,
workerToFragments);
+ executionTask.execute();
+ }
- private volatile QueueToken queueToken = null;
- private QueryQueue queryQueue = null;
+ public boolean isTimeout() {
+ return System.currentTimeMillis() >
coordinatorContext.timeoutDeadline.get();
+ }
- public ExecutionProfile getExecutionProfile() {
- return executionProfile;
+ @Override
+ public void cancel(Status cancelReason) {
+ if (cancelReason.ok()) {
+ throw new RuntimeException("Should use correct cancel reason, but
it is " + cancelReason);
+ }
+
+ TUniqueId queryId = coordinatorContext.queryId;
+ try {
+ Status originQueryStatus =
coordinatorContext.updateStatusIfOk(cancelReason);
+ if (!originQueryStatus.ok()) {
+ if (LOG.isDebugEnabled()) {
+ // Print an error stack here to know why send cancel again.
+ LOG.warn("Query {} already in abnormal status {}, but
received cancel again,"
+ + "so that send cancel to BE again",
+ DebugUtil.printId(queryId),
originQueryStatus.toString(),
+ new Exception("cancel failed"));
+ }
+ } else {
+ LOG.warn("Cancel execution of query {}, this is a outside
invoke, cancelReason {}",
+ DebugUtil.printId(queryId), cancelReason);
+ }
+ } finally {
+ // Publishing the status above can itself cancel a partially
initialized processor. Start the
+ // non-throwing cleanup scope before that publication so the queue
token, the scan nodes, and
+ // the final internal cancel are never skipped, even if the
publication was only half wired.
+ try {
+
coordinatorContext.getQueueToken().ifPresent(QueueToken::cancel);
+ // Scan cleanup is best-effort and must never escape: a
throwing scan would otherwise skip
+ // the remaining scans (and the caller's coordinator close),
masking the retained reason. A
+ // scan whose first stop() threw still removes its own sources
on the close-time retry
+ // because SplitAssignment.stop() is idempotent.
+ for (ScanNode scanNode : coordinatorContext.scanNodes) {
+ try {
+ scanNode.stop();
+ } catch (Throwable t) {
+ LOG.error("error happens when scannode stop during
cancel, query id: {}",
+ DebugUtil.printId(queryId), t);
+ }
+ }
+ } finally {
+ cancelInternal(cancelReason);
+ }
+ }
}
- // True if all scan node are ExternalScanNode.
- private boolean isAllExternalScan = true;
+ public QueryProcessor asQueryProcessor() {
+ return coordinatorContext.asQueryProcessor();
+ }
- // Used for query/insert
- public Coordinator(ConnectContext context, Planner planner,
- StatsErrorEstimator statsErrorEstimator) {
- this(context, planner);
- this.statsErrorEstimator = statsErrorEstimator;
+ public JobProcessor getJobProcessor() {
+ return coordinatorContext.getJobProcessor();
}
- public Coordinator(ConnectContext context, Planner planner,
- StatsErrorEstimator statsErrorEstimator, long jobId) {
- this(context, planner, statsErrorEstimator);
- this.jobId = jobId;
+ public LoadProcessor asLoadProcessor() {
+ return coordinatorContext.asLoadProcessor();
}
- // Used for query/insert/test
- public Coordinator(ConnectContext context, Planner planner) {
- this.context = context;
- this.queryId = context.queryId();
- this.fragments = planner.getFragments();
- this.scanNodes = planner.getScanNodes();
- this.descTable =
DescriptorToThriftConverter.toThrift(planner.getDescTable());
+ public void setTWorkloadGroups(List<TPipelineWorkloadGroup>
tWorkloadGroups) {
+ coordinatorContext.setWorkloadGroups(tWorkloadGroups);
+ }
- this.returnedAllResults = false;
- this.enableShareHashTableForBroadcastJoin =
context.getSessionVariable().enableShareHashTableForBroadcastJoin;
+ public List<TPipelineWorkloadGroup> getTWorkloadGroups() {
+ return coordinatorContext.getWorkloadGroups();
+ }
- this.queryOptions = CoordinatorContext.initQueryOptions(
- context, planner.getQueryOptions().getBeExecVersion());
- // Old Coordinator never plans local exchange in FE. Force BE to plan
its own.
- this.queryOptions.setEnableLocalShufflePlanner(false);
- useNereids = planner instanceof NereidsPlanner;
- if (!useNereids) {
- // Enable local shuffle on pipelineX engine only if Nereids
planner is applied.
- queryOptions.setEnableLocalShuffle(false);
- } else {
- distributedPlans = ((NereidsPlanner)
planner).getDistributedPlans();
- }
- CoordinatorContext.refreshQueryGlobals(this.queryGlobals, context);
- this.queryGlobals.setLoadZeroTolerance(false);
- this.assignedRuntimeFilters = planner.getRuntimeFilters();
- this.topnFilters = planner.getTopnFilters();
-
- List<Integer> fragmentIds = new ArrayList<>();
- for (PlanFragment fragment : fragments) {
- fragmentIds.add(fragment.getFragmentId().asInt());
+ public boolean isQueryCancelled() {
+ return coordinatorContext.readCloneStatus().isCancelled();
+ }
+
+ protected Status getQueryStatus() {
+ return coordinatorContext.readCloneStatus();
+ }
+
+ @Override
+ public RowBatch getNext() throws Exception {
+ return coordinatorContext.asQueryProcessor().getNext();
+ }
+
+ public boolean mustOutliveDispatch() {
+ return
coordinatorContext.scanNodes.stream().anyMatch(ScanNode::coordinatorMustOutliveDispatch);
+ }
+
+ public List<Long> getScanBackendIds() {
+ return coordinatorContext.scanNodes.stream()
+ .flatMap(scan -> scan.getScanBackendIds().stream())
+ .collect(Collectors.toList());
+ }
+
+ public Status shouldCancel(List<Backend> currentBackends) {
+ PipelineExecutionTask task = executionTask;
+ if (task == null) {
+ return Status.OK;
}
- this.executionProfile = new ExecutionProfile(queryId, fragmentIds);
- }
-
- // Used for broker load task/export task/update coordinator
- // Constructor of Coordinator is too complicated.
- public Coordinator(Long jobId, TUniqueId queryId, DescriptorTable
descTable, List<PlanFragment> fragments,
- List<ScanNode> scanNodes, String timezone, boolean
loadZeroTolerance, boolean enableProfile) {
- this.jobId = jobId;
- this.queryId = queryId;
- this.descTable = DescriptorToThriftConverter.toThrift(descTable);
- this.fragments = fragments;
- this.scanNodes = scanNodes;
- this.queryOptions = new TQueryOptions();
- this.queryOptions.setEnableProfile(enableProfile);
- this.queryOptions.setProfileLevel(2);
- CoordinatorContext.setQueryGlobalsForLoad(this.queryGlobals, timezone,
loadZeroTolerance);
- this.queryOptions.setBeExecVersion(Config.be_exec_version);
- this.queryOptions.setNewVersionUnixTimestamp(true);
- this.queryOptions.setNewVersionPercentile(true);
- this.queryOptions.setNewVersionBitmapOpCount(true);
-
- List<Integer> fragmentIds = new ArrayList<>();
- for (PlanFragment fragment : fragments) {
- fragmentIds.add(fragment.getFragmentId().asInt());
+ Set<Long> aliveBackendIds =
currentBackends.stream().filter(Backend::isAlive)
+ .map(Backend::getId).collect(Collectors.toSet());
+ for (Map.Entry<BackendWorker, MultiFragmentsPipelineTask> entry :
task.getChildrenTasks().entrySet()) {
+ // Remote Doris workers belong to another catalog and are absent
from the local heartbeat inventory.
+ if (entry.getKey().getCatalogId() !=
InternalCatalog.INTERNAL_CATALOG_ID) {
+ continue;
+ }
+ MultiFragmentsPipelineTask backendTask = entry.getValue();
+ if (!aliveBackendIds.contains(backendTask.getBackend().getId())) {
+ return new Status(TStatusCode.CANCELLED, "Backend {} not
exists or dead, query {} should be cancelled",
+ backendTask.getBackend(),
DebugUtil.printId(getQueryId()));
+ }
+ for (SingleFragmentPipelineTask fragmentTask :
backendTask.getChildrenTasks().values()) {
+ Status status =
fragmentTask.getBackendHealthStatus(getJobId());
Review Comment:
[P1] Preserve cancellation when a backend epoch changes from the
uninitialized value — `shouldCancel()` now delegates to
`SingleFragmentPipelineTask.getBackendHealthStatus()`, whose guard only reports
a restart when both the captured and current epochs are positive. A `Backend`
starts with `lastStartTime=-1` (and can be observed at epoch 0 during replay),
then heartbeat processing updates it to a positive start time; the legacy check
treated any `previous != current && current != 0` as a restart. With the new
guard, a query whose fragment was captured at -1/0 can keep waiting for reports
until its execution timeout after that backend restarts, instead of being
cancelled promptly. Keep the zero-current upgrade exception, but treat a
nonpositive captured epoch followed by a positive current epoch as unhealthy
and add a regression.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
