This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 2600bf8 [IOTDB-749] Handle select * from root OOM (#1367)
2600bf8 is described below
commit 2600bf840bd7b4dd52b73466c09a34eb8c32564c
Author: Jackie Tien <[email protected]>
AuthorDate: Fri Oct 23 17:49:44 2020 +0800
[IOTDB-749] Handle select * from root OOM (#1367)
Co-authored-by: xiangdong huang <[email protected]>
---
.../resources/conf/iotdb-engine.properties | 10 ++--
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 23 ++++++++
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 39 ++++++++------
.../exception/query/PathNumOverLimitException.java | 35 ++++++++++++
.../org/apache/iotdb/db/metadata/MManager.java | 14 ++---
.../java/org/apache/iotdb/db/metadata/MTree.java | 27 ++++++++--
.../main/java/org/apache/iotdb/db/qp/Planner.java | 33 +++++++-----
.../iotdb/db/qp/strategy/PhysicalGenerator.java | 53 +++++++++++-------
.../qp/strategy/optimizer/ConcatPathOptimizer.java | 50 +++++++++--------
.../qp/strategy/optimizer/ILogicalOptimizer.java | 4 +-
.../db/query/control/QueryResourceManager.java | 49 +++++++++++++++--
.../org/apache/iotdb/db/service/TSServiceImpl.java | 62 +++++++++++++++-------
.../db/integration/IoTDBSequenceDataQueryIT.java | 9 ++--
.../iotdb/db/integration/IoTDBSeriesReaderIT.java | 11 ++--
.../org/apache/iotdb/db/metadata/MTreeTest.java | 6 +--
.../iotdb/db/qp/plan/LogicalPlanSmallTest.java | 2 +-
.../apache/iotdb/db/utils/EnvironmentUtils.java | 15 +++---
.../apache/iotdb/spark/db/EnvironmentUtils.java | 14 ++---
18 files changed, 324 insertions(+), 132 deletions(-)
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties
b/server/src/assembly/resources/conf/iotdb-engine.properties
index 7810559..5332cff 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -236,6 +236,10 @@ write_read_free_memory_proportion=6:3:1
# primitive array size (length of each array) in array pool
primitive_array_size=128
+# allowed max numbers of deduplicated path in one query
+# it's just an advised value, the real limitation will be the smaller one
between this and the one we calculated
+max_deduplicated_path_num=1000
+
####################
### Upgrade Configurations
####################
@@ -340,9 +344,9 @@ merge_read_throughput_mb_per_sec=16
# whether to cache meta data(ChunkMetadata and TimeSeriesMetadata) or not.
meta_data_cache_enable=true
-# Read memory Allocation Ratio: ChunkMetadataCache, ChunkCache,
TimeSeriesMetadataCache and Free Memory Used in Query.
-# The parameter form is a:b:c:d, where a, b, c and d are integers. for
example: 1:1:1:1 , 6:10:5:15
-chunkmeta_chunk_timeseriesmeta_free_memory_proportion=1:1:1:7
+# Read memory Allocation Ratio: ChunkMetadataCache, ChunkCache,
TimeSeriesMetadataCache, memory used for constructing QueryDataSet and Free
Memory Used in Query.
+# The parameter form is a:b:c:d:e, where a, b, c, d and e are integers. for
example: 1:1:1:1:1 , 1:1:1:3:4
+chunkmeta_chunk_timeseriesmeta_free_memory_proportion=1:1:1:3:4
# cache size for MManager.
# This cache is used to improve insert speed where all path check and
TSDataType will be cached in MManager with corresponding Path.
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index ee06c00..c65b744 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -131,6 +131,13 @@ public class IoTDBConfig {
private long allocateMemoryForRead = Runtime.getRuntime().maxMemory() * 3 /
10;
/**
+ * Memory allocated for the read process besides cache
+ */
+ private long allocateMemoryForReadWithoutCache =
Runtime.getRuntime().maxMemory() * 9 / 100;
+
+ private volatile int maxQueryDeduplicatedPathNum = 1000;
+
+ /**
* Is dynamic parameter adapter enable.
*/
private boolean enableParameterAdapter = true;
@@ -1195,6 +1202,14 @@ public class IoTDBConfig {
this.allocateMemoryForRead = allocateMemoryForRead;
}
+ public long getAllocateMemoryForReadWithoutCache() {
+ return allocateMemoryForReadWithoutCache;
+ }
+
+ public void setAllocateMemoryForReadWithoutCache(long
allocateMemoryForReadWithoutCache) {
+ this.allocateMemoryForReadWithoutCache = allocateMemoryForReadWithoutCache;
+ }
+
public boolean isEnableExternalSort() {
return enableExternalSort;
}
@@ -1858,4 +1873,12 @@ public class IoTDBConfig {
public long getStartUpNanosecond() {
return startUpNanosecond;
}
+
+ public int getMaxQueryDeduplicatedPathNum() {
+ return maxQueryDeduplicatedPathNum;
+ }
+
+ public void setMaxQueryDeduplicatedPathNum(int maxQueryDeduplicatedPathNum) {
+ this.maxQueryDeduplicatedPathNum = maxQueryDeduplicatedPathNum;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index a552d64..7b324d7 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -98,7 +98,8 @@ public class IoTDBDescriptor {
if (urlString == null) {
urlString = System.getProperty(IoTDBConstant.IOTDB_HOME, null);
if (urlString != null) {
- urlString = urlString + File.separatorChar + "conf" +
File.separatorChar + IoTDBConfig.CONFIG_NAME;
+ urlString =
+ urlString + File.separatorChar + "conf" + File.separatorChar +
IoTDBConfig.CONFIG_NAME;
} else {
// If this too wasn't provided, try to find a default config in the
root of the classpath.
URL uri = IoTDBConfig.class.getResource("/" + IoTDBConfig.CONFIG_NAME);
@@ -116,13 +117,13 @@ public class IoTDBDescriptor {
}
// If a config location was provided, but it doesn't end with a properties
file,
// append the default location.
- else if(!urlString.endsWith(".properties")) {
+ else if (!urlString.endsWith(".properties")) {
urlString += (File.separatorChar + IoTDBConfig.CONFIG_NAME);
}
// If the url doesn't start with "file:" or "classpath:", it's provided as
a normal path.
// So we need to add it to make it a real URL.
- if(!urlString.startsWith("file:") && !urlString.startsWith("classpath:")) {
+ if (!urlString.startsWith("file:") && !urlString.startsWith("classpath:"))
{
urlString = "file:" + urlString;
}
try {
@@ -138,7 +139,7 @@ public class IoTDBDescriptor {
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity
warning
private void loadProps() {
URL url = getPropsUrl();
- if(url == null) {
+ if (url == null) {
logger.warn("Couldn't load the configuration from any of the known
sources.");
return;
}
@@ -207,7 +208,7 @@ public class IoTDBDescriptor {
Boolean.toString(conf.isMetaDataCacheEnable()))));
conf.setEnableLastCache(Boolean.parseBoolean(properties.getProperty("enable_last_cache",
- Boolean.toString(conf.isLastCacheEnabled()))));
+ Boolean.toString(conf.isLastCacheEnabled()))));
initMemoryAllocate(properties);
@@ -563,7 +564,7 @@ public class IoTDBDescriptor {
conf.setEnableDiscardOutOfOrderData(Boolean.parseBoolean(
properties.getProperty("enable_discard_out_of_order_data",
- Boolean.toString(conf.isEnableDiscardOutOfOrderData()))));
+ Boolean.toString(conf.isEnableDiscardOutOfOrderData()))));
}
@@ -696,6 +697,10 @@ public class IoTDBDescriptor {
// update tsfile-format config
loadTsFileProps(properties);
+ // update max_deduplicated_path_num
+ conf.setMaxQueryDeduplicatedPathNum(
+
Integer.parseInt(properties.getProperty("max_deduplicated_path_num")));
+
} catch (Exception e) {
throw new QueryProcessException(
String.format("Fail to reload configuration because %s", e));
@@ -704,7 +709,7 @@ public class IoTDBDescriptor {
public void loadHotModifiedProps() throws QueryProcessException {
URL url = getPropsUrl();
- if(url == null) {
+ if (url == null) {
logger.warn("Couldn't load the configuration from any of the known
sources.");
return;
}
@@ -732,9 +737,9 @@ public class IoTDBDescriptor {
long maxMemoryAvailable = Runtime.getRuntime().maxMemory();
if (proportionSum != 0) {
conf.setAllocateMemoryForWrite(
- maxMemoryAvailable * Integer.parseInt(proportions[0].trim()) /
proportionSum);
+ maxMemoryAvailable * Integer.parseInt(proportions[0].trim()) /
proportionSum);
conf.setAllocateMemoryForRead(
- maxMemoryAvailable * Integer.parseInt(proportions[1].trim()) /
proportionSum);
+ maxMemoryAvailable * Integer.parseInt(proportions[1].trim()) /
proportionSum);
}
}
@@ -757,19 +762,23 @@ public class IoTDBDescriptor {
if (proportionSum != 0) {
try {
conf.setAllocateMemoryForChunkMetaDataCache(
- maxMemoryAvailable * Integer.parseInt(proportions[0].trim())
/ proportionSum);
+ maxMemoryAvailable * Integer.parseInt(proportions[0].trim()) /
proportionSum);
conf.setAllocateMemoryForChunkCache(
- maxMemoryAvailable * Integer.parseInt(proportions[1].trim())
/ proportionSum);
+ maxMemoryAvailable * Integer.parseInt(proportions[1].trim()) /
proportionSum);
conf.setAllocateMemoryForTimeSeriesMetaDataCache(
- maxMemoryAvailable * Integer.parseInt(proportions[2].trim())
/ proportionSum);
+ maxMemoryAvailable * Integer.parseInt(proportions[2].trim()) /
proportionSum);
+ conf.setAllocateMemoryForReadWithoutCache(
+ maxMemoryAvailable * Integer.parseInt(proportions[3].trim()) /
proportionSum);
} catch (Exception e) {
throw new RuntimeException(
- "Each subsection of configuration item
chunkmeta_chunk_timeseriesmeta_free_memory_proportion"
- + " should be an integer, which is "
- + queryMemoryAllocateProportion);
+ "Each subsection of configuration item
chunkmeta_chunk_timeseriesmeta_free_memory_proportion"
+ + " should be an integer, which is "
+ + queryMemoryAllocateProportion);
}
}
+ conf.setMaxQueryDeduplicatedPathNum(
+
Integer.parseInt(properties.getProperty("max_deduplicated_path_num")));
}
}
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
new file mode 100644
index 0000000..6829d68
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/exception/query/PathNumOverLimitException.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.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.",
+ maxDeduplicatedPathNum));
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
index efeb4af..f4c6c3f 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
@@ -737,9 +737,9 @@ public class MManager {
/**
* Similar to method getAllTimeseriesPath(), but return Path with alias
alias.
*/
- public List<PartialPath> getAllTimeseriesPathWithAlias(PartialPath
prefixPath)
- throws MetadataException {
- return mtree.getAllTimeseriesPathWithAlias(prefixPath);
+ public Pair<List<PartialPath>, Integer>
getAllTimeseriesPathWithAlias(PartialPath prefixPath,
+ int limit, int offset) throws MetadataException {
+ return mtree.getAllTimeseriesPathWithAlias(prefixPath, limit, offset);
}
/**
@@ -1715,7 +1715,8 @@ public class MManager {
internalCreateTimeseries(deviceId.concatNode(measurementList[i]),
dataType);
}
- MeasurementMNode measurementMNode = (MeasurementMNode)
deviceMNode.getChild(measurementList[i]);
+ MeasurementMNode measurementMNode = (MeasurementMNode) deviceMNode
+ .getChild(measurementList[i]);
// check type is match
TSDataType insertDataType = null;
@@ -1778,8 +1779,9 @@ public class MManager {
Collections.emptyMap());
} catch (PathAlreadyExistException | AliasAlreadyExistException e) {
if (logger.isDebugEnabled()) {
- logger.debug("Ignore PathAlreadyExistException and
AliasAlreadyExistException when Concurrent inserting"
- + " a non-exist time series {}", path);
+ logger.debug(
+ "Ignore PathAlreadyExistException and AliasAlreadyExistException
when Concurrent inserting"
+ + " a non-exist time series {}", path);
}
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java
b/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java
index 1fa8160..387c7b5 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java
@@ -776,11 +776,16 @@ public class MTree implements Serializable {
* Get all timeseries paths under the given path
*
* @param prefixPath a prefix path or a full path, may contain '*'.
+ *
+ * @return Pair.left contains all the satisfied paths
+ * Pair.right means the current offset or zero if we don't set
offset.
*/
- List<PartialPath> getAllTimeseriesPathWithAlias(PartialPath prefixPath)
throws MetadataException {
+ Pair<List<PartialPath>, Integer> getAllTimeseriesPathWithAlias(PartialPath
prefixPath, int limit, int offset) throws MetadataException {
PartialPath prePath = new PartialPath(prefixPath.getNodes());
ShowTimeSeriesPlan plan = new ShowTimeSeriesPlan(prefixPath);
- List<Pair<PartialPath, String[]>> res = getAllMeasurementSchema(plan);
+ plan.setLimit(limit);
+ plan.setOffset(offset);
+ List<Pair<PartialPath, String[]>> res = getAllMeasurementSchema(plan,
false);
List<PartialPath> paths = new ArrayList<>();
for (Pair<PartialPath, String[]> p : res) {
if (prePath.getMeasurement().equals(p.right[0])) {
@@ -788,7 +793,13 @@ public class MTree implements Serializable {
}
paths.add(p.left);
}
- return paths;
+ if (curOffset.get() == null) {
+ offset = 0;
+ } else {
+ offset = curOffset.get() + 1;
+ }
+ curOffset.remove();
+ return new Pair<>(paths, offset);
}
/**
@@ -986,6 +997,12 @@ public class MTree implements Serializable {
*/
List<Pair<PartialPath, String[]>> getAllMeasurementSchema(ShowTimeSeriesPlan
plan)
throws MetadataException {
+ return getAllMeasurementSchema(plan, true);
+ }
+
+
+ List<Pair<PartialPath, String[]>> getAllMeasurementSchema(ShowTimeSeriesPlan
plan, boolean removeCurrentOffset)
+ throws MetadataException {
List<Pair<PartialPath, String[]>> res;
String[] nodes = plan.getPath().getNodes();
if (nodes.length == 0 || !nodes[0].equals(root.getName())) {
@@ -1005,7 +1022,9 @@ public class MTree implements Serializable {
// avoid memory leaks
limit.remove();
offset.remove();
- curOffset.remove();
+ if (removeCurrentOffset) {
+ curOffset.remove();
+ }
count.remove();
return res;
}
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 3d005e9..b8632d2 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
@@ -29,6 +29,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.query.LogicalOperatorException;
import org.apache.iotdb.db.exception.query.LogicalOptimizeException;
+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.qp.constant.SQLConstant;
@@ -46,6 +47,7 @@ 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;
@@ -64,15 +66,20 @@ public class Planner {
public PhysicalPlan parseSQLToPhysicalPlan(String sqlStr)
throws QueryProcessException {
IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
- return parseSQLToPhysicalPlan(sqlStr, config.getZoneID());
+ return parseSQLToPhysicalPlan(sqlStr, config.getZoneID(), 1024);
}
- public PhysicalPlan parseSQLToPhysicalPlan(String sqlStr, ZoneId zoneId)
+ /**
+ * @param fetchSize this parameter only take effect when it is a query plan
+ */
+ public PhysicalPlan parseSQLToPhysicalPlan(String sqlStr, ZoneId zoneId, int
fetchSize)
throws QueryProcessException {
Operator operator = parseDriver.parse(sqlStr, zoneId);
- operator = logicalOptimize(operator);
+ int maxDeduplicatedPathNum = QueryResourceManager.getInstance()
+ .getMaxDeduplicatedPathNum(fetchSize);
+ operator = logicalOptimize(operator, maxDeduplicatedPathNum);
PhysicalGenerator physicalGenerator = new PhysicalGenerator();
- return physicalGenerator.transformToPhysicalPlan(operator);
+ return physicalGenerator.transformToPhysicalPlan(operator, fetchSize);
}
/**
@@ -115,9 +122,11 @@ public class Planner {
queryOp.setFilterOperator(filterOp);
- SFWOperator op = (SFWOperator) logicalOptimize(queryOp);
+ int maxDeduplicatedPathNum = QueryResourceManager.getInstance()
+ .getMaxDeduplicatedPathNum(rawDataQueryReq.fetchSize);
+ SFWOperator op = (SFWOperator) logicalOptimize(queryOp,
maxDeduplicatedPathNum);
PhysicalGenerator physicalGenerator = new PhysicalGenerator();
- return physicalGenerator.transformToPhysicalPlan(op);
+ return physicalGenerator.transformToPhysicalPlan(op,
rawDataQueryReq.fetchSize);
}
/**
@@ -127,8 +136,8 @@ public class Planner {
* @return optimized logical operator
* @throws LogicalOptimizeException exception in logical optimizing
*/
- protected Operator logicalOptimize(Operator operator)
- throws LogicalOperatorException {
+ protected Operator logicalOptimize(Operator operator, int
maxDeduplicatedPathNum)
+ throws LogicalOperatorException, PathNumOverLimitException {
switch (operator.getType()) {
case AUTHOR:
case METADATA:
@@ -162,7 +171,7 @@ public class Planner {
case UPDATE:
case DELETE:
SFWOperator root = (SFWOperator) operator;
- return optimizeSFWOperator(root);
+ return optimizeSFWOperator(root, maxDeduplicatedPathNum);
default:
throw new LogicalOperatorException(operator.getType().toString(), "");
}
@@ -175,10 +184,10 @@ public class Planner {
* @return optimized select-from-where operator
* @throws LogicalOptimizeException exception in SFW optimizing
*/
- private SFWOperator optimizeSFWOperator(SFWOperator root)
- throws LogicalOperatorException {
+ private SFWOperator optimizeSFWOperator(SFWOperator root, int
maxDeduplicatedPathNum)
+ throws LogicalOperatorException, PathNumOverLimitException {
ConcatPathOptimizer concatPathOptimizer = getConcatPathOptimizer();
- root = (SFWOperator) concatPathOptimizer.transform(root);
+ root = (SFWOperator) concatPathOptimizer.transform(root,
maxDeduplicatedPathNum);
FilterOperator filter = root.getFilterOperator();
if (filter == null) {
return root;
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
index 6a721e3..3aaab37 100644
---
a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
@@ -28,6 +28,7 @@ import java.util.Map;
import java.util.Set;
import org.apache.iotdb.db.auth.AuthException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.exception.query.PathNumOverLimitException;
import org.apache.iotdb.db.exception.query.LogicalOperatorException;
import org.apache.iotdb.db.exception.query.LogicalOptimizeException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
@@ -102,6 +103,7 @@ import
org.apache.iotdb.db.qp.physical.sys.ShowPlan.ShowContentType;
import org.apache.iotdb.db.qp.physical.sys.ShowStorageGroupPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowTTLPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowTimeSeriesPlan;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
import org.apache.iotdb.db.qp.physical.sys.TracingPlan;
import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.utils.SchemaUtils;
@@ -115,8 +117,10 @@ import org.apache.iotdb.tsfile.utils.Pair;
*/
public class PhysicalGenerator {
+
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity
warning
- public PhysicalPlan transformToPhysicalPlan(Operator operator) throws
QueryProcessException {
+ public PhysicalPlan transformToPhysicalPlan(Operator operator, int fetchSize)
+ throws QueryProcessException {
List<PartialPath> paths;
switch (operator.getType()) {
case AUTHOR:
@@ -144,10 +148,9 @@ public class PhysicalGenerator {
return new
DeleteStorageGroupPlan(deleteStorageGroup.getDeletePathList());
case CREATE_TIMESERIES:
CreateTimeSeriesOperator createOperator = (CreateTimeSeriesOperator)
operator;
- if (createOperator.getTags() != null
- && !createOperator.getTags().isEmpty()
- && createOperator.getAttributes() != null
- && !createOperator.getAttributes().isEmpty()) {
+ if (createOperator.getTags() != null &&
!createOperator.getTags().isEmpty()
+ && createOperator.getAttributes() != null &&
!createOperator.getAttributes()
+ .isEmpty()) {
for (String tagKey : createOperator.getTags().keySet()) {
if (createOperator.getAttributes().containsKey(tagKey)) {
throw new QueryProcessException(
@@ -198,7 +201,7 @@ public class PhysicalGenerator {
return new TracingPlan(tracingOperator.isTracingon());
case QUERY:
QueryOperator query = (QueryOperator) operator;
- return transformQuery(query);
+ return transformQuery(query, fetchSize);
case TTL:
switch (operator.getTokenIntType()) {
case SQLConstant.TOK_SET:
@@ -211,9 +214,8 @@ public class PhysicalGenerator {
ShowTTLOperator showTTLOperator = (ShowTTLOperator) operator;
return new ShowTTLPlan(showTTLOperator.getStorageGroups());
default:
- throw new LogicalOperatorException(
- String.format(
- "not supported operator type %s in ttl operation.",
operator.getType()));
+ throw new LogicalOperatorException(String
+ .format("not supported operator type %s in ttl operation.",
operator.getType()));
}
case LOAD_CONFIGURATION:
LoadConfigurationOperatorType type = ((LoadConfigurationOperator)
operator)
@@ -246,14 +248,10 @@ public class PhysicalGenerator {
return new CountPlan(
ShowContentType.COUNT_STORAGE_GROUP, ((CountOperator)
operator).getPath());
case SQLConstant.TOK_COUNT_NODE_TIMESERIES:
- return new CountPlan(
- ShowContentType.COUNT_NODE_TIMESERIES,
- ((CountOperator) operator).getPath(),
- ((CountOperator) operator).getLevel());
+ return new CountPlan(ShowContentType.COUNT_NODE_TIMESERIES,
+ ((CountOperator) operator).getPath(), ((CountOperator)
operator).getLevel());
case SQLConstant.TOK_COUNT_NODES:
- return new CountPlan(
- ShowContentType.COUNT_NODES,
- ((CountOperator) operator).getPath(),
+ return new CountPlan(ShowContentType.COUNT_NODES, ((CountOperator)
operator).getPath(),
((CountOperator) operator).getLevel());
case SQLConstant.TOK_COUNT_TIMESERIES:
return new CountPlan(
@@ -329,8 +327,10 @@ public class PhysicalGenerator {
return SchemaUtils.getSeriesTypesByPath(paths);
}
+
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity
warning
- private PhysicalPlan transformQuery(QueryOperator queryOperator) throws
QueryProcessException {
+ private PhysicalPlan transformQuery(QueryOperator queryOperator, int
fetchSize)
+ throws QueryProcessException {
QueryPlan queryPlan;
if (queryOperator.hasAggregation()) {
@@ -539,6 +539,13 @@ public class PhysicalGenerator {
measurements = slimitTrimColumn(measurements, seriesSlimit,
seriesOffset);
}
+ int maxDeduplicatedPathNum = QueryResourceManager.getInstance()
+ .getMaxDeduplicatedPathNum(fetchSize);
+
+ if (measurements.size() > maxDeduplicatedPathNum) {
+ throw new PathNumOverLimitException(maxDeduplicatedPathNum,
measurements.size());
+ }
+
// assigns to alignByDevicePlan
alignByDevicePlan.setMeasurements(measurements);
alignByDevicePlan.setMeasurementAliasMap(measurementAliasMap);
@@ -584,7 +591,7 @@ public class PhysicalGenerator {
}
}
try {
- deduplicate(queryPlan);
+ deduplicate(queryPlan, fetchSize);
} catch (MetadataException e) {
throw new QueryProcessException(e);
}
@@ -662,7 +669,8 @@ public class PhysicalGenerator {
}
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity
warning
- private void deduplicate(QueryPlan queryPlan) throws MetadataException {
+ private void deduplicate(QueryPlan queryPlan, int fetchSize)
+ throws MetadataException, PathNumOverLimitException {
// generate dataType first
List<PartialPath> paths = queryPlan.getPaths();
List<TSDataType> dataTypes = getSeriesTypes(paths);
@@ -700,6 +708,9 @@ public class PhysicalGenerator {
}
indexedPaths.sort(Comparator.comparing(pair -> pair.left));
+ int maxDeduplicatedPathNum = QueryResourceManager.getInstance()
+ .getMaxDeduplicatedPathNum(fetchSize);
+ int deduplicatedPathNum = 0;
int index = 0;
for (Pair<PartialPath, Integer> indexedPath : indexedPaths) {
String column = indexedPath.left.getTsAlias();
@@ -714,6 +725,10 @@ public class PhysicalGenerator {
TSDataType seriesType = dataTypes.get(indexedPath.right);
rawDataQueryPlan.addDeduplicatedPaths(indexedPath.left);
rawDataQueryPlan.addDeduplicatedDataTypes(seriesType);
+ deduplicatedPathNum++;
+ if (deduplicatedPathNum > maxDeduplicatedPathNum) {
+ throw new PathNumOverLimitException(maxDeduplicatedPathNum,
deduplicatedPathNum);
+ }
columnSet.add(column);
rawDataQueryPlan.addPathToIndex(column, index++);
if (queryPlan instanceof AggregationPlan) {
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 b8048fc..5d089a6 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
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Set;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.LogicalOptimizeException;
+import org.apache.iotdb.db.exception.query.PathNumOverLimitException;
import org.apache.iotdb.db.exception.runtime.SQLParserException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.qp.constant.SQLConstant;
@@ -36,6 +37,7 @@ import org.apache.iotdb.db.qp.logical.crud.QueryOperator;
import org.apache.iotdb.db.qp.logical.crud.SFWOperator;
import org.apache.iotdb.db.qp.logical.crud.SelectOperator;
import org.apache.iotdb.db.service.IoTDB;
+import org.apache.iotdb.tsfile.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,7 +53,8 @@ public class ConcatPathOptimizer implements ILogicalOptimizer
{
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity
warning
@Override
- public Operator transform(Operator operator) throws LogicalOptimizeException
{
+ public Operator transform(Operator operator, int maxDeduplicatedPathNum)
+ throws LogicalOptimizeException, PathNumOverLimitException {
if (!(operator instanceof SFWOperator)) {
logger.warn("given operator isn't SFWOperator, cannot concat
seriesPath");
return operator;
@@ -88,12 +91,21 @@ public class ConcatPathOptimizer implements
ILogicalOptimizer {
if (operator instanceof QueryOperator) {
if (!((QueryOperator) operator).isAlignByDevice() || ((QueryOperator)
operator)
.isLastQuery()) {
- concatSelect(prefixPaths, select); // concat and remove star
+ // concat and remove star
if (((QueryOperator) operator).hasSlimit()) {
int seriesLimit = ((QueryOperator) operator).getSeriesLimit();
int seriesOffset = ((QueryOperator) operator).getSeriesOffset();
- slimitTrim(select, seriesLimit, seriesOffset);
+ if (seriesLimit > maxDeduplicatedPathNum) {
+ throw new PathNumOverLimitException(maxDeduplicatedPathNum,
seriesLimit);
+ }
+ concatSelect(prefixPaths, select, seriesLimit, seriesOffset);
+ slimitTrim(select, seriesOffset);
+ } else {
+ concatSelect(prefixPaths, select, maxDeduplicatedPathNum + 1, 0);
+ if (select.getSuffixPaths().size() > maxDeduplicatedPathNum) {
+ throw new PathNumOverLimitException(maxDeduplicatedPathNum);
+ }
}
} else {
isAlignByDevice = true;
@@ -159,7 +171,7 @@ public class ConcatPathOptimizer implements
ILogicalOptimizer {
* Extract paths from select&from cql, expand them into complete versions,
and reassign them to
* selectOperator's suffixPathList. Treat aggregations similarly.
*/
- private void concatSelect(List<PartialPath> fromPaths, SelectOperator
selectOperator)
+ private void concatSelect(List<PartialPath> fromPaths, SelectOperator
selectOperator, int limit, int offset)
throws LogicalOptimizeException {
List<PartialPath> suffixPaths = judgeSelectOperator(selectOperator);
@@ -180,34 +192,26 @@ public class ConcatPathOptimizer implements
ILogicalOptimizer {
}
}
- removeStarsInPath(allPaths, afterConcatAggregations, selectOperator);
+ removeStarsInPath(allPaths, afterConcatAggregations, selectOperator,
limit, offset);
}
/**
- * Make 'SLIMIT&SOFFSET' take effect by trimming the suffixList and
aggregations of the
+ * Make 'SOFFSET' take effect by trimming the suffixList and aggregations of
the
* selectOperator.
*
- * @param seriesLimit is ensured to be positive integer
* @param seriesOffset is ensured to be non-negative integer
*/
- private void slimitTrim(SelectOperator select, int seriesLimit, int
seriesOffset)
+ private void slimitTrim(SelectOperator select, int seriesOffset)
throws LogicalOptimizeException {
List<PartialPath> suffixList = select.getSuffixPaths();
List<String> aggregations = select.getAggregations();
int size = suffixList.size();
// check parameter range
- if (seriesOffset >= size) {
+ if (size == 0) {
throw new LogicalOptimizeException("SOFFSET <SOFFSETValue>: SOFFSETValue
exceeds the range.");
}
- int endPosition = seriesOffset + seriesLimit;
- if (endPosition > size) {
- endPosition = size;
- }
-
- // trim seriesPath list
- List<PartialPath> trimedSuffixList = new
ArrayList<>(suffixList.subList(seriesOffset, endPosition));
- select.setSuffixPathList(trimedSuffixList);
+ int endPosition = seriesOffset + size;
// trim aggregations if exists
if (aggregations != null && !aggregations.isEmpty()) {
@@ -287,7 +291,7 @@ public class ConcatPathOptimizer implements
ILogicalOptimizer {
HashSet<PartialPath> pathSet = new HashSet<>();
try {
for (PartialPath path : paths) {
- List<PartialPath> all = removeWildcard(path);
+ List<PartialPath> all = removeWildcard(path, 0, 0).left;
for (PartialPath subPath : all) {
if (!pathSet.contains(subPath)) {
pathSet.add(subPath);
@@ -302,12 +306,14 @@ public class ConcatPathOptimizer implements
ILogicalOptimizer {
}
private void removeStarsInPath(List<PartialPath> paths, List<String>
afterConcatAggregations,
- SelectOperator selectOperator) throws LogicalOptimizeException {
+ SelectOperator selectOperator, int limit, int offset) throws
LogicalOptimizeException {
List<PartialPath> retPaths = new ArrayList<>();
List<String> newAggregations = new ArrayList<>();
for (int i = 0; i < paths.size(); i++) {
try {
- List<PartialPath> actualPaths = removeWildcard(paths.get(i));
+ Pair<List<PartialPath>, Integer> pair = removeWildcard(paths.get(i),
limit, offset);
+ List<PartialPath> actualPaths = pair.left;
+ offset = offset - pair.right;
if (paths.get(i).getTsAlias() != null) {
if (actualPaths.size() == 1) {
actualPaths.get(0).setTsAlias(paths.get(i).getTsAlias());
@@ -330,7 +336,7 @@ public class ConcatPathOptimizer implements
ILogicalOptimizer {
selectOperator.setAggregations(newAggregations);
}
- protected List<PartialPath> removeWildcard(PartialPath path) throws
MetadataException {
- return IoTDB.metaManager.getAllTimeseriesPathWithAlias(path);
+ protected Pair<List<PartialPath>, Integer> removeWildcard(PartialPath path,
int limit, int offset) throws MetadataException {
+ return IoTDB.metaManager.getAllTimeseriesPathWithAlias(path, limit,
offset);
}
}
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 bd542d6..4467a8a 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
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.qp.strategy.optimizer;
import org.apache.iotdb.db.exception.query.LogicalOptimizeException;
+import org.apache.iotdb.db.exception.query.PathNumOverLimitException;
import org.apache.iotdb.db.qp.logical.Operator;
/**
@@ -27,5 +28,6 @@ import org.apache.iotdb.db.qp.logical.Operator;
@FunctionalInterface
public interface ILogicalOptimizer {
- Operator transform(Operator operator) throws LogicalOptimizeException;
+ Operator transform(Operator operator, int maxDeduplicatedPathNum)
+ 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 8aaa311..7d14885 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
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.db.query.control;
+
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
@@ -43,6 +44,7 @@ import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+
/**
* <p>
* QueryResourceManager manages resource (file streams) used by each query
job, and assign Ids to
@@ -53,8 +55,8 @@ import org.slf4j.LoggerFactory;
*/
public class QueryResourceManager {
- private AtomicLong queryIdAtom = new AtomicLong();
- private QueryFileManager filePathsManager;
+ private final AtomicLong queryIdAtom = new AtomicLong();
+ private final QueryFileManager filePathsManager;
private static final Logger logger =
LoggerFactory.getLogger(QueryResourceManager.class);
// record the total number and size of chunks for each query id
private Map<Long, Integer> chunkNumMap = new ConcurrentHashMap<>();
@@ -64,29 +66,58 @@ public class QueryResourceManager {
private Map<Long, Set<TsFileResource>> seqFileNumMap = new
ConcurrentHashMap<>();
private Map<Long, Set<TsFileResource>> unseqFileNumMap = new
ConcurrentHashMap<>();
private IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+
/**
* Record temporary files used for external sorting.
* <p>
* Key: query job id. Value: temporary file list used for external sorting.
*/
- private Map<Long, List<IExternalSortFileDeserializer>> externalSortFileMap;
+ 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) {
+ return Math.min((int) ((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) {
+ public long assignQueryId(boolean isDataQuery, int fetchSize, int
deduplicatedPathNum) {
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;
}
@@ -138,7 +169,8 @@ public class QueryResourceManager {
if (config.isEnablePerformanceTracing()) {
boolean isprinted = false;
if (seqFileNumMap.get(queryId) != null && unseqFileNumMap.get(queryId)
!= null) {
- TracingManager.getInstance().writeTsFileInfo(queryId,
seqFileNumMap.remove(queryId).size(),
+ TracingManager.getInstance()
+ .writeTsFileInfo(queryId, seqFileNumMap.remove(queryId).size(),
unseqFileNumMap.remove(queryId).size());
isprinted = true;
}
@@ -167,6 +199,13 @@ 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 8444949..76ad37d 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
@@ -68,6 +68,7 @@ import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
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.sys.AuthorPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateMultiTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
@@ -144,7 +145,7 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
private static final int MAX_SIZE =
IoTDBDescriptor.getInstance().getConfig().getQueryCacheSizeInMetric();
private static final int DELETE_SIZE = 20;
- private static final int FETCH_SIZE = 10000;
+ private static final int DEFAULT_FETCH_SIZE = 10000;
private static final String ERROR_PARSING_SQL =
"meet error while parsing SQL to physical plan: {}";
private static final String SERVER_INTERNAL_ERROR = "{}: server Internal
Error: ";
@@ -354,8 +355,10 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
status = RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
break;
case "ALL_COLUMNS":
- resp.setColumnsList(getPaths(new
PartialPath(req.getColumnPath())).stream().map(PartialPath::getFullPath).collect(
- Collectors.toList()));
+ resp.setColumnsList(
+ getPaths(new
PartialPath(req.getColumnPath())).stream().map(PartialPath::getFullPath)
+ .collect(
+ Collectors.toList()));
status = RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
break;
default:
@@ -425,8 +428,8 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
// on finding queries in a batch, such query will be ignored and an error
will be generated
private boolean executeStatementInBatch(String statement, List<TSStatus>
result, long sessionId) {
try {
- PhysicalPlan physicalPlan =
- processor.parseSQLToPhysicalPlan(statement,
sessionIdZoneIdMap.get(sessionId));
+ PhysicalPlan physicalPlan = processor
+ .parseSQLToPhysicalPlan(statement,
sessionIdZoneIdMap.get(sessionId), DEFAULT_FETCH_SIZE);
if (physicalPlan.isQuery()) {
throw new QueryInBatchStatementException(statement);
}
@@ -478,8 +481,9 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
}
String statement = req.getStatement();
- PhysicalPlan physicalPlan =
- processor.parseSQLToPhysicalPlan(statement,
sessionIdZoneIdMap.get(req.getSessionId()));
+ PhysicalPlan physicalPlan = processor
+ .parseSQLToPhysicalPlan(statement,
sessionIdZoneIdMap.get(req.getSessionId()),
+ req.fetchSize);
if (physicalPlan.isQuery()) {
return internalExecuteQueryStatement(statement, req.statementId,
physicalPlan,
req.fetchSize,
@@ -516,8 +520,9 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
String statement = req.getStatement();
PhysicalPlan physicalPlan;
try {
- physicalPlan =
- processor.parseSQLToPhysicalPlan(statement,
sessionIdZoneIdMap.get(req.getSessionId()));
+ physicalPlan = processor
+ .parseSQLToPhysicalPlan(statement,
sessionIdZoneIdMap.get(req.getSessionId()),
+ req.fetchSize);
} catch (QueryProcessException | SQLParserException e) {
logger.info(ERROR_PARSING_SQL, e.getMessage());
return
RpcUtils.getTSExecuteStatementResp(TSStatusCode.SQL_PARSE_ERROR,
e.getMessage());
@@ -601,7 +606,7 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
// In case users forget to set this field in query, use the default value
if (fetchSize == 0) {
- fetchSize = FETCH_SIZE;
+ fetchSize = DEFAULT_FETCH_SIZE;
}
if (plan instanceof ShowTimeSeriesPlan) {
@@ -628,8 +633,20 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
resp.setIgnoreTimeStamp(true);
} // else default ignoreTimeStamp is false
resp.setOperationType(plan.getOperatorType().toString());
+
+ // get deduplicated path num
+ int deduplicatedPathNum = -1;
+ if (plan instanceof AlignByDevicePlan) {
+ deduplicatedPathNum = ((AlignByDevicePlan)
plan).getMeasurements().size();
+ } else if (plan instanceof LastQueryPlan) {
+ deduplicatedPathNum = 0;
+ } else if (plan instanceof RawDataQueryPlan) {
+ deduplicatedPathNum = ((RawDataQueryPlan)
plan).getDeduplicatedPaths().size();
+ }
+
// generate the queryId for the operation
- queryId = generateQueryId(true);
+
+ queryId = generateQueryId(true, fetchSize, deduplicatedPathNum);
if (plan instanceof QueryPlan && config.isEnablePerformanceTracing()) {
if (!(plan instanceof AlignByDevicePlan)) {
TracingManager.getInstance().writeQueryInfo(queryId, statement,
plan.getPaths().size());
@@ -825,7 +842,8 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
for (PartialPath path : paths) {
String column = path.getTsAlias();
if (column == null) {
- column = path.getMeasurementAlias() != null ?
path.getFullPathWithAlias() : path.getFullPath();
+ column = path.getMeasurementAlias() != null ?
path.getFullPathWithAlias()
+ : path.getFullPath();
}
respColumns.add(column);
seriesTypes.add(getSeriesTypeByPath(path));
@@ -1068,7 +1086,7 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
status = executeNonQueryPlan(plan);
TSExecuteStatementResp resp = RpcUtils.getTSExecuteStatementResp(status);
- long queryId = generateQueryId(false);
+ long queryId = generateQueryId(false, DEFAULT_FETCH_SIZE, -1);
resp.setQueryId(queryId);
return resp;
}
@@ -1086,7 +1104,8 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
PhysicalPlan physicalPlan;
try {
- physicalPlan = processor.parseSQLToPhysicalPlan(statement,
sessionIdZoneIdMap.get(sessionId));
+ physicalPlan = processor
+ .parseSQLToPhysicalPlan(statement,
sessionIdZoneIdMap.get(sessionId), DEFAULT_FETCH_SIZE);
} catch (QueryProcessException | SQLParserException e) {
logger.warn(ERROR_PARSING_SQL, statement, e);
return RpcUtils.getTSExecuteStatementResp(TSStatusCode.SQL_PARSE_ERROR,
e.getMessage());
@@ -1381,7 +1400,8 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
return RpcUtils.getStatus(TSStatusCode.NOT_LOGIN_ERROR);
}
- InsertTabletPlan insertTabletPlan = new InsertTabletPlan(new
PartialPath(req.deviceId), req.measurements);
+ InsertTabletPlan insertTabletPlan = new InsertTabletPlan(new
PartialPath(req.deviceId),
+ req.measurements);
insertTabletPlan.setTimes(QueryDataSetUtils.readTimesFromBuffer(req.timestamps,
req.size));
insertTabletPlan.setColumns(
QueryDataSetUtils.readValuesFromBuffer(
@@ -1415,7 +1435,8 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
List<TSStatus> statusList = new ArrayList<>();
for (int i = 0; i < req.deviceIds.size(); i++) {
- InsertTabletPlan insertTabletPlan = new InsertTabletPlan(new
PartialPath(req.deviceIds.get(i)),
+ InsertTabletPlan insertTabletPlan = new InsertTabletPlan(
+ new PartialPath(req.deviceIds.get(i)),
req.measurementsList.get(i));
insertTabletPlan.setTimes(
QueryDataSetUtils.readTimesFromBuffer(req.timestampsList.get(i),
req.sizeList.get(i)));
@@ -1689,11 +1710,14 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
: RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR);
}
- private long generateQueryId(boolean isDataQuery) {
- return QueryResourceManager.getInstance().assignQueryId(isDataQuery);
+
+ private long generateQueryId(boolean isDataQuery, int fetchSize, int
deduplicatedPathNum) {
+ return QueryResourceManager.getInstance()
+ .assignQueryId(isDataQuery, fetchSize, deduplicatedPathNum);
}
- protected List<TSDataType> getSeriesTypesByPaths(List<PartialPath> paths,
List<String> aggregations)
+ protected List<TSDataType> getSeriesTypesByPaths(List<PartialPath> paths,
+ List<String> aggregations)
throws MetadataException {
return SchemaUtils.getSeriesTypesByPaths(paths, aggregations);
}
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 8f8c82b..415fae6 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
@@ -185,7 +185,8 @@ public class IoTDBSequenceDataQueryIT {
pathList.add(new PartialPath(TestConstant.d1 +
TsFileConstant.PATH_SEPARATOR + TestConstant.s1));
dataTypes.add(TSDataType.INT64);
- TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true);
+ TEST_QUERY_JOB_ID = QueryResourceManager.getInstance()
+ .assignQueryId(true, 1024, pathList.size());
TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID);
RawDataQueryPlan queryPlan = new RawDataQueryPlan();
queryPlan.setDeduplicatedDataTypes(dataTypes);
@@ -216,7 +217,8 @@ public class IoTDBSequenceDataQueryIT {
dataTypes.add(TSDataType.INT64);
GlobalTimeExpression globalTimeExpression = new
GlobalTimeExpression(TimeFilter.gtEq(800L));
- TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true);
+ TEST_QUERY_JOB_ID = QueryResourceManager.getInstance()
+ .assignQueryId(true, 1024, pathList.size());
TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID);
RawDataQueryPlan queryPlan = new RawDataQueryPlan();
@@ -264,7 +266,8 @@ public class IoTDBSequenceDataQueryIT {
SingleSeriesExpression singleSeriesExpression = new
SingleSeriesExpression(queryPath,
ValueFilter.gtEq(14));
- TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true);
+ TEST_QUERY_JOB_ID = QueryResourceManager.getInstance()
+ .assignQueryId(true, 1024, pathList.size());
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 54bba5f..a5304df 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
@@ -278,7 +278,8 @@ public class IoTDBSeriesReaderIT {
pathList.add(new PartialPath(TestConstant.d1 +
TsFileConstant.PATH_SEPARATOR + TestConstant.s1));
dataTypes.add(TSDataType.INT64);
- TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true);
+ TEST_QUERY_JOB_ID = QueryResourceManager.getInstance()
+ .assignQueryId(true, 1024, pathList.size());
TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID);
RawDataQueryPlan queryPlan = new RawDataQueryPlan();
@@ -308,7 +309,8 @@ public class IoTDBSeriesReaderIT {
SingleSeriesExpression singleSeriesExpression = new
SingleSeriesExpression(p,
ValueFilter.gtEq(20));
- TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true);
+ TEST_QUERY_JOB_ID = QueryResourceManager.getInstance()
+ .assignQueryId(true, 1024, pathList.size());
TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID);
RawDataQueryPlan queryPlan = new RawDataQueryPlan();
@@ -335,7 +337,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);
+ TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true,
1024, 1);
TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID);
RawDataQueryPlan queryPlan = new RawDataQueryPlan();
@@ -373,7 +375,8 @@ public class IoTDBSeriesReaderIT {
dataTypes.add(TSDataType.INT64);
queryPlan.setDeduplicatedDataTypes(dataTypes);
- TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true);
+ TEST_QUERY_JOB_ID = QueryResourceManager.getInstance()
+ .assignQueryId(true, 1024, pathList.size());
TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID);
SingleSeriesExpression singleSeriesExpression = new
SingleSeriesExpression(path1,
diff --git a/server/src/test/java/org/apache/iotdb/db/metadata/MTreeTest.java
b/server/src/test/java/org/apache/iotdb/db/metadata/MTreeTest.java
index bab01f1..86a4901 100644
--- a/server/src/test/java/org/apache/iotdb/db/metadata/MTreeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/metadata/MTreeTest.java
@@ -185,14 +185,14 @@ public class MTreeTest {
assertEquals("root.a.d1.s0", result.get(1).getFullPath());
List<PartialPath> result2 = root
- .getAllTimeseriesPathWithAlias(new PartialPath("root.a.*.s0"));
+ .getAllTimeseriesPathWithAlias(new PartialPath("root.a.*.s0"), 0,
0).left;
assertEquals(2, result2.size());
assertEquals("root.a.d0.s0", result2.get(0).getFullPath());
assertNull(result2.get(0).getMeasurementAlias());
assertEquals("root.a.d1.s0", result2.get(1).getFullPath());
assertNull(result2.get(1).getMeasurementAlias());
- result2 = root.getAllTimeseriesPathWithAlias(new
PartialPath("root.a.*.temperature"));
+ result2 = root.getAllTimeseriesPathWithAlias(new
PartialPath("root.a.*.temperature"), 0, 0).left;
assertEquals(2, result2.size());
assertEquals("root.a.d0.temperature",
result2.get(0).getFullPathWithAlias());
assertEquals("root.a.d1.temperature",
result2.get(1).getFullPathWithAlias());
@@ -464,7 +464,7 @@ public class MTreeTest {
assertEquals(2, root.getDevices(new PartialPath("root")).size());
assertEquals(2, root.getAllTimeseriesCount(new PartialPath("root")));
assertEquals(2, root.getAllTimeseriesPath(new PartialPath("root")).size());
- assertEquals(2, root.getAllTimeseriesPathWithAlias(new
PartialPath("root")).size());
+ assertEquals(2, root.getAllTimeseriesPathWithAlias(new
PartialPath("root"), 0, 0).left.size());
}
@Test
diff --git
a/server/src/test/java/org/apache/iotdb/db/qp/plan/LogicalPlanSmallTest.java
b/server/src/test/java/org/apache/iotdb/db/qp/plan/LogicalPlanSmallTest.java
index f44567e..c68eea7 100644
--- a/server/src/test/java/org/apache/iotdb/db/qp/plan/LogicalPlanSmallTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/qp/plan/LogicalPlanSmallTest.java
@@ -182,7 +182,7 @@ public class LogicalPlanSmallTest {
.parse(sqlStr, IoTDBDescriptor.getInstance().getConfig().getZoneID());
IoTDB.metaManager.init();
ConcatPathOptimizer concatPathOptimizer = new ConcatPathOptimizer();
- concatPathOptimizer.transform(operator);
+ concatPathOptimizer.transform(operator, 1000);
IoTDB.metaManager.clear();
// expected to throw LogicalOptimizeException: SOFFSET <SOFFSETValue>:
SOFFSETValue exceeds the range.
}
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 4877a32..a9222a8 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
@@ -103,11 +103,11 @@ public class EnvironmentUtils {
}
//try jmx connection
try {
- JMXServiceURL url =
- new
JMXServiceURL("service:jmx:rmi:///jndi/rmi://localhost:31999/jmxrmi");
- JMXConnector jmxConnector = JMXConnectorFactory.connect(url);
+ JMXServiceURL url =
+ new
JMXServiceURL("service:jmx:rmi:///jndi/rmi://localhost:31999/jmxrmi");
+ JMXConnector jmxConnector = JMXConnectorFactory.connect(url);
logger.error("stop JMX failed. 31999 can be connected now.");
- jmxConnector.close();
+ jmxConnector.close();
} catch (IOException e) {
//do nothing
}
@@ -130,7 +130,6 @@ public class EnvironmentUtils {
IoTDBDescriptor.getInstance().getConfig().setReadOnly(false);
-
// clean cache
if (config.isMetaDataCacheEnable()) {
ChunkMetadataCache.getInstance().clear();
@@ -206,12 +205,12 @@ public class EnvironmentUtils {
createAllDir();
// disable the system monitor
config.setEnableStatMonitor(false);
- TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true);
+ TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true,
1024, 0);
TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID);
}
public static void stopDaemon() {
- if(daemon != null) {
+ if (daemon != null) {
daemon.stop();
}
}
@@ -223,7 +222,7 @@ public class EnvironmentUtils {
}
public static void activeDaemon() {
- if(daemon != null) {
+ if (daemon != null) {
daemon.active();
}
}
diff --git
a/spark-iotdb-connector/src/test/scala/org/apache/iotdb/spark/db/EnvironmentUtils.java
b/spark-iotdb-connector/src/test/scala/org/apache/iotdb/spark/db/EnvironmentUtils.java
index 75bca6c..f54c6af 100644
---
a/spark-iotdb-connector/src/test/scala/org/apache/iotdb/spark/db/EnvironmentUtils.java
+++
b/spark-iotdb-connector/src/test/scala/org/apache/iotdb/spark/db/EnvironmentUtils.java
@@ -57,6 +57,7 @@ import java.util.Locale;
* </p>
*/
public class EnvironmentUtils {
+
private static String[] creationSqls = new String[]{
"SET STORAGE GROUP TO root.vehicle.d0",
"SET STORAGE GROUP TO root.vehicle.d1",
@@ -92,7 +93,8 @@ public class EnvironmentUtils {
private static IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
private static DirectoryManager directoryManager =
DirectoryManager.getInstance();
- public static long TEST_QUERY_JOB_ID =
QueryResourceManager.getInstance().assignQueryId(true);
+ public static long TEST_QUERY_JOB_ID = QueryResourceManager.getInstance()
+ .assignQueryId(true, 1024, 0);
public static QueryContext TEST_QUERY_CONTEXT = new
QueryContext(TEST_QUERY_JOB_ID);
private static long oldTsFileThreshold = config.getTsFileSizeThreshold();
@@ -159,16 +161,14 @@ public class EnvironmentUtils {
}
/**
- * disable the system monitor</br>
- * this function should be called before all code in the setup
+ * disable the system monitor</br> this function should be called before all
code in the setup
*/
public static void closeStatMonitor() {
config.setEnableStatMonitor(false);
}
/**
- * disable memory control</br>
- * this function should be called before all code in the setup
+ * disable memory control</br> this function should be called before all
code in the setup
*/
public static void envSetUp() throws StartupException, IOException {
IoTDBDescriptor.getInstance().getConfig().setEnableParameterAdapter(false);
@@ -192,7 +192,7 @@ public class EnvironmentUtils {
StorageEngine.getInstance().reset();
MultiFileLogNodeManager.getInstance().start();
FlushManager.getInstance().start();
- TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true);
+ TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignQueryId(true,
1024, 0);
TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID);
}
@@ -210,7 +210,7 @@ public class EnvironmentUtils {
// create wal
createDir(config.getWalDir());
// create data
- for (String dataDir: config.getDataDirs()) {
+ for (String dataDir : config.getDataDirs()) {
createDir(dataDir);
}
}
