This is an automated email from the ASF dual-hosted git repository. caogaofei pushed a commit to branch beyyes/dev/1.3 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e9ffd049a86ca26cecc310cd6b8aca10fe9d4212 Author: Beyyes <[email protected]> AuthorDate: Thu Mar 20 12:20:21 2025 +0800 save cpu --- .../protocol/thrift/impl/ClientRPCServiceImpl.java | 46 +++++++++++++++++----- .../execution/driver/DriverContext.java | 6 +++ .../fragment/FragmentInstanceContext.java | 6 ++- .../execution/fragment/QueryContext.java | 6 +++ .../execution/schedule/task/DriverTaskId.java | 6 +++ 5 files changed, 59 insertions(+), 11 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java index 4c2c4a4e654..bfd440bd319 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java @@ -655,6 +655,14 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { } } + private final QueryId queryId = new QueryId("stub_query"); + private final FragmentInstanceId instanceId = + new FragmentInstanceId(new PlanFragmentId(queryId, 0), "stub-instance"); + private final PlanNodeId planNodeId = new PlanNodeId("1"); + private final List<InputLocation[]> inputLocationList = + Collections.singletonList(new InputLocation[] {new InputLocation(0, 0)}); + private final String testDriverFullId = "driver_full_id"; + @SuppressWarnings("java:S2095") // close() do nothing private List<TsBlock> executeGroupByQueryInternal( SessionInfo sessionInfo, @@ -677,17 +685,13 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { Filter timeFilter = TimeFilterApi.between(startTime, endTime - 1); - QueryId queryId = new QueryId("stub_query"); - FragmentInstanceId instanceId = - new FragmentInstanceId(new PlanFragmentId(queryId, 0), "stub-instance"); FragmentInstanceStateMachine stateMachine = new FragmentInstanceStateMachine( instanceId, FragmentInstanceManager.getInstance().instanceNotificationExecutor); FragmentInstanceContext fragmentInstanceContext = createFragmentInstanceContext( instanceId, stateMachine, sessionInfo, dataRegionList.get(0), timeFilter); - DriverContext driverContext = new DriverContext(fragmentInstanceContext, 0); - PlanNodeId planNodeId = new PlanNodeId("1"); + DriverContext driverContext = new DriverContext(fragmentInstanceContext, 0, testDriverFullId); driverContext.addOperatorContext(1, planNodeId, SeriesScanOperator.class.getSimpleName()); driverContext .getOperatorContexts() @@ -709,7 +713,7 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { true, true), AggregationStep.SINGLE, - Collections.singletonList(new InputLocation[] {new InputLocation(0, 0)})); + inputLocationList); GroupByTimeParameter groupByTimeParameter = new GroupByTimeParameter( @@ -721,7 +725,7 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { if (isAligned) { path = new AlignedPath( - device, + device.split("\\."), Collections.singletonList(measurement), Collections.singletonList(measurementSchema)); operator = @@ -1035,8 +1039,7 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { return executeAggregationQueryInternal(req, SELECT_RESULT); } - @Override - public TSExecuteStatementResp executeGroupByQueryIntervalQuery(TSGroupByQueryIntervalReq req) + public TSExecuteStatementResp executeGroupByQueryIntervalQuery2(TSGroupByQueryIntervalReq req) throws TException { try { @@ -1101,6 +1104,31 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { } } + @Override + public TSExecuteStatementResp executeGroupByQueryIntervalQuery(TSGroupByQueryIntervalReq req) + throws TException { + try { + TSDataType dataType = TSDataType.getTsDataType((byte) req.getDataType()); + List<TsBlock> blockResult = Collections.emptyList(); + String outputColumnName = req.getAggregationType().name(); + List<ColumnHeader> columnHeaders = + Collections.singletonList(new ColumnHeader(outputColumnName, dataType)); + DatasetHeader header = new DatasetHeader(columnHeaders, false); + header.setColumnToTsBlockIndexMap(Collections.singletonList(outputColumnName)); + + TSExecuteStatementResp resp = createResponse(header, 1); + TSQueryDataSet queryDataSet = convertTsBlockByFetchSize(blockResult); + resp.setQueryDataSet(queryDataSet); + + return resp; + } catch (Exception e) { + return RpcUtils.getTSExecuteStatementResp( + onQueryException(e, "\"" + req + "\". " + OperationType.EXECUTE_AGG_QUERY)); + } finally { + SESSION_MANAGER.updateIdleTime(); + } + } + @Override public TSFetchResultsResp fetchResultsV2(TSFetchResultsReq req) { long startTime = System.nanoTime(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/DriverContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/DriverContext.java index 79cb4acf3b9..f1e8ee635e7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/DriverContext.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/DriverContext.java @@ -55,6 +55,12 @@ public class DriverContext { this.driverTaskID = new DriverTaskId(fragmentInstanceContext.getId(), pipelineId); } + public DriverContext( + FragmentInstanceContext fragmentInstanceContext, int pipelineId, String fullId) { + this.fragmentInstanceContext = fragmentInstanceContext; + this.driverTaskID = new DriverTaskId(fragmentInstanceContext.getId(), pipelineId, fullId); + } + public OperatorContext addOperatorContext( int operatorId, PlanNodeId planNodeId, String operatorType) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java index 22077803e97..5f587f4463e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java @@ -138,7 +138,7 @@ public class FragmentInstanceContext extends QueryContext { IDataRegionForQuery dataRegion, Filter timeFilter) { FragmentInstanceContext instanceContext = - new FragmentInstanceContext(id, stateMachine, sessionInfo, dataRegion, timeFilter); + new FragmentInstanceContext(id, stateMachine, sessionInfo, dataRegion, timeFilter, false); instanceContext.initialize(); instanceContext.start(); return instanceContext; @@ -220,7 +220,8 @@ public class FragmentInstanceContext extends QueryContext { FragmentInstanceStateMachine stateMachine, SessionInfo sessionInfo, IDataRegionForQuery dataRegion, - Filter globalTimeFilter) { + Filter globalTimeFilter, + boolean mayHaveDeleteOperation) { this.id = id; this.stateMachine = stateMachine; this.executionEndTime.set(END_TIME_INITIAL_VALUE); @@ -230,6 +231,7 @@ public class FragmentInstanceContext extends QueryContext { this.dataNodeQueryContextMap = null; this.memoryReservationManager = new ThreadSafeMemoryReservationManager(id.getQueryId(), this.getClass().getName()); + this.mayHaveDeleteOperation = mayHaveDeleteOperation; } @TestOnly diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/QueryContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/QueryContext.java index cbde4cca787..ac7d9db16c8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/QueryContext.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/QueryContext.java @@ -64,6 +64,8 @@ public class QueryContext { private final Set<TsFileID> nonExistentModFiles = new CopyOnWriteArraySet<>(); + protected boolean mayHaveDeleteOperation = true; + public QueryContext() {} public QueryContext(long queryId) { @@ -79,6 +81,10 @@ public class QueryContext { } private boolean checkIfModificationExists(TsFileResource tsFileResource) { + if (!mayHaveDeleteOperation) { + return false; + } + if (nonExistentModFiles.contains(tsFileResource.getTsFileID())) { return false; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/task/DriverTaskId.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/task/DriverTaskId.java index 865af681b37..7a25359a340 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/task/DriverTaskId.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/task/DriverTaskId.java @@ -43,6 +43,12 @@ public class DriverTaskId implements ID, Comparable<DriverTaskId> { this.fullId = String.format("%s.%d", id.getFullId(), pipelineId); } + public DriverTaskId(FragmentInstanceId id, int pipelineId, String fullId) { + this.fragmentInstanceId = id; + this.pipelineId = pipelineId; + this.fullId = fullId; + } + @Override public int hashCode() { return Objects.hash(fragmentInstanceId, pipelineId);
