This is an automated email from the ASF dual-hosted git repository. xiangweiwei pushed a commit to branch to0.12removeQMC in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d27fe2c4ccabe14529f4e98c73c84e6f8a47e0d5 Author: Alima777 <[email protected]> AuthorDate: Mon Jul 12 22:33:44 2021 +0800 [To rel/0.12] remove query memory control --- .../apache/iotdb/cluster/query/ClusterPlanner.java | 13 +---- .../iotdb/cluster/query/LocalQueryExecutor.java | 32 +++-------- .../cluster/query/manage/ClusterQueryManager.java | 8 +-- .../org/apache/iotdb/cluster/common/IoTDBTest.java | 3 +- .../query/ClusterAggregateExecutorTest.java | 4 +- .../query/ClusterDataQueryExecutorTest.java | 10 ++-- .../cluster/query/ClusterFillExecutorTest.java | 4 +- .../cluster/query/ClusterPlanExecutorTest.java | 4 +- .../cluster/query/ClusterQueryRouterTest.java | 12 ++-- .../ClusterGroupByNoVFilterDataSetTest.java | 2 +- .../groupby/ClusterGroupByVFilterDataSetTest.java | 2 +- .../query/groupby/MergeGroupByExecutorTest.java | 4 +- .../query/groupby/RemoteGroupByExecutorTest.java | 4 +- .../query/manage/ClusterQueryManagerTest.java | 12 ++-- .../query/reader/ClusterReaderFactoryTest.java | 2 +- .../query/reader/ClusterTimeGeneratorTest.java | 4 +- .../cluster/server/member/DataGroupMemberTest.java | 2 +- .../cluster/server/member/MetaGroupMemberTest.java | 4 +- .../exception/query/PathNumOverLimitException.java | 9 +-- .../org/apache/iotdb/db/monitor/StatMonitor.java | 2 +- .../main/java/org/apache/iotdb/db/qp/Planner.java | 31 ++--------- .../qp/strategy/optimizer/ConcatPathOptimizer.java | 18 +++--- .../qp/strategy/optimizer/ILogicalOptimizer.java | 3 +- .../db/query/control/QueryResourceManager.java | 48 +--------------- .../org/apache/iotdb/db/service/TSServiceImpl.java | 65 ++-------------------- .../db/integration/IoTDBSequenceDataQueryIT.java | 9 +-- .../iotdb/db/integration/IoTDBSeriesReaderIT.java | 11 ++-- .../iotdb/db/qp/logical/LogicalPlanSmallTest.java | 2 +- .../apache/iotdb/db/utils/EnvironmentUtils.java | 9 ++- 29 files changed, 77 insertions(+), 256 deletions(-) diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanner.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanner.java index ecbfb74..90fd0be 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanner.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanner.java @@ -22,11 +22,9 @@ package org.apache.iotdb.cluster.query; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.qp.Planner; import org.apache.iotdb.db.qp.logical.Operator; -import org.apache.iotdb.db.qp.logical.crud.SFWOperator; import org.apache.iotdb.db.qp.physical.PhysicalPlan; import org.apache.iotdb.db.qp.strategy.PhysicalGenerator; import org.apache.iotdb.db.qp.strategy.optimizer.ConcatPathOptimizer; -import org.apache.iotdb.db.query.control.QueryResourceManager; import java.time.ZoneId; @@ -37,16 +35,7 @@ public class ClusterPlanner extends Planner { public PhysicalPlan parseSQLToPhysicalPlan(String sqlStr, ZoneId zoneId, int fetchSize) throws QueryProcessException { Operator operator = logicalGenerator.generate(sqlStr, zoneId); - int maxDeduplicatedPathNum = - QueryResourceManager.getInstance().getMaxDeduplicatedPathNum(fetchSize); - if (operator instanceof SFWOperator && ((SFWOperator) operator).isLastQuery()) { - // Dataset of last query actually has only three columns, so we shouldn't limit the path num - // while constructing logical plan - // To avoid overflowing because logicalOptimize function may do maxDeduplicatedPathNum + 1, we - // set it to Integer.MAX_VALUE - 1 - maxDeduplicatedPathNum = Integer.MAX_VALUE - 1; - } - operator = logicalOptimize(operator, maxDeduplicatedPathNum); + operator = logicalOptimize(operator); PhysicalGenerator physicalGenerator = new ClusterPhysicalGenerator(); return physicalGenerator.transformToPhysicalPlan(operator, fetchSize); } diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/LocalQueryExecutor.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/LocalQueryExecutor.java index 460180d..903167d 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/query/LocalQueryExecutor.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/LocalQueryExecutor.java @@ -86,8 +86,6 @@ import java.util.List; import java.util.Map; import java.util.Set; -import static org.apache.iotdb.session.Config.DEFAULT_FETCH_SIZE; - public class LocalQueryExecutor { private static final Logger logger = LoggerFactory.getLogger(LocalQueryExecutor.class); @@ -231,11 +229,7 @@ public class LocalQueryExecutor { // the same query from a requester correspond to a context here RemoteQueryContext queryContext = - queryManager.getQueryContext( - request.getRequester(), - request.getQueryId(), - request.getFetchSize(), - request.getDeduplicatedPathNum()); + queryManager.getQueryContext(request.getRequester(), request.getQueryId()); logger.debug( DEBUG_SHOW_QUERY_ID, name, @@ -325,11 +319,7 @@ public class LocalQueryExecutor { // the same query from a requester correspond to a context here RemoteQueryContext queryContext = - queryManager.getQueryContext( - request.getRequester(), - request.getQueryId(), - request.getFetchSize(), - request.getDeduplicatedPathNum()); + queryManager.getQueryContext(request.getRequester(), request.getQueryId()); logger.debug( DEBUG_SHOW_QUERY_ID, name, @@ -492,11 +482,7 @@ public class LocalQueryExecutor { Set<String> deviceMeasurements = request.getDeviceMeasurements(); RemoteQueryContext queryContext = - queryManager.getQueryContext( - request.getRequester(), - request.getQueryId(), - request.getFetchSize(), - request.getDeduplicatedPathNum()); + queryManager.getQueryContext(request.getRequester(), request.getQueryId()); logger.debug( DEBUG_SHOW_QUERY_ID, name, @@ -584,8 +570,7 @@ public class LocalQueryExecutor { timeFilter = FilterFactory.deserialize(request.timeFilterBytes); } RemoteQueryContext queryContext = - queryManager.getQueryContext( - request.getRequestor(), request.queryId, DEFAULT_FETCH_SIZE, -1); + queryManager.getQueryContext(request.getRequestor(), request.queryId); Set<String> deviceMeasurements = request.getDeviceMeasurements(); boolean ascending = request.ascending; @@ -780,8 +765,7 @@ public class LocalQueryExecutor { Set<String> deviceMeasurements = request.getDeviceMeasurements(); boolean ascending = request.ascending; - RemoteQueryContext queryContext = - queryManager.getQueryContext(request.getRequestor(), queryId, DEFAULT_FETCH_SIZE, -1); + RemoteQueryContext queryContext = queryManager.getQueryContext(request.getRequestor(), queryId); LocalGroupByExecutor executor = getGroupByExecutor( path, @@ -867,8 +851,7 @@ public class LocalQueryExecutor { long beforeRange = request.getBeforeRange(); Node requester = request.getRequester(); Set<String> deviceMeasurements = request.getDeviceMeasurements(); - RemoteQueryContext queryContext = - queryManager.getQueryContext(requester, queryId, DEFAULT_FETCH_SIZE, -1); + RemoteQueryContext queryContext = queryManager.getQueryContext(requester, queryId); ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream); @@ -933,8 +916,7 @@ public class LocalQueryExecutor { dataGroupMember.syncLeaderWithConsistencyCheck(false); RemoteQueryContext queryContext = - queryManager.getQueryContext( - request.getRequestor(), request.getQueryId(), DEFAULT_FETCH_SIZE, -1); + queryManager.getQueryContext(request.getRequestor(), request.getQueryId()); List<PartialPath> partialPaths = new ArrayList<>(); for (String path : request.getPaths()) { partialPaths.add(new PartialPath(path)); diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/manage/ClusterQueryManager.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/manage/ClusterQueryManager.java index 455e7fb..f3376ba 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/query/manage/ClusterQueryManager.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/manage/ClusterQueryManager.java @@ -43,16 +43,12 @@ public class ClusterQueryManager { private Map<Long, IAggregateReader> aggrReaderMap = new ConcurrentHashMap<>(); private Map<Long, GroupByExecutor> groupByExecutorMap = new ConcurrentHashMap<>(); - public synchronized RemoteQueryContext getQueryContext( - Node node, long queryId, int fetchSize, int deduplicatedPathNum) { + public synchronized RemoteQueryContext getQueryContext(Node node, long queryId) { Map<Long, RemoteQueryContext> nodeContextMap = queryContextMap.computeIfAbsent(node, n -> new HashMap<>()); return nodeContextMap.computeIfAbsent( queryId, - qId -> - new RemoteQueryContext( - QueryResourceManager.getInstance() - .assignQueryId(true, fetchSize, deduplicatedPathNum))); + qId -> new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true))); } public long registerReader(IBatchReader reader) { diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/common/IoTDBTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/common/IoTDBTest.java index e7b1b7b..6a25ce8 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/common/IoTDBTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/common/IoTDBTest.java @@ -158,8 +158,7 @@ public abstract class IoTDBTest { protected QueryDataSet query(List<String> pathStrs, IExpression expression) throws QueryProcessException, QueryFilterOptimizationException, StorageEngineException, IOException, MetadataException, InterruptedException { - QueryContext context = - new QueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + QueryContext context = new QueryContext(QueryResourceManager.getInstance().assignQueryId(true)); RawDataQueryPlan queryPlan = new RawDataQueryPlan(); queryPlan.setExpression(expression); List<PartialPath> paths = new ArrayList<>(); diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterAggregateExecutorTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterAggregateExecutorTest.java index db4b6fe..086cdc7 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterAggregateExecutorTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterAggregateExecutorTest.java @@ -85,7 +85,7 @@ public class ClusterAggregateExecutorTest extends BaseQueryTest { plan.setDeduplicatedAggregations(aggregations); QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { executor = new ClusterAggregateExecutor(plan, testMetaMember); QueryDataSet queryDataSet = executor.executeWithoutValueFilter(context, plan); @@ -142,7 +142,7 @@ public class ClusterAggregateExecutorTest extends BaseQueryTest { new PartialPath(TestUtils.getTestSeries(0, 0)), TimeFilter.gtEq(3)))); QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { executor = new ClusterAggregateExecutor(plan, testMetaMember); QueryDataSet queryDataSet = executor.executeWithValueFilter(context, plan); diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterDataQueryExecutorTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterDataQueryExecutorTest.java index c6dfb93..034a1a6 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterDataQueryExecutorTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterDataQueryExecutorTest.java @@ -66,7 +66,7 @@ public class ClusterDataQueryExecutorTest extends BaseQueryTest { plan.setDeduplicatedDataTypes(dataTypes); queryExecutor = new ClusterDataQueryExecutor(plan, testMetaMember); RemoteQueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { QueryDataSet dataSet = queryExecutor.executeWithoutValueFilter(context); checkSequentialDataset(dataSet, 0, 20); @@ -87,7 +87,7 @@ public class ClusterDataQueryExecutorTest extends BaseQueryTest { plan.setExpression(expression); queryExecutor = new ClusterDataQueryExecutor(plan, testMetaMember); RemoteQueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { QueryDataSet dataSet = queryExecutor.executeWithValueFilter(context); checkSequentialDataset(dataSet, 5, 15); @@ -104,7 +104,7 @@ public class ClusterDataQueryExecutorTest extends BaseQueryTest { plan.setEnableRedirect(true); queryExecutor = new ClusterDataQueryExecutor(plan, testMetaMember); RemoteQueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { QueryDataSet dataSet = queryExecutor.executeWithoutValueFilter(context); assertNull(dataSet.getEndPoint()); @@ -126,7 +126,7 @@ public class ClusterDataQueryExecutorTest extends BaseQueryTest { plan.setEnableRedirect(true); queryExecutor = new ClusterDataQueryExecutor(plan, testMetaMember); RemoteQueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { QueryDataSet dataSet = queryExecutor.executeWithValueFilter(context); assertNull(dataSet.getEndPoint()); @@ -147,7 +147,7 @@ public class ClusterDataQueryExecutorTest extends BaseQueryTest { plan.setEnableRedirect(true); queryExecutor = new ClusterDataQueryExecutor(plan, testMetaMember); RemoteQueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { QueryDataSet dataSet = queryExecutor.executeWithoutValueFilter(context); assertEquals("ip:port=0.0.0.0:6667", dataSet.getEndPoint().toString()); diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterFillExecutorTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterFillExecutorTest.java index 8061851..8bbd9fa 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterFillExecutorTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterFillExecutorTest.java @@ -60,7 +60,7 @@ public class ClusterFillExecutorTest extends BaseQueryTest { TSDataType.DOUBLE, new PreviousFill(TSDataType.DOUBLE, 0, defaultFillInterval)); plan.setFillType(tsDataTypeIFillMap); QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { ClusterFillExecutor fillExecutor; @@ -107,7 +107,7 @@ public class ClusterFillExecutorTest extends BaseQueryTest { new LinearFill(TSDataType.DOUBLE, 0, defaultFillInterval, defaultFillInterval)); plan.setFillType(tsDataTypeIFillMap); QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { ClusterFillExecutor fillExecutor; diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterPlanExecutorTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterPlanExecutorTest.java index 5de2ce2..a6254b9 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterPlanExecutorTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterPlanExecutorTest.java @@ -61,7 +61,7 @@ public class ClusterPlanExecutorTest extends BaseQueryTest { queryPlan.setPaths(pathList); queryPlan.setDataTypes(dataTypes); QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { QueryDataSet dataSet = queryExecutor.processQuery(queryPlan, context); @@ -96,7 +96,7 @@ public class ClusterPlanExecutorTest extends BaseQueryTest { IOException, InterruptedException, QueryProcessException { ShowTimeSeriesPlan showTimeSeriesPlan = new ShowTimeSeriesPlan(pathList.get(0)); QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { QueryDataSet dataSet = queryExecutor.processQuery(showTimeSeriesPlan, context); int count = 0; diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterQueryRouterTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterQueryRouterTest.java index 9cd7690..7c963c2 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterQueryRouterTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/query/ClusterQueryRouterTest.java @@ -73,7 +73,7 @@ public class ClusterQueryRouterTest extends BaseQueryTest { queryPlan.setDeduplicatedPaths(pathList); queryPlan.setDeduplicatedDataTypes(dataTypes); QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { QueryDataSet dataSet = clusterQueryRouter.rawDataQuery(queryPlan, context); @@ -117,7 +117,7 @@ public class ClusterQueryRouterTest extends BaseQueryTest { plan.setDeduplicatedAggregations(aggregations); QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { QueryDataSet queryDataSet = clusterQueryRouter.aggregate(plan, context); checkDoubleDataset(queryDataSet, new Object[] {0.0, 19.0, 9.5, 20.0, 190.0}); @@ -142,7 +142,7 @@ public class ClusterQueryRouterTest extends BaseQueryTest { TSDataType.DOUBLE, new PreviousFill(TSDataType.DOUBLE, 0, defaultFillInterval)); plan.setFillType(tsDataTypeIFillMap); QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { QueryDataSet queryDataSet; @@ -183,7 +183,7 @@ public class ClusterQueryRouterTest extends BaseQueryTest { plan.setFillType(tsDataTypeIFillMap); QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { QueryDataSet queryDataSet; @@ -212,7 +212,7 @@ public class ClusterQueryRouterTest extends BaseQueryTest { throws IOException, StorageEngineException, QueryFilterOptimizationException, QueryProcessException, IllegalPathException { QueryContext queryContext = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { GroupByTimePlan groupByPlan = new GroupByTimePlan(); List<PartialPath> pathList = new ArrayList<>(); @@ -265,7 +265,7 @@ public class ClusterQueryRouterTest extends BaseQueryTest { throws StorageEngineException, IOException, QueryFilterOptimizationException, QueryProcessException, IllegalPathException { QueryContext queryContext = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { GroupByTimePlan groupByPlan = new GroupByTimePlan(); List<PartialPath> pathList = new ArrayList<>(); diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/ClusterGroupByNoVFilterDataSetTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/ClusterGroupByNoVFilterDataSetTest.java index 66aa6e5..db40a5f 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/ClusterGroupByNoVFilterDataSetTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/ClusterGroupByNoVFilterDataSetTest.java @@ -48,7 +48,7 @@ public class ClusterGroupByNoVFilterDataSetTest extends BaseQueryTest { public void test() throws StorageEngineException, IOException, QueryProcessException, IllegalPathException { QueryContext queryContext = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { GroupByTimePlan groupByPlan = new GroupByTimePlan(); List<PartialPath> pathList = new ArrayList<>(); diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/ClusterGroupByVFilterDataSetTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/ClusterGroupByVFilterDataSetTest.java index 8e39e93..fb89912 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/ClusterGroupByVFilterDataSetTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/ClusterGroupByVFilterDataSetTest.java @@ -51,7 +51,7 @@ public class ClusterGroupByVFilterDataSetTest extends BaseQueryTest { public void test() throws IOException, StorageEngineException, QueryProcessException, IllegalPathException { QueryContext queryContext = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { GroupByTimePlan groupByPlan = new GroupByTimePlan(); List<PartialPath> pathList = new ArrayList<>(); diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/MergeGroupByExecutorTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/MergeGroupByExecutorTest.java index e04e5a6..aa5bb34 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/MergeGroupByExecutorTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/MergeGroupByExecutorTest.java @@ -49,7 +49,7 @@ public class MergeGroupByExecutorTest extends BaseQueryTest { PartialPath path = new PartialPath(TestUtils.getTestSeries(0, 0)); TSDataType dataType = TSDataType.DOUBLE; QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { Filter timeFilter = null; @@ -88,7 +88,7 @@ public class MergeGroupByExecutorTest extends BaseQueryTest { PartialPath path = new PartialPath(TestUtils.getTestSeries(0, 0)); TSDataType dataType = TSDataType.DOUBLE; QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { Filter timeFilter = TimeFilter.gtEq(3); diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/RemoteGroupByExecutorTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/RemoteGroupByExecutorTest.java index dac449a..a89e811 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/RemoteGroupByExecutorTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/query/groupby/RemoteGroupByExecutorTest.java @@ -54,7 +54,7 @@ public class RemoteGroupByExecutorTest extends BaseQueryTest { PartialPath path = new PartialPath(TestUtils.getTestSeries(0, 0)); TSDataType dataType = TSDataType.DOUBLE; QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { Filter timeFilter = null; List<Integer> aggregationTypes = new ArrayList<>(); @@ -116,7 +116,7 @@ public class RemoteGroupByExecutorTest extends BaseQueryTest { PartialPath path = new PartialPath(TestUtils.getTestSeries(0, 0)); TSDataType dataType = TSDataType.DOUBLE; QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { Filter timeFilter = TimeFilter.gtEq(3); List<Integer> aggregationTypes = new ArrayList<>(); diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/query/manage/ClusterQueryManagerTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/query/manage/ClusterQueryManagerTest.java index 35a404e..a50b037 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/query/manage/ClusterQueryManagerTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/query/manage/ClusterQueryManagerTest.java @@ -46,12 +46,9 @@ public class ClusterQueryManagerTest { @Test public void testContext() { - RemoteQueryContext queryContext1 = - queryManager.getQueryContext(TestUtils.getNode(0), 1, 1024, -1); - RemoteQueryContext queryContext2 = - queryManager.getQueryContext(TestUtils.getNode(0), 1, 1024, -1); - RemoteQueryContext queryContext3 = - queryManager.getQueryContext(TestUtils.getNode(1), 1, 1024, -1); + RemoteQueryContext queryContext1 = queryManager.getQueryContext(TestUtils.getNode(0), 1); + RemoteQueryContext queryContext2 = queryManager.getQueryContext(TestUtils.getNode(0), 1); + RemoteQueryContext queryContext3 = queryManager.getQueryContext(TestUtils.getNode(1), 1); assertSame(queryContext1, queryContext2); assertNotEquals(queryContext2, queryContext3); } @@ -158,8 +155,7 @@ public class ClusterQueryManagerTest { @Test public void testEndQuery() throws StorageEngineException { - RemoteQueryContext queryContext = - queryManager.getQueryContext(TestUtils.getNode(0), 1, 1024, -1); + RemoteQueryContext queryContext = queryManager.getQueryContext(TestUtils.getNode(0), 1); for (int i = 0; i < 10; i++) { IBatchReader reader = new IBatchReader() { diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/ClusterReaderFactoryTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/ClusterReaderFactoryTest.java index e40be6a..dd3ccc4 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/ClusterReaderFactoryTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/ClusterReaderFactoryTest.java @@ -46,7 +46,7 @@ public class ClusterReaderFactoryTest extends BaseQueryTest { ClusterReaderFactory readerFactory = new ClusterReaderFactory(testMetaMember); RemoteQueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { SeriesRawDataBatchReader seriesReader = diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/ClusterTimeGeneratorTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/ClusterTimeGeneratorTest.java index 17261b2..fbbee45 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/ClusterTimeGeneratorTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/ClusterTimeGeneratorTest.java @@ -57,7 +57,7 @@ public class ClusterTimeGeneratorTest extends BaseQueryTest { throws StorageEngineException, IOException, IllegalPathException, QueryProcessException { RawDataQueryPlan dataQueryPlan = new RawDataQueryPlan(); QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { IExpression expression = BinaryExpression.and( @@ -86,7 +86,7 @@ public class ClusterTimeGeneratorTest extends BaseQueryTest { throws StorageEngineException, IOException, IllegalPathException, QueryProcessException { RawDataQueryPlan dataQueryPlan = new RawDataQueryPlan(); QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); Filter valueFilter = ValueFilter.gtEq(3.0); Filter timeFilter = TimeFilter.ltEq(8); try { diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/server/member/DataGroupMemberTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/server/member/DataGroupMemberTest.java index 05f4c60..5d138fa 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/server/member/DataGroupMemberTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/server/member/DataGroupMemberTest.java @@ -1108,7 +1108,7 @@ public class DataGroupMemberTest extends BaseMember { Filter timeFilter = TimeFilter.gtEq(5); request.setTimeFilterBytes(SerializeUtils.serializeFilter(timeFilter)); QueryContext queryContext = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { request.setQueryId(queryContext.getQueryId()); request.setRequestor(TestUtils.getNode(0)); diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java index f98bc22..0e1e677 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java @@ -878,7 +878,7 @@ public class MetaGroupMemberTest extends BaseMember { } QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { ClusterReaderFactory readerFactory = new ClusterReaderFactory(testMetaMember); @@ -940,7 +940,7 @@ public class MetaGroupMemberTest extends BaseMember { } QueryContext context = - new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, -1)); + new RemoteQueryContext(QueryResourceManager.getInstance().assignQueryId(true)); try { ClusterReaderFactory readerFactory = new ClusterReaderFactory(testMetaMember); diff --git a/server/src/main/java/org/apache/iotdb/db/exception/query/PathNumOverLimitException.java b/server/src/main/java/org/apache/iotdb/db/exception/query/PathNumOverLimitException.java index 7714fc8..c4eb10f 100644 --- a/server/src/main/java/org/apache/iotdb/db/exception/query/PathNumOverLimitException.java +++ b/server/src/main/java/org/apache/iotdb/db/exception/query/PathNumOverLimitException.java @@ -21,17 +21,10 @@ package org.apache.iotdb.db.exception.query; public class PathNumOverLimitException extends QueryProcessException { - public PathNumOverLimitException(long maxDeduplicatedPathNum, long deduplicatedPathNum) { - super( - String.format( - "Too many paths in one query! Currently allowed max deduplicated path number is %d, this query contains %d deduplicated path. Please use slimit to choose what you real want or adjust max_deduplicated_path_num in iotdb-engine.properties.", - maxDeduplicatedPathNum, deduplicatedPathNum)); - } - public PathNumOverLimitException(long maxDeduplicatedPathNum) { super( String.format( - "Too many paths in one query! Currently allowed max deduplicated path number is %d, this query contains unknown deduplicated path. Please use slimit to choose what you real want or adjust max_deduplicated_path_num in iotdb-engine.properties.", + "Too many paths in one query! Currently allowed max deduplicated path number is %d. Please use slimit or adjust max_deduplicated_path_num in iotdb-engine.properties.", maxDeduplicatedPathNum)); } } diff --git a/server/src/main/java/org/apache/iotdb/db/monitor/StatMonitor.java b/server/src/main/java/org/apache/iotdb/db/monitor/StatMonitor.java index c5a7848..e97d17a 100644 --- a/server/src/main/java/org/apache/iotdb/db/monitor/StatMonitor.java +++ b/server/src/main/java/org/apache/iotdb/db/monitor/StatMonitor.java @@ -164,7 +164,7 @@ public class StatMonitor implements StatMonitorMBean, IService { LastQueryExecutor.calculateLastPairForSeriesLocally( Collections.singletonList(monitorSeries), Collections.singletonList(TSDataType.INT64), - new QueryContext(QueryResourceManager.getInstance().assignQueryId(true, 1024, 1)), + new QueryContext(QueryResourceManager.getInstance().assignQueryId(true)), null, Collections.singletonMap( monitorSeries.getDevice(), diff --git a/server/src/main/java/org/apache/iotdb/db/qp/Planner.java b/server/src/main/java/org/apache/iotdb/db/qp/Planner.java index 355b7ef..dd9abd1 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/Planner.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/Planner.java @@ -39,7 +39,6 @@ import org.apache.iotdb.db.qp.strategy.optimizer.ConcatPathOptimizer; import org.apache.iotdb.db.qp.strategy.optimizer.DnfFilterOptimizer; import org.apache.iotdb.db.qp.strategy.optimizer.MergeSingleFilterOptimizer; import org.apache.iotdb.db.qp.strategy.optimizer.RemoveNotOptimizer; -import org.apache.iotdb.db.query.control.QueryResourceManager; import org.apache.iotdb.db.utils.TestOnly; import org.apache.iotdb.service.rpc.thrift.TSRawDataQueryReq; @@ -68,16 +67,7 @@ public class Planner { public PhysicalPlan parseSQLToPhysicalPlan(String sqlStr, ZoneId zoneId, int fetchSize) throws QueryProcessException { Operator operator = logicalGenerator.generate(sqlStr, zoneId); - int maxDeduplicatedPathNum = - QueryResourceManager.getInstance().getMaxDeduplicatedPathNum(fetchSize); - if (operator instanceof SFWOperator && ((SFWOperator) operator).isLastQuery()) { - // Dataset of last query actually has only three columns, so we shouldn't limit the path num - // while constructing logical plan - // To avoid overflowing because logicalOptimize function may do maxDeduplicatedPathNum + 1, we - // set it to Integer.MAX_VALUE - 1 - maxDeduplicatedPathNum = Integer.MAX_VALUE - 1; - } - operator = logicalOptimize(operator, maxDeduplicatedPathNum); + operator = logicalOptimize(operator); PhysicalGenerator physicalGenerator = new PhysicalGenerator(); return physicalGenerator.transformToPhysicalPlan(operator, fetchSize); } @@ -124,16 +114,7 @@ public class Planner { queryOp.setFilterOperator(filterOp); - int maxDeduplicatedPathNum = - QueryResourceManager.getInstance().getMaxDeduplicatedPathNum(rawDataQueryReq.fetchSize); - if (queryOp.isLastQuery()) { - // Dataset of last query actually has only three columns, so we shouldn't limit the path num - // while constructing logical plan - // To avoid overflowing because logicalOptimize function may do maxDeduplicatedPathNum + 1, we - // set it to Integer.MAX_VALUE - 1 - maxDeduplicatedPathNum = Integer.MAX_VALUE - 1; - } - SFWOperator op = (SFWOperator) logicalOptimize(queryOp, maxDeduplicatedPathNum); + SFWOperator op = (SFWOperator) logicalOptimize(queryOp); PhysicalGenerator physicalGenerator = new PhysicalGenerator(); return physicalGenerator.transformToPhysicalPlan(op, rawDataQueryReq.fetchSize); @@ -146,7 +127,7 @@ public class Planner { * @return optimized logical operator * @throws LogicalOptimizeException exception in logical optimizing */ - protected Operator logicalOptimize(Operator operator, int maxDeduplicatedPathNum) + protected Operator logicalOptimize(Operator operator) throws LogicalOperatorException, PathNumOverLimitException { switch (operator.getType()) { case AUTHOR: @@ -188,7 +169,7 @@ public class Planner { case DROP_INDEX: case QUERY_INDEX: SFWOperator root = (SFWOperator) operator; - return optimizeSFWOperator(root, maxDeduplicatedPathNum); + return optimizeSFWOperator(root); default: throw new LogicalOperatorException(operator.getType().toString(), ""); } @@ -201,10 +182,10 @@ public class Planner { * @return optimized select-from-where operator * @throws LogicalOptimizeException exception in SFW optimizing */ - private SFWOperator optimizeSFWOperator(SFWOperator root, int maxDeduplicatedPathNum) + private SFWOperator optimizeSFWOperator(SFWOperator root) throws LogicalOperatorException, PathNumOverLimitException { ConcatPathOptimizer concatPathOptimizer = getConcatPathOptimizer(); - root = (SFWOperator) concatPathOptimizer.transform(root, maxDeduplicatedPathNum); + root = (SFWOperator) concatPathOptimizer.transform(root); FilterOperator filter = root.getFilterOperator(); if (filter == null) { return root; diff --git a/server/src/main/java/org/apache/iotdb/db/qp/strategy/optimizer/ConcatPathOptimizer.java b/server/src/main/java/org/apache/iotdb/db/qp/strategy/optimizer/ConcatPathOptimizer.java index 0991328..4752eee 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/optimizer/ConcatPathOptimizer.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/optimizer/ConcatPathOptimizer.java @@ -18,6 +18,7 @@ */ package org.apache.iotdb.db.qp.strategy.optimizer; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.metadata.MetadataException; import org.apache.iotdb.db.exception.query.LogicalOptimizeException; import org.apache.iotdb.db.exception.query.PathNumOverLimitException; @@ -55,7 +56,7 @@ public class ConcatPathOptimizer implements ILogicalOptimizer { @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning @Override - public Operator transform(Operator operator, int maxDeduplicatedPathNum) + public Operator transform(Operator operator) throws LogicalOptimizeException, PathNumOverLimitException { if (!(operator instanceof SFWOperator)) { logger.warn("given operator isn't SFWOperator, cannot concat seriesPath"); @@ -101,7 +102,6 @@ public class ConcatPathOptimizer implements ILogicalOptimizer { select, seriesLimit, seriesOffset, - maxDeduplicatedPathNum, ((QueryOperator) operator).getIndexType() == null); } else { isAlignByDevice = true; @@ -178,7 +178,6 @@ public class ConcatPathOptimizer implements ILogicalOptimizer { SelectOperator selectOperator, int limit, int offset, - int maxDeduplicatedPathNum, boolean needRemoveStar) throws LogicalOptimizeException, PathNumOverLimitException { List<PartialPath> suffixPaths = judgeSelectOperator(selectOperator); @@ -238,8 +237,7 @@ public class ConcatPathOptimizer implements ILogicalOptimizer { afterConcatUdfList, selectOperator, limit, - offset, - maxDeduplicatedPathNum); + offset); } else { selectOperator.setSuffixPathList(afterConcatPaths); } @@ -343,14 +341,10 @@ public class ConcatPathOptimizer implements ILogicalOptimizer { List<UDFContext> afterConcatUdfList, SelectOperator selectOperator, int finalLimit, - int finalOffset, - int maxDeduplicatedPathNum) + int finalOffset) throws LogicalOptimizeException, PathNumOverLimitException { int offset = finalOffset; - int limit = - finalLimit == 0 || maxDeduplicatedPathNum < finalLimit - ? maxDeduplicatedPathNum + 1 - : finalLimit; + int limit = finalLimit == 0 ? Integer.MAX_VALUE : finalLimit; int consumed = 0; List<PartialPath> newSuffixPathList = new ArrayList<>(); @@ -424,6 +418,8 @@ public class ConcatPathOptimizer implements ILogicalOptimizer { } if (limit == 0) { + int maxDeduplicatedPathNum = + IoTDBDescriptor.getInstance().getConfig().getMaxQueryDeduplicatedPathNum(); if (maxDeduplicatedPathNum < newSuffixPathList.size()) { throw new PathNumOverLimitException(maxDeduplicatedPathNum); } diff --git a/server/src/main/java/org/apache/iotdb/db/qp/strategy/optimizer/ILogicalOptimizer.java b/server/src/main/java/org/apache/iotdb/db/qp/strategy/optimizer/ILogicalOptimizer.java index 26ab0a5..e96019d 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/optimizer/ILogicalOptimizer.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/optimizer/ILogicalOptimizer.java @@ -26,6 +26,5 @@ import org.apache.iotdb.db.qp.logical.Operator; @FunctionalInterface public interface ILogicalOptimizer { - Operator transform(Operator operator, int maxDeduplicatedPathNum) - throws LogicalOptimizeException, PathNumOverLimitException; + Operator transform(Operator operator) throws LogicalOptimizeException, PathNumOverLimitException; } diff --git a/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java b/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java index d0042a1..feec5e0 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java +++ b/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java @@ -25,7 +25,6 @@ import org.apache.iotdb.db.engine.StorageEngine; import org.apache.iotdb.db.engine.querycontext.QueryDataSource; import org.apache.iotdb.db.engine.storagegroup.TsFileResource; import org.apache.iotdb.db.exception.StorageEngineException; -import org.apache.iotdb.db.exception.query.PathNumOverLimitException; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.db.query.context.QueryContext; @@ -74,59 +73,20 @@ public class QueryResourceManager { */ private final Map<Long, List<IExternalSortFileDeserializer>> externalSortFileMap; - private final Map<Long, Long> queryIdEstimatedMemoryMap; - - // current total free memory for reading process(not including the cache memory) - private final AtomicLong totalFreeMemoryForRead; - - // estimated size for one point memory size, the unit is byte - private static final long POINT_ESTIMATED_SIZE = 16L; - - private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig(); - private QueryResourceManager() { filePathsManager = new QueryFileManager(); externalSortFileMap = new ConcurrentHashMap<>(); - queryIdEstimatedMemoryMap = new ConcurrentHashMap<>(); - totalFreeMemoryForRead = - new AtomicLong( - IoTDBDescriptor.getInstance().getConfig().getAllocateMemoryForReadWithoutCache()); } public static QueryResourceManager getInstance() { return QueryTokenManagerHelper.INSTANCE; } - public int getMaxDeduplicatedPathNum(int fetchSize) { - if (fetchSize == 0) { - return CONFIG.getMaxQueryDeduplicatedPathNum(); - } - return (int) - Math.min( - ((totalFreeMemoryForRead.get() / fetchSize) / POINT_ESTIMATED_SIZE), - CONFIG.getMaxQueryDeduplicatedPathNum()); - } - /** Register a new query. When a query request is created firstly, this method must be invoked. */ - public long assignQueryId(boolean isDataQuery, int fetchSize, int deduplicatedPathNum) { - int maxDeduplicatedPathNum = getMaxDeduplicatedPathNum(fetchSize); - if (deduplicatedPathNum > maxDeduplicatedPathNum) { - throw new RuntimeException( - new PathNumOverLimitException(maxDeduplicatedPathNum, deduplicatedPathNum)); - } + public long assignQueryId(boolean isDataQuery) { long queryId = queryIdAtom.incrementAndGet(); if (isDataQuery) { filePathsManager.addQueryId(queryId); - if (deduplicatedPathNum > 0) { - long estimatedMemoryUsage = - (long) deduplicatedPathNum * POINT_ESTIMATED_SIZE * (long) fetchSize; - // apply the memory successfully - if (totalFreeMemoryForRead.addAndGet(-estimatedMemoryUsage) >= 0) { - queryIdEstimatedMemoryMap.put(queryId, estimatedMemoryUsage); - } else { - totalFreeMemoryForRead.addAndGet(estimatedMemoryUsage); - } - } } return queryId; } @@ -212,12 +172,6 @@ public class QueryResourceManager { externalSortFileMap.remove(queryId); } - // put back the memory usage - Long estimatedMemoryUsage = queryIdEstimatedMemoryMap.remove(queryId); - if (estimatedMemoryUsage != null) { - totalFreeMemoryForRead.addAndGet(estimatedMemoryUsage); - } - // remove usage of opened file paths of current thread filePathsManager.removeUsedFilesForQuery(queryId); diff --git a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java index 67f0467..a61aedd 100644 --- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java +++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java @@ -50,7 +50,6 @@ import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan; import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan.MeasurementType; import org.apache.iotdb.db.qp.physical.crud.CreateTemplatePlan; import org.apache.iotdb.db.qp.physical.crud.DeletePlan; -import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan; import org.apache.iotdb.db.qp.physical.crud.InsertMultiTabletPlan; import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan; import org.apache.iotdb.db.qp.physical.crud.InsertRowsOfOneDevicePlan; @@ -58,7 +57,6 @@ import org.apache.iotdb.db.qp.physical.crud.InsertRowsPlan; import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan; import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan; import org.apache.iotdb.db.qp.physical.crud.QueryPlan; -import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan; import org.apache.iotdb.db.qp.physical.crud.SetDeviceTemplatePlan; import org.apache.iotdb.db.qp.physical.crud.UDFPlan; import org.apache.iotdb.db.qp.physical.crud.UDTFPlan; @@ -127,7 +125,6 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding; import org.apache.iotdb.tsfile.read.common.Path; import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet; -import org.apache.iotdb.tsfile.utils.Pair; import org.antlr.v4.runtime.misc.ParseCancellationException; import org.apache.thrift.TException; @@ -140,7 +137,6 @@ import java.sql.SQLException; import java.time.ZoneId; import java.util.ArrayList; import java.util.Arrays; -import java.util.Calendar; import java.util.Collections; import java.util.LinkedHashSet; import java.util.List; @@ -756,13 +752,8 @@ public class TSServiceImpl implements TSIService.Iface { long startTime = System.currentTimeMillis(); long queryId = -1; try { - - // pair.left = fetchSize, pair.right = deduplicatedNum - Pair<Integer, Integer> p = getMemoryParametersFromPhysicalPlan(plan, fetchSize); - fetchSize = p.left; - // generate the queryId for the operation - queryId = generateQueryId(true, fetchSize, p.right); + queryId = generateQueryId(true); // register query info to queryTimeManager if (!(plan instanceof ShowQueryProcesslistPlan)) { queryTimeManager.registerQuery(queryId, startTime, statement, timeout); @@ -881,53 +872,6 @@ public class TSServiceImpl implements TSIService.Iface { } } - /** - * get fetchSize and deduplicatedPathNum that are used for memory estimation - * - * @return Pair - fetchSize, deduplicatedPathNum - */ - private Pair<Integer, Integer> getMemoryParametersFromPhysicalPlan( - PhysicalPlan plan, int fetchSizeBefore) { - // In case users forget to set this field in query, use the default value - int fetchSize = fetchSizeBefore == 0 ? DEFAULT_FETCH_SIZE : fetchSizeBefore; - int deduplicatedPathNum = -1; - if (plan instanceof GroupByTimePlan) { - fetchSize = Math.min(getFetchSizeForGroupByTimePlan((GroupByTimePlan) plan), fetchSize); - } else if (plan.getOperatorType() == OperatorType.AGGREGATION) { - // the actual row number of aggregation query is 1 - fetchSize = 1; - } - if (plan instanceof AlignByDevicePlan) { - deduplicatedPathNum = ((AlignByDevicePlan) plan).getMeasurements().size(); - } else if (plan instanceof LastQueryPlan) { - // dataset of last query consists of three column: time column + value column = 1 - // deduplicatedPathNum - // and we assume that the memory which sensor name takes equals to 1 deduplicatedPathNum - deduplicatedPathNum = 2; - // last query's actual row number should be the minimum between the number of series and - // fetchSize - fetchSize = Math.min(((LastQueryPlan) plan).getDeduplicatedPaths().size(), fetchSize); - } else if (plan instanceof RawDataQueryPlan) { - deduplicatedPathNum = ((RawDataQueryPlan) plan).getDeduplicatedPaths().size(); - } - return new Pair<>(fetchSize, deduplicatedPathNum); - } - - /* - calculate fetch size for group by time plan - */ - private int getFetchSizeForGroupByTimePlan(GroupByTimePlan plan) { - int rows = (int) ((plan.getEndTime() - plan.getStartTime()) / plan.getInterval()); - // rows gets 0 is caused by: the end time - the start time < the time interval. - if (rows == 0 && plan.isIntervalByMonth()) { - Calendar calendar = Calendar.getInstance(); - calendar.setTimeInMillis(plan.getStartTime()); - calendar.add(Calendar.MONTH, (int) (plan.getInterval() / MS_TO_MONTH)); - rows = calendar.getTimeInMillis() <= plan.getEndTime() ? 1 : 0; - } - return rows; - } - private TSExecuteStatementResp getListDataSetHeaders(QueryDataSet dataSet) { return StaticResps.getNoTimeExecuteResp( dataSet.getPaths().stream().map(Path::getFullPath).collect(Collectors.toList()), @@ -1239,7 +1183,7 @@ public class TSServiceImpl implements TSIService.Iface { status = executeNonQueryPlan(plan); TSExecuteStatementResp resp = RpcUtils.getTSExecuteStatementResp(status); - long queryId = generateQueryId(false, DEFAULT_FETCH_SIZE, -1); + long queryId = generateQueryId(false); return resp.setQueryId(queryId); } @@ -2000,9 +1944,8 @@ public class TSServiceImpl implements TSIService.Iface { : RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR); } - private long generateQueryId(boolean isDataQuery, int fetchSize, int deduplicatedPathNum) { - return QueryResourceManager.getInstance() - .assignQueryId(isDataQuery, fetchSize, deduplicatedPathNum); + private long generateQueryId(boolean isDataQuery) { + return QueryResourceManager.getInstance().assignQueryId(isDataQuery); } protected List<TSDataType> getSeriesTypesByPaths( diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSequenceDataQueryIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSequenceDataQueryIT.java index 540bf53..a822949 100644 --- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSequenceDataQueryIT.java +++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSequenceDataQueryIT.java @@ -205,8 +205,7 @@ public class IoTDBSequenceDataQueryIT { new PartialPath(TestConstant.d1 + TsFileConstant.PATH_SEPARATOR + TestConstant.s1)); dataTypes.add(TSDataType.INT64); - TEST_QUERY_JOB_ID = - QueryResourceManager.getInstance().assignQueryId(true, 1024, pathList.size()); + TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true); TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID); RawDataQueryPlan queryPlan = new RawDataQueryPlan(); queryPlan.setDeduplicatedDataTypes(dataTypes); @@ -240,8 +239,7 @@ public class IoTDBSequenceDataQueryIT { dataTypes.add(TSDataType.INT64); GlobalTimeExpression globalTimeExpression = new GlobalTimeExpression(TimeFilter.gtEq(800L)); - TEST_QUERY_JOB_ID = - QueryResourceManager.getInstance().assignQueryId(true, 1024, pathList.size()); + TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true); TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID); RawDataQueryPlan queryPlan = new RawDataQueryPlan(); @@ -297,8 +295,7 @@ public class IoTDBSequenceDataQueryIT { SingleSeriesExpression singleSeriesExpression = new SingleSeriesExpression(queryPath, ValueFilter.gtEq(14)); - TEST_QUERY_JOB_ID = - QueryResourceManager.getInstance().assignQueryId(true, 1024, pathList.size()); + TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true); TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID); RawDataQueryPlan queryPlan = new RawDataQueryPlan(); diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSeriesReaderIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSeriesReaderIT.java index 1710979..55ec4c8 100644 --- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSeriesReaderIT.java +++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSeriesReaderIT.java @@ -313,8 +313,7 @@ public class IoTDBSeriesReaderIT { new PartialPath(TestConstant.d1 + TsFileConstant.PATH_SEPARATOR + TestConstant.s1)); dataTypes.add(TSDataType.INT64); - TEST_QUERY_JOB_ID = - QueryResourceManager.getInstance().assignQueryId(true, 1024, pathList.size()); + TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true); TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID); RawDataQueryPlan queryPlan = new RawDataQueryPlan(); @@ -345,8 +344,7 @@ public class IoTDBSeriesReaderIT { SingleSeriesExpression singleSeriesExpression = new SingleSeriesExpression(p, ValueFilter.gtEq(20)); - TEST_QUERY_JOB_ID = - QueryResourceManager.getInstance().assignQueryId(true, 1024, pathList.size()); + TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true); TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID); RawDataQueryPlan queryPlan = new RawDataQueryPlan(); @@ -374,7 +372,7 @@ public class IoTDBSeriesReaderIT { List<TSDataType> dataTypes = Collections.singletonList(TSDataType.INT32); SingleSeriesExpression expression = new SingleSeriesExpression(path, TimeFilter.gt(22987L)); - TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true, 1024, 1); + TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true); TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID); RawDataQueryPlan queryPlan = new RawDataQueryPlan(); @@ -414,8 +412,7 @@ public class IoTDBSeriesReaderIT { dataTypes.add(TSDataType.INT64); queryPlan.setDeduplicatedDataTypes(dataTypes); - TEST_QUERY_JOB_ID = - QueryResourceManager.getInstance().assignQueryId(true, 1024, pathList.size()); + TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true); TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID); SingleSeriesExpression singleSeriesExpression = diff --git a/server/src/test/java/org/apache/iotdb/db/qp/logical/LogicalPlanSmallTest.java b/server/src/test/java/org/apache/iotdb/db/qp/logical/LogicalPlanSmallTest.java index e384edd..1ea618f 100644 --- a/server/src/test/java/org/apache/iotdb/db/qp/logical/LogicalPlanSmallTest.java +++ b/server/src/test/java/org/apache/iotdb/db/qp/logical/LogicalPlanSmallTest.java @@ -193,7 +193,7 @@ public class LogicalPlanSmallTest { (RootOperator) logicalGenerator.generate(sqlStr, ZoneId.systemDefault()); IoTDB.metaManager.init(); ConcatPathOptimizer concatPathOptimizer = new ConcatPathOptimizer(); - concatPathOptimizer.transform(operator, 1000); + concatPathOptimizer.transform(operator); IoTDB.metaManager.clear(); // expected to throw LogicalOptimizeException: The value of SOFFSET (%d) is equal to or exceeds // the number of sequences (%d) that can actually be returned. diff --git a/server/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java b/server/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java index 7245c19..6a523b5 100644 --- a/server/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java +++ b/server/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java @@ -41,6 +41,9 @@ import org.apache.iotdb.db.service.IoTDB; import org.apache.iotdb.rpc.TConfigurationConst; import org.apache.iotdb.rpc.TSocketWrapper; +import javax.management.remote.JMXConnector; +import javax.management.remote.JMXConnectorFactory; +import javax.management.remote.JMXServiceURL; import org.apache.commons.io.FileUtils; import org.apache.thrift.TConfiguration; import org.apache.thrift.transport.TTransport; @@ -48,10 +51,6 @@ import org.apache.thrift.transport.TTransportException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.management.remote.JMXConnector; -import javax.management.remote.JMXConnectorFactory; -import javax.management.remote.JMXServiceURL; - import java.io.File; import java.io.IOException; import java.net.InetSocketAddress; @@ -247,7 +246,7 @@ public class EnvironmentUtils { } createAllDir(); - TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true, 1024, 0); + TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true); TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID); }
