database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[opt](query cancel) cancel query if it has pipeline task leakage #39223 (#39537)

Browse Source 
pick #39223 with some modifications. Optimization will only be applied
to pipeline x.
This commit is contained in:
zhiqiang
2024-08-19 14:33:59 +08:00
committed by GitHub
parent c0cbb2362c
commit 830f250a80
15 changed files with 295 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
View File

@ -141,6 +141,8 @@ import org.apache.doris.thrift.TDropPlsqlStoredProcedureRequest;
import org.apache.doris.thrift.TExecPlanFragmentParams;
import org.apache.doris.thrift.TFeResult;
import org.apache.doris.thrift.TFetchResourceResult;
import org.apache.doris.thrift.TFetchRunningQueriesRequest;
import org.apache.doris.thrift.TFetchRunningQueriesResult;
import org.apache.doris.thrift.TFetchSchemaTableDataRequest;
import org.apache.doris.thrift.TFetchSchemaTableDataResult;
import org.apache.doris.thrift.TFetchSplitBatchRequest;
@ -237,6 +239,7 @@ import org.apache.doris.thrift.TTableRef;
import org.apache.doris.thrift.TTableStatus;
import org.apache.doris.thrift.TTabletLocation;
import org.apache.doris.thrift.TTxnParams;
import org.apache.doris.thrift.TUniqueId;
import org.apache.doris.thrift.TUpdateExportTaskStatusRequest;
import org.apache.doris.thrift.TUpdateFollowerStatsCacheRequest;
import org.apache.doris.thrift.TWaitingTxnStatusRequest;
@ -4013,4 +4016,24 @@ public class FrontendServiceImpl implements FrontendService.Iface {
result.setUserinfoList(userInfo);
return result;
}
@Override
public TFetchRunningQueriesResult fetchRunningQueries(TFetchRunningQueriesRequest request) {
TFetchRunningQueriesResult result = new TFetchRunningQueriesResult();
if (!Env.getCurrentEnv().isReady()) {
result.setStatus(new TStatus(TStatusCode.ILLEGAL_STATE));
return result;
}
List<TUniqueId> runningQueries = Lists.newArrayList();
List<Coordinator> allCoordinators = QeProcessorImpl.INSTANCE.getAllCoordinators();
for (Coordinator coordinator : allCoordinators) {
runningQueries.add(coordinator.getQueryId());
}
result.setStatus(new TStatus(TStatusCode.OK));
result.setRunningQueries(runningQueries);
return result;
}
}