You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2020/03/02 02:12:58 UTC
[incubator-iotdb] 01/01: cherry pick the changes of the distributed
version
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch cherry_pick_cluster
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit be0e107ebfb8f9806865bed600f218ef61556b6f
Author: jt2594838 <jt...@163.com>
AuthorDate: Mon Mar 2 10:12:43 2020 +0800
cherry pick the changes of the distributed version
---
.../resources/conf/iotdb-engine.properties | 1 +
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 123 +++++++++++----------
.../ServerConfigConsistent.java} | 14 ++-
.../org/apache/iotdb/db/engine/StorageEngine.java | 83 ++++++++++++--
.../engine/storagegroup/StorageGroupProcessor.java | 93 +++++++++-------
.../db/engine/storagegroup/TsFileResource.java | 35 +++++-
.../iotdb/db/exception/StorageEngineException.java | 6 +-
.../exception/metadata/IllegalPathException.java | 4 +-
.../db/exception/metadata/MetadataException.java | 4 +
.../exception/metadata/PathNotExistException.java | 3 -
.../metadata/StorageGroupAlreadySetException.java | 3 +-
.../db/exception/query/QueryProcessException.java | 2 +-
.../runtime/StorageEngineFailureException.java | 4 +-
.../StorageGroupProcessorException.java | 2 +-
.../org/apache/iotdb/db/metadata/MManager.java | 68 ++++++++++++
.../java/org/apache/iotdb/db/metadata/MTree.java | 61 ++++++++++
.../main/java/org/apache/iotdb/db/qp/Planner.java | 9 +-
.../apache/iotdb/db/qp/executor/IPlanExecutor.java | 7 +-
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 67 ++++++-----
.../iotdb/db/qp/logical/crud/FilterOperator.java | 10 +-
.../apache/iotdb/db/qp/physical/PhysicalPlan.java | 37 ++++++-
.../iotdb/db/qp/physical/crud/BatchInsertPlan.java | 93 +++++++++++++++-
.../iotdb/db/qp/physical/crud/DeletePlan.java | 10 ++
.../iotdb/db/qp/physical/crud/InsertPlan.java | 25 +++++
.../apache/iotdb/db/qp/physical/sys/CountPlan.java | 3 +
.../db/qp/physical/sys/CreateTimeSeriesPlan.java | 66 +++++++++--
.../db/qp/physical/sys/SetStorageGroupPlan.java | 48 +++++++-
.../db/qp/physical/sys/ShowChildPathsPlan.java | 2 +
.../iotdb/db/qp/strategy/PhysicalGenerator.java | 23 +++-
.../qp/strategy/optimizer/ConcatPathOptimizer.java | 23 +++-
.../db/query/aggregation/AggregateResult.java | 5 +
.../db/query/control/QueryResourceManager.java | 2 +-
.../db/query/dataset/AlignByDeviceDataSet.java | 2 +-
.../dataset/RawQueryDataSetWithValueFilter.java | 6 +-
.../groupby/GroupByWithValueFilterDataSet.java | 1 +
.../groupby/GroupByWithoutValueFilterDataSet.java | 1 +
.../db/query/executor/AggregationExecutor.java | 73 ++++++++----
.../iotdb/db/query/executor/QueryRouter.java | 12 +-
.../db/query/executor/RawDataQueryExecutor.java | 30 +++--
.../db/query/externalsort/ExternalSortJob.java | 3 +-
.../externalsort/ExternalSortJobScheduler.java | 19 ++--
.../externalsort/SimpleExternalSortEngine.java | 18 +--
.../reader/series/SeriesReaderByTimestamp.java | 4 +
.../reader/universal/PriorityMergeReader.java | 2 +-
.../query/timegenerator/ServerTimeGenerator.java | 8 +-
.../java/org/apache/iotdb/db/service/IoTDB.java | 4 +-
.../org/apache/iotdb/db/service/ServiceType.java | 3 +-
.../org/apache/iotdb/db/service/TSServiceImpl.java | 79 +++++--------
.../db/sync/sender/manage/SyncFileManager.java | 1 +
.../apache/iotdb/db/tools/IoTDBDataDirViewer.java | 2 +-
.../iotdb/db/tools/TsFileResourcePrinter.java | 2 +-
.../org/apache/iotdb/db/utils/FileLoaderUtils.java | 2 +-
.../org/apache/iotdb/db/utils/SchemaUtils.java | 85 +++++++++++++-
.../java/org/apache/iotdb/db/utils/TestOnly.java | 4 +
.../writelog/recover/TsFileRecoverPerformer.java | 2 +-
.../apache/iotdb/db/engine/merge/MergeLogTest.java | 3 +-
.../iotdb/db/engine/merge/MergeOverLapTest.java | 5 +-
.../iotdb/db/engine/merge/MergeTaskTest.java | 5 +-
.../apache/iotdb/db/engine/merge/MergeTest.java | 5 +-
.../engine/modification/DeletionFileNodeTest.java | 9 +-
.../db/engine/modification/DeletionQueryTest.java | 6 +-
.../storagegroup/FileNodeManagerBenchmark.java | 7 +-
.../iotdb/db/engine/storagegroup/TTLTest.java | 4 +-
.../iotdb/db/integration/IoTDBMultiSeriesIT.java | 4 +-
.../db/integration/IoTDBSequenceDataQueryIT.java | 5 +-
.../iotdb/db/integration/IoTDBSeriesReaderIT.java | 3 +-
.../apache/iotdb/db/integration/IoTDBTtlIT.java | 1 -
.../iotdb/db/qp/plan/ConcatOptimizerTest.java | 3 +-
.../apache/iotdb/db/qp/plan/PhysicalPlanTest.java | 6 +-
.../apache/iotdb/tsfile/read/TimeValuePair.java | 2 +
.../apache/iotdb/tsfile/read/common/BatchData.java | 12 +-
.../iotdb/tsfile/read/filter/GroupByFilter.java | 8 ++
.../iotdb/tsfile/read/filter/operator/In.java | 4 +
.../apache/iotdb/tsfile/utils/Murmur128Hash.java | 70 +++++++-----
.../iotdb/tsfile/utils/ReadWriteIOUtils.java | 7 +-
.../tsfile/write/schema/MeasurementSchema.java | 1 +
76 files changed, 1074 insertions(+), 395 deletions(-)
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties b/server/src/assembly/resources/conf/iotdb-engine.properties
index eb4e32d..b429d7c 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -22,6 +22,7 @@
####################
metrics_port=8181
+enable_metric_service=false
####################
### RPC Configuration
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 8e8ed0f..deb6c2b 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
@@ -47,6 +47,9 @@ public class IoTDBConfig {
* Port which the metrics service listens to.
*/
private int metricsPort = 8181;
+
+ private boolean enableMetricService = false;
+
private String rpcAddress = "0.0.0.0";
/**
@@ -494,7 +497,7 @@ public class IoTDBConfig {
return memtableNumInEachStorageGroup;
}
- public void setMemtableNumInEachStorageGroup(int memtableNumInEachStorageGroup) {
+ void setMemtableNumInEachStorageGroup(int memtableNumInEachStorageGroup) {
this.memtableNumInEachStorageGroup = memtableNumInEachStorageGroup;
}
@@ -610,6 +613,14 @@ public class IoTDBConfig {
return dataDirs;
}
+ public boolean isEnableMetricService() {
+ return enableMetricService;
+ }
+
+ void setEnableMetricService(boolean enableMetricService) {
+ this.enableMetricService = enableMetricService;
+ }
+
void setDataDirs(String[] dataDirs) {
this.dataDirs = dataDirs;
}
@@ -618,7 +629,7 @@ public class IoTDBConfig {
return metricsPort;
}
- public void setMetricsPort(int metricsPort) {
+ void setMetricsPort(int metricsPort) {
this.metricsPort = metricsPort;
}
@@ -642,7 +653,7 @@ public class IoTDBConfig {
return timestampPrecision;
}
- public void setTimestampPrecision(String timestampPrecision) {
+ void setTimestampPrecision(String timestampPrecision) {
this.timestampPrecision = timestampPrecision;
}
@@ -690,7 +701,7 @@ public class IoTDBConfig {
return syncDir;
}
- public void setSyncDir(String syncDir) {
+ void setSyncDir(String syncDir) {
this.syncDir = syncDir;
}
@@ -698,7 +709,7 @@ public class IoTDBConfig {
return queryDir;
}
- public void setQueryDir(String queryDir) {
+ void setQueryDir(String queryDir) {
this.queryDir = queryDir;
}
@@ -778,7 +789,7 @@ public class IoTDBConfig {
return rpcMaxConcurrentClientNum;
}
- public void setRpcMaxConcurrentClientNum(int rpcMaxConcurrentClientNum) {
+ void setRpcMaxConcurrentClientNum(int rpcMaxConcurrentClientNum) {
this.rpcMaxConcurrentClientNum = rpcMaxConcurrentClientNum;
}
@@ -822,7 +833,7 @@ public class IoTDBConfig {
this.syncServerPort = syncServerPort;
}
- public String getLanguageVersion() {
+ String getLanguageVersion() {
return languageVersion;
}
@@ -834,7 +845,7 @@ public class IoTDBConfig {
return baseDir;
}
- public void setBaseDir(String baseDir) {
+ void setBaseDir(String baseDir) {
this.baseDir = baseDir;
}
@@ -890,7 +901,7 @@ public class IoTDBConfig {
return mergeMemoryBudget;
}
- public void setMergeMemoryBudget(long mergeMemoryBudget) {
+ void setMergeMemoryBudget(long mergeMemoryBudget) {
this.mergeMemoryBudget = mergeMemoryBudget;
}
@@ -898,7 +909,7 @@ public class IoTDBConfig {
return mergeThreadNum;
}
- public void setMergeThreadNum(int mergeThreadNum) {
+ void setMergeThreadNum(int mergeThreadNum) {
this.mergeThreadNum = mergeThreadNum;
}
@@ -906,7 +917,7 @@ public class IoTDBConfig {
return continueMergeAfterReboot;
}
- public void setContinueMergeAfterReboot(boolean continueMergeAfterReboot) {
+ void setContinueMergeAfterReboot(boolean continueMergeAfterReboot) {
this.continueMergeAfterReboot = continueMergeAfterReboot;
}
@@ -914,7 +925,7 @@ public class IoTDBConfig {
return mergeIntervalSec;
}
- public void setMergeIntervalSec(long mergeIntervalSec) {
+ void setMergeIntervalSec(long mergeIntervalSec) {
this.mergeIntervalSec = mergeIntervalSec;
}
@@ -934,11 +945,11 @@ public class IoTDBConfig {
this.allocateMemoryForWrite = allocateMemoryForWrite;
}
- public long getAllocateMemoryForRead() {
+ long getAllocateMemoryForRead() {
return allocateMemoryForRead;
}
- public void setAllocateMemoryForRead(long allocateMemoryForRead) {
+ void setAllocateMemoryForRead(long allocateMemoryForRead) {
this.allocateMemoryForRead = allocateMemoryForRead;
}
@@ -946,7 +957,7 @@ public class IoTDBConfig {
return enableExternalSort;
}
- public void setEnableExternalSort(boolean enableExternalSort) {
+ void setEnableExternalSort(boolean enableExternalSort) {
this.enableExternalSort = enableExternalSort;
}
@@ -954,7 +965,7 @@ public class IoTDBConfig {
return externalSortThreshold;
}
- public void setExternalSortThreshold(int externalSortThreshold) {
+ void setExternalSortThreshold(int externalSortThreshold) {
this.externalSortThreshold = externalSortThreshold;
}
@@ -970,7 +981,7 @@ public class IoTDBConfig {
return performanceStatDisplayInterval;
}
- public void setPerformanceStatDisplayInterval(long performanceStatDisplayInterval) {
+ void setPerformanceStatDisplayInterval(long performanceStatDisplayInterval) {
this.performanceStatDisplayInterval = performanceStatDisplayInterval;
}
@@ -978,7 +989,7 @@ public class IoTDBConfig {
return performanceStatMemoryInKB;
}
- public void setPerformanceStatMemoryInKB(int performanceStatMemoryInKB) {
+ void setPerformanceStatMemoryInKB(int performanceStatMemoryInKB) {
this.performanceStatMemoryInKB = performanceStatMemoryInKB;
}
@@ -986,7 +997,7 @@ public class IoTDBConfig {
return forceFullMerge;
}
- public void setForceFullMerge(boolean forceFullMerge) {
+ void setForceFullMerge(boolean forceFullMerge) {
this.forceFullMerge = forceFullMerge;
}
@@ -1019,7 +1030,7 @@ public class IoTDBConfig {
return mergeChunkSubThreadNum;
}
- public void setMergeChunkSubThreadNum(int mergeChunkSubThreadNum) {
+ void setMergeChunkSubThreadNum(int mergeChunkSubThreadNum) {
this.mergeChunkSubThreadNum = mergeChunkSubThreadNum;
}
@@ -1027,7 +1038,7 @@ public class IoTDBConfig {
return mergeFileSelectionTimeBudget;
}
- public void setMergeFileSelectionTimeBudget(long mergeFileSelectionTimeBudget) {
+ void setMergeFileSelectionTimeBudget(long mergeFileSelectionTimeBudget) {
this.mergeFileSelectionTimeBudget = mergeFileSelectionTimeBudget;
}
@@ -1035,7 +1046,7 @@ public class IoTDBConfig {
return rpcThriftCompressionEnable;
}
- public void setRpcThriftCompressionEnable(boolean rpcThriftCompressionEnable) {
+ void setRpcThriftCompressionEnable(boolean rpcThriftCompressionEnable) {
this.rpcThriftCompressionEnable = rpcThriftCompressionEnable;
}
@@ -1051,7 +1062,7 @@ public class IoTDBConfig {
return allocateMemoryForFileMetaDataCache;
}
- public void setAllocateMemoryForFileMetaDataCache(long allocateMemoryForFileMetaDataCache) {
+ void setAllocateMemoryForFileMetaDataCache(long allocateMemoryForFileMetaDataCache) {
this.allocateMemoryForFileMetaDataCache = allocateMemoryForFileMetaDataCache;
}
@@ -1095,7 +1106,7 @@ public class IoTDBConfig {
this.watermarkBitString = watermarkBitString;
}
- public String getWatermarkMethod() {
+ String getWatermarkMethod() {
return this.watermarkMethod;
}
@@ -1115,7 +1126,7 @@ public class IoTDBConfig {
return Integer.parseInt(getWatermarkParamValue("embed_lsb_num", "5"));
}
- public String getWatermarkParamValue(String key, String defaultValue) {
+ private String getWatermarkParamValue(String key, String defaultValue) {
String res = getWatermarkParamValue(key);
if (res != null) {
return res;
@@ -1123,7 +1134,7 @@ public class IoTDBConfig {
return defaultValue;
}
- public String getWatermarkParamValue(String key) {
+ private String getWatermarkParamValue(String key) {
String pattern = key + "=(\\w*)";
Pattern r = Pattern.compile(pattern);
Matcher m = r.matcher(watermarkMethod);
@@ -1145,7 +1156,7 @@ public class IoTDBConfig {
return defaultStorageGroupLevel;
}
- public void setDefaultStorageGroupLevel(int defaultStorageGroupLevel) {
+ void setDefaultStorageGroupLevel(int defaultStorageGroupLevel) {
this.defaultStorageGroupLevel = defaultStorageGroupLevel;
}
@@ -1157,7 +1168,7 @@ public class IoTDBConfig {
this.defaultBooleanEncoding = defaultBooleanEncoding;
}
- public void setDefaultBooleanEncoding(String defaultBooleanEncoding) {
+ void setDefaultBooleanEncoding(String defaultBooleanEncoding) {
this.defaultBooleanEncoding = TSEncoding.valueOf(defaultBooleanEncoding);
}
@@ -1169,7 +1180,7 @@ public class IoTDBConfig {
this.defaultInt32Encoding = defaultInt32Encoding;
}
- public void setDefaultInt32Encoding(String defaultInt32Encoding) {
+ void setDefaultInt32Encoding(String defaultInt32Encoding) {
this.defaultInt32Encoding = TSEncoding.valueOf(defaultInt32Encoding);
}
@@ -1181,7 +1192,7 @@ public class IoTDBConfig {
this.defaultInt64Encoding = defaultInt64Encoding;
}
- public void setDefaultInt64Encoding(String defaultInt64Encoding) {
+ void setDefaultInt64Encoding(String defaultInt64Encoding) {
this.defaultInt64Encoding = TSEncoding.valueOf(defaultInt64Encoding);
}
@@ -1193,7 +1204,7 @@ public class IoTDBConfig {
this.defaultFloatEncoding = defaultFloatEncoding;
}
- public void setDefaultFloatEncoding(String defaultFloatEncoding) {
+ void setDefaultFloatEncoding(String defaultFloatEncoding) {
this.defaultFloatEncoding = TSEncoding.valueOf(defaultFloatEncoding);
}
@@ -1205,7 +1216,7 @@ public class IoTDBConfig {
this.defaultDoubleEncoding = defaultDoubleEncoding;
}
- public void setDefaultDoubleEncoding(String defaultDoubleEncoding) {
+ void setDefaultDoubleEncoding(String defaultDoubleEncoding) {
this.defaultDoubleEncoding = TSEncoding.valueOf(defaultDoubleEncoding);
}
@@ -1217,7 +1228,7 @@ public class IoTDBConfig {
this.defaultTextEncoding = defaultTextEncoding;
}
- public void setDefaultTextEncoding(String defaultTextEncoding) {
+ void setDefaultTextEncoding(String defaultTextEncoding) {
this.defaultTextEncoding = TSEncoding.valueOf(defaultTextEncoding);
}
@@ -1229,27 +1240,27 @@ public class IoTDBConfig {
this.systemFileStorageFs = FSType.valueOf(systemFileStorageFs);
}
- public FSType getTsFileStorageFs() {
+ FSType getTsFileStorageFs() {
return tsFileStorageFs;
}
- public void setTsFileStorageFs(String tsFileStorageFs) {
+ void setTsFileStorageFs(String tsFileStorageFs) {
this.tsFileStorageFs = FSType.valueOf(tsFileStorageFs);
}
- public String getCoreSitePath() {
+ String getCoreSitePath() {
return coreSitePath;
}
- public void setCoreSitePath(String coreSitePath) {
+ void setCoreSitePath(String coreSitePath) {
this.coreSitePath = coreSitePath;
}
- public String getHdfsSitePath() {
+ String getHdfsSitePath() {
return hdfsSitePath;
}
- public void setHdfsSitePath(String hdfsSitePath) {
+ void setHdfsSitePath(String hdfsSitePath) {
this.hdfsSitePath = hdfsSitePath;
}
@@ -1261,15 +1272,15 @@ public class IoTDBConfig {
return hdfsIp;
}
- public void setHdfsIp(String[] hdfsIp) {
+ void setHdfsIp(String[] hdfsIp) {
this.hdfsIp = String.join(",", hdfsIp);
}
- public String getHdfsPort() {
+ String getHdfsPort() {
return hdfsPort;
}
- public void setHdfsPort(String hdfsPort) {
+ void setHdfsPort(String hdfsPort) {
this.hdfsPort = hdfsPort;
}
@@ -1277,15 +1288,15 @@ public class IoTDBConfig {
return upgradeThreadNum;
}
- public void setUpgradeThreadNum(int upgradeThreadNum) {
+ void setUpgradeThreadNum(int upgradeThreadNum) {
this.upgradeThreadNum = upgradeThreadNum;
}
- public String getDfsNameServices() {
+ String getDfsNameServices() {
return dfsNameServices;
}
- public void setDfsNameServices(String dfsNameServices) {
+ void setDfsNameServices(String dfsNameServices) {
this.dfsNameServices = dfsNameServices;
}
@@ -1297,47 +1308,47 @@ public class IoTDBConfig {
return dfsHaNamenodes;
}
- public void setDfsHaNamenodes(String[] dfsHaNamenodes) {
+ void setDfsHaNamenodes(String[] dfsHaNamenodes) {
this.dfsHaNamenodes = String.join(",", dfsHaNamenodes);
}
- public boolean isDfsHaAutomaticFailoverEnabled() {
+ boolean isDfsHaAutomaticFailoverEnabled() {
return dfsHaAutomaticFailoverEnabled;
}
- public void setDfsHaAutomaticFailoverEnabled(boolean dfsHaAutomaticFailoverEnabled) {
+ void setDfsHaAutomaticFailoverEnabled(boolean dfsHaAutomaticFailoverEnabled) {
this.dfsHaAutomaticFailoverEnabled = dfsHaAutomaticFailoverEnabled;
}
- public String getDfsClientFailoverProxyProvider() {
+ String getDfsClientFailoverProxyProvider() {
return dfsClientFailoverProxyProvider;
}
- public void setDfsClientFailoverProxyProvider(String dfsClientFailoverProxyProvider) {
+ void setDfsClientFailoverProxyProvider(String dfsClientFailoverProxyProvider) {
this.dfsClientFailoverProxyProvider = dfsClientFailoverProxyProvider;
}
- public boolean isUseKerberos() {
+ boolean isUseKerberos() {
return useKerberos;
}
- public void setUseKerberos(boolean useKerberos) {
+ void setUseKerberos(boolean useKerberos) {
this.useKerberos = useKerberos;
}
- public String getKerberosKeytabFilePath() {
+ String getKerberosKeytabFilePath() {
return kerberosKeytabFilePath;
}
- public void setKerberosKeytabFilePath(String kerberosKeytabFilePath) {
+ void setKerberosKeytabFilePath(String kerberosKeytabFilePath) {
this.kerberosKeytabFilePath = kerberosKeytabFilePath;
}
- public String getKerberosPrincipal() {
+ String getKerberosPrincipal() {
return kerberosPrincipal;
}
- public void setKerberosPrincipal(String kerberosPrincipal) {
+ void setKerberosPrincipal(String kerberosPrincipal) {
this.kerberosPrincipal = kerberosPrincipal;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/TestOnly.java b/server/src/main/java/org/apache/iotdb/db/conf/ServerConfigConsistent.java
similarity index 76%
copy from server/src/main/java/org/apache/iotdb/db/utils/TestOnly.java
copy to server/src/main/java/org/apache/iotdb/db/conf/ServerConfigConsistent.java
index d4058ba..18ad0a4 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/TestOnly.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/ServerConfigConsistent.java
@@ -15,17 +15,23 @@
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
- *
*/
-package org.apache.iotdb.db.utils;
+package org.apache.iotdb.db.conf;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
-@Target({ElementType.METHOD, ElementType.CONSTRUCTOR})
+/**
+ * Configurations using this annotation should be the same across all nodes in a cluster.
+ */
+@Target({ElementType.FIELD})
@Retention(RetentionPolicy.SOURCE)
-public @interface TestOnly {
+public @interface ServerConfigConsistent {
+ //TODO#IOTDB-436: the restarted server should check the
+ // configuration consistency as before.
+
}
+
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
index cf99279..ac23eef 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
@@ -23,8 +23,10 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.ConcurrentModificationException;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
@@ -38,6 +40,7 @@ import org.apache.commons.io.FileUtils;
import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.ServerConfigConsistent;
import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy;
import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy.DirectFlushPolicy;
@@ -91,15 +94,41 @@ public class StorageEngine implements IService {
private static final ExecutorService recoveryThreadPool = IoTDBThreadPoolFactory
.newFixedThreadPool(Runtime.getRuntime().availableProcessors(), "Recovery-Thread-Pool");
- private static final StorageEngine INSTANCE = new StorageEngine();
+ static class InstanceHolder {
+ private static final StorageEngine INSTANCE = new StorageEngine();
+ }
public static StorageEngine getInstance() {
- return INSTANCE;
+ return InstanceHolder.INSTANCE;
}
private ScheduledExecutorService ttlCheckThread;
private TsFileFlushPolicy fileFlushPolicy = new DirectFlushPolicy();
+ /**
+ * Time range for dividing storage group, the time unit is the same with IoTDB's TimestampPrecision
+ */
+ @ServerConfigConsistent
+ static long timePartitionInterval;
+ static {
+ // build time Interval to divide time partition
+ String timePrecision = IoTDBDescriptor.getInstance().getConfig().getTimestampPrecision();
+ switch (timePrecision) {
+ case "ns":
+ timePartitionInterval = IoTDBDescriptor.getInstance().
+ getConfig().getPartitionInterval() * 1000_000_000L;
+ break;
+ case "us":
+ timePartitionInterval = IoTDBDescriptor.getInstance().
+ getConfig().getPartitionInterval() * 1000_000L;
+ break;
+ default:
+ timePartitionInterval = IoTDBDescriptor.getInstance().
+ getConfig().getPartitionInterval() * 1000;
+ break;
+ }
+ }
+
private StorageEngine() {
logger = LoggerFactory.getLogger(StorageEngine.class);
systemDir = FilePathUtils.regularizePath(config.getSystemDir()) + "storage_groups";
@@ -109,7 +138,6 @@ public class StorageEngine implements IService {
} catch (IOException e) {
throw new StorageEngineFailureException(e);
}
-
// recover upgrade process
UpgradeUtils.recoverUpgrade();
/*
@@ -160,14 +188,16 @@ public class StorageEngine implements IService {
@Override
public void stop() {
syncCloseAllProcessor();
- ttlCheckThread.shutdownNow();
+ if (ttlCheckThread != null) {
+ ttlCheckThread.shutdownNow();
+ try {
+ ttlCheckThread.awaitTermination(30, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ logger.warn("TTL check thread still doesn't exit after 30s");
+ }
+ }
recoveryThreadPool.shutdownNow();
this.reset();
- try {
- ttlCheckThread.awaitTermination(30, TimeUnit.SECONDS);
- } catch (InterruptedException e) {
- logger.warn("TTL check thread still doesn't exit after 30s");
- }
}
@Override
@@ -176,7 +206,7 @@ public class StorageEngine implements IService {
}
public StorageGroupProcessor getProcessor(String path) throws StorageEngineException {
- String storageGroupName = "";
+ String storageGroupName;
try {
storageGroupName = MManager.getInstance().getStorageGroupName(path);
StorageGroupProcessor processor;
@@ -490,4 +520,37 @@ public class StorageEngine implements IService {
return file.getParentFile().getParentFile().getName();
}
+ /**
+ *
+ * @return TsFiles (seq or unseq) grouped by their storage group.
+ */
+ public Map<String, List<TsFileResource>> getAllClosedStorageGroupTsFile() {
+ Map<String, List<TsFileResource>> ret = new HashMap<>();
+ for (Entry<String, StorageGroupProcessor> entry : processorMap
+ .entrySet()) {
+ ret.computeIfAbsent(entry.getKey(), sg -> new ArrayList<>()).addAll(entry.getValue().getSequenceFileTreeSet());
+ ret.get(entry.getKey()).addAll(entry.getValue().getUnSequenceFileList());
+ ret.get(entry.getKey()).removeIf(file -> !file.isClosed());
+ }
+ return ret;
+ }
+
+ public void setFileFlushPolicy(TsFileFlushPolicy fileFlushPolicy) {
+ this.fileFlushPolicy = fileFlushPolicy;
+ }
+
+ public boolean isFileAlreadyExist(TsFileResource tsFileResource, String storageGroup) {
+ // TODO-Cluster#350: integrate with time partitioning
+ StorageGroupProcessor processor = processorMap.get(storageGroup);
+ return processor != null && processor.isFileAlreadyExist(tsFileResource);
+ }
+
+ public static long getTimePartitionInterval() {
+ return timePartitionInterval;
+ }
+
+ public static long fromTimeToTimePartition(long time) {
+
+ return time / timePartitionInterval;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index edaa0c8..fca11ae 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -44,6 +44,7 @@ import org.apache.commons.io.FileUtils;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy;
import org.apache.iotdb.db.engine.merge.manage.MergeManager;
@@ -75,7 +76,6 @@ import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.QueryFileManager;
import org.apache.iotdb.db.utils.CopyOnReadLinkedList;
-import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.db.utils.UpgradeUtils;
import org.apache.iotdb.db.writelog.recover.TsFileRecoverPerformer;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -143,10 +143,7 @@ public class StorageGroupProcessor {
* time partition id in the storage group -> tsFileProcessor for this time partition
*/
private final TreeMap<Long, TsFileProcessor> workUnsequenceTsFileProcessors = new TreeMap<>();
- /**
- * Time range for dividing storage group, unit is second
- */
- private long partitionIntervalForStorageGroup;
+
/**
* the schema of time series that belong this storage group
*/
@@ -232,23 +229,6 @@ public class StorageGroupProcessor {
storageGroupSysDir.getPath());
}
- // build time Interval to divide time partition
- String timePrecision = IoTDBDescriptor.getInstance().getConfig().getTimestampPrecision();
- switch (timePrecision) {
- case "ns":
- partitionIntervalForStorageGroup = IoTDBDescriptor.getInstance().
- getConfig().getPartitionInterval() * 1000_000_000L;
- break;
- case "us":
- partitionIntervalForStorageGroup = IoTDBDescriptor.getInstance().
- getConfig().getPartitionInterval() * 1000_000L;
- break;
- default:
- partitionIntervalForStorageGroup = IoTDBDescriptor.getInstance().
- getConfig().getPartitionInterval() * 1000;
- break;
- }
-
recover();
}
@@ -317,7 +297,7 @@ public class StorageGroupProcessor {
// just find any time of device
Iterator<Long> iterator = startTimeMap.values().iterator();
if (iterator.hasNext()) {
- return fromTimeToTimePartition(iterator.next());
+ return StorageEngine.fromTimeToTimePartition(iterator.next());
}
return -1;
@@ -490,7 +470,7 @@ public class StorageGroupProcessor {
writeLock();
try {
// init map
- long timePartitionId = fromTimeToTimePartition(insertPlan.getTime());
+ long timePartitionId = StorageEngine.fromTimeToTimePartition(insertPlan.getTime());
latestTimeForEachDevice.computeIfAbsent(timePartitionId, l -> new HashMap<>())
.putIfAbsent(insertPlan.getDeviceId(), Long.MIN_VALUE);
latestFlushedTimeForEachDevice.computeIfAbsent(timePartitionId, id -> new HashMap<>())
@@ -531,7 +511,7 @@ public class StorageGroupProcessor {
// before is first start point
int before = loc;
// before time partition
- long beforeTimePartition = fromTimeToTimePartition(batchInsertPlan.getTimes()[before]);
+ long beforeTimePartition = StorageEngine.fromTimeToTimePartition(batchInsertPlan.getTimes()[before]);
// init map
long lastFlushTime = latestFlushedTimeForEachDevice.
computeIfAbsent(beforeTimePartition, id -> new HashMap<>()).
@@ -540,7 +520,7 @@ public class StorageGroupProcessor {
boolean isSequence = false;
while (loc < batchInsertPlan.getRowCount()) {
long time = batchInsertPlan.getTimes()[loc];
- long curTimePartition = fromTimeToTimePartition(time);
+ long curTimePartition = StorageEngine.fromTimeToTimePartition(time);
results[loc] = TSStatusCode.SUCCESS_STATUS.getStatusCode();
// start next partition
if (curTimePartition != beforeTimePartition) {
@@ -636,7 +616,7 @@ public class StorageGroupProcessor {
throws QueryProcessException {
TsFileProcessor tsFileProcessor;
boolean result;
- long timePartitionId = fromTimeToTimePartition(insertPlan.getTime());
+ long timePartitionId = StorageEngine.fromTimeToTimePartition(insertPlan.getTime());
tsFileProcessor = getOrCreateTsFileProcessor(timePartitionId, sequence);
@@ -735,10 +715,6 @@ public class StorageGroupProcessor {
}
- private long fromTimeToTimePartition(long time) {
-
- return time / partitionIntervalForStorageGroup;
- }
private TsFileProcessor createTsFileProcessor(boolean sequence, long timePartitionId)
throws IOException, DiskSpaceInsufficientException {
@@ -784,6 +760,7 @@ public class StorageGroupProcessor {
}
private String getNewTsFileName(long time, long version, int mergeCnt) {
+ allDirectFileVersions.add(version);
return time + IoTDBConstant.TSFILE_NAME_SEPARATOR + version
+ IoTDBConstant.TSFILE_NAME_SEPARATOR + mergeCnt + TSFILE_SUFFIX;
}
@@ -1165,7 +1142,7 @@ public class StorageGroupProcessor {
}
// time partition to divide storage group
- long timePartitionId = fromTimeToTimePartition(timestamp);
+ long timePartitionId = StorageEngine.fromTimeToTimePartition(timestamp);
// write log
if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
DeletePlan deletionPlan = new DeletePlan(timestamp, new Path(deviceId, measurementId));
@@ -1616,6 +1593,41 @@ public class StorageGroupProcessor {
}
/**
+ * If the historical versions of a file is a sub-set of the given file's, remove it to reduce
+ * unnecessary merge. Only used when the file sender and the receiver share the same file
+ * close policy.
+ * @param resource
+ */
+ public void removeFullyOverlapFiles(TsFileResource resource) {
+ writeLock();
+ closeQueryLock.writeLock().lock();
+ try {
+ Iterator<TsFileResource> iterator = sequenceFileTreeSet.iterator();
+ removeFullyOverlapFiles(resource, iterator);
+
+ iterator = unSequenceFileList.iterator();
+ removeFullyOverlapFiles(resource, iterator);
+ } finally {
+ closeQueryLock.writeLock().unlock();
+ writeUnlock();
+ }
+ }
+
+ private void removeFullyOverlapFiles(TsFileResource resource, Iterator<TsFileResource> iterator) {
+ while (iterator.hasNext()) {
+ TsFileResource seqFile = iterator.next();
+ if (resource.getHistoricalVersions().containsAll(seqFile.getHistoricalVersions())
+ && !resource.getHistoricalVersions().equals(seqFile.getHistoricalVersions())) {
+ if (seqFile.getWriteQueryLock().writeLock().tryLock()) {
+ iterator.remove();
+ seqFile.remove();
+ seqFile.getWriteQueryLock().writeLock().unlock();
+ }
+ }
+ }
+ }
+
+ /**
* Get an appropriate filename to ensure the order between files. The tsfile is named after
* ({systemTime}-{versionNum}-{mergeNum}.tsfile).
* <p>
@@ -1679,7 +1691,7 @@ public class StorageGroupProcessor {
for (Entry<String, Long> entry : newTsFileResource.getEndTimeMap().entrySet()) {
String device = entry.getKey();
long endTime = newTsFileResource.getEndTimeMap().get(device);
- long timePartitionId = fromTimeToTimePartition(endTime);
+ long timePartitionId = StorageEngine.fromTimeToTimePartition(endTime);
if (!latestTimeForEachDevice.computeIfAbsent(timePartitionId, id -> new HashMap<>())
.containsKey(device)
|| latestTimeForEachDevice.get(timePartitionId).get(device) < endTime) {
@@ -1692,7 +1704,7 @@ public class StorageGroupProcessor {
if (!latestFlushTimeForPartition.containsKey(device)
|| latestFlushTimeForPartition.get(device) < endTime) {
latestFlushedTimeForEachDevice
- .computeIfAbsent(timePartitionId, id -> new HashMap<String, Long>())
+ .computeIfAbsent(timePartitionId, id -> new HashMap<>())
.put(device, endTime);
}
}
@@ -1709,7 +1721,7 @@ public class StorageGroupProcessor {
TsFileResource tsFileResource)
throws TsFileProcessorException, DiskSpaceInsufficientException {
File targetFile;
- long timeRangeId = fromTimeToTimePartition(
+ long timeRangeId = StorageEngine.fromTimeToTimePartition(
tsFileResource.getStartTimeMap().entrySet().iterator().next().getValue());
switch (type) {
case LOAD_UNSEQUENCE:
@@ -1889,12 +1901,10 @@ public class StorageGroupProcessor {
checkFilesTTL();
}
- @TestOnly
public List<TsFileResource> getSequenceFileTreeSet() {
return new ArrayList<>(sequenceFileTreeSet);
}
- @TestOnly
public List<TsFileResource> getUnSequenceFileList() {
return unSequenceFileList;
}
@@ -1909,10 +1919,17 @@ public class StorageGroupProcessor {
void call(TsFileProcessor caller) throws TsFileProcessorException, IOException;
}
+ public String getStorageGroupName() {
+ return storageGroupName;
+ }
+
+ public boolean isFileAlreadyExist(TsFileResource tsFileResource) {
+ return allDirectFileVersions.containsAll(tsFileResource.getHistoricalVersions());
+ }
+
@FunctionalInterface
public interface UpdateEndTimeCallBack {
boolean call(TsFileProcessor caller);
}
-
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
index e9356db..58ecd56 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
@@ -60,12 +60,12 @@ public class TsFileResource {
/**
* device -> start time
*/
- private Map<String, Long> startTimeMap;
+ protected Map<String, Long> startTimeMap;
/**
* device -> end time. It is null if it's an unsealed sequence tsfile
*/
- private Map<String, Long> endTimeMap;
+ protected Map<String, Long> endTimeMap;
public TsFileProcessor getProcessor() {
return processor;
@@ -101,6 +101,25 @@ public class TsFileResource {
private FSFactory fsFactory = FSFactoryProducer.getFSFactory();
+ public TsFileResource() {
+ }
+
+ public TsFileResource(TsFileResource other) {
+ this.file = other.file;
+ this.startTimeMap = other.startTimeMap;
+ this.endTimeMap = other.endTimeMap;
+ this.processor = other.processor;
+ this.modFile = other.modFile;
+ this.closed = other.closed;
+ this.deleted = other.deleted;
+ this.isMerging = other.isMerging;
+ this.chunkMetaDataList = other.chunkMetaDataList;
+ this.readOnlyMemChunk = other.readOnlyMemChunk;
+ this.writeQueryLock = other.writeQueryLock;
+ this.fsFactory = other.fsFactory;
+ this.historicalVersions = other.historicalVersions;
+ }
+
/**
* for sealed TsFile, call setClosed to close TsFileResource
*/
@@ -162,7 +181,7 @@ public class TsFileResource {
fsFactory.moveFile(src, dest);
}
- public void deSerialize() throws IOException {
+ public void deserialize() throws IOException {
try (InputStream inputStream = fsFactory.getBufferedInputStream(
file + RESOURCE_SUFFIX)) {
int size = ReadWriteIOUtils.readInt(inputStream);
@@ -363,11 +382,19 @@ public class TsFileResource {
return false;
}
+ protected void setStartTimeMap(Map<String, Long> startTimeMap) {
+ this.startTimeMap = startTimeMap;
+ }
+
+ protected void setEndTimeMap(Map<String, Long> endTimeMap) {
+ this.endTimeMap = endTimeMap;
+ }
+
/**
* set a file flag indicating that the file is being closed, so during recovery we could know we
* should close the file.
*/
- public void setCloseFlag() {
+ void setCloseFlag() {
try {
new File(file.getAbsoluteFile() + CLOSING_SUFFIX).createNewFile();
} catch (IOException e) {
diff --git a/server/src/main/java/org/apache/iotdb/db/exception/StorageEngineException.java b/server/src/main/java/org/apache/iotdb/db/exception/StorageEngineException.java
index 7944bd4..bd03a29 100644
--- a/server/src/main/java/org/apache/iotdb/db/exception/StorageEngineException.java
+++ b/server/src/main/java/org/apache/iotdb/db/exception/StorageEngineException.java
@@ -24,6 +24,10 @@ public class StorageEngineException extends IoTDBException {
private static final long serialVersionUID = 9001649171768311032L;
+ public StorageEngineException(Throwable cause) {
+ super(cause, TSStatusCode.STORAGE_ENGINE_ERROR.getStatusCode());
+ }
+
public StorageEngineException(String message) {
super(message, TSStatusCode.STORAGE_ENGINE_ERROR.getStatusCode());
}
@@ -33,6 +37,6 @@ public class StorageEngineException extends IoTDBException {
}
public StorageEngineException(IoTDBException e) {
- super(e.getMessage(), e.getErrorCode());
+ super(e, e.getErrorCode());
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/exception/metadata/IllegalPathException.java b/server/src/main/java/org/apache/iotdb/db/exception/metadata/IllegalPathException.java
index 8f9ee27..da40603 100644
--- a/server/src/main/java/org/apache/iotdb/db/exception/metadata/IllegalPathException.java
+++ b/server/src/main/java/org/apache/iotdb/db/exception/metadata/IllegalPathException.java
@@ -15,7 +15,6 @@
* 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.metadata;
@@ -23,11 +22,10 @@ package org.apache.iotdb.db.exception.metadata;
import org.apache.iotdb.rpc.TSStatusCode;
public class IllegalPathException extends MetadataException {
-
private static final long serialVersionUID = 2693272249167539978L;
public IllegalPathException(String path) {
super(String.format("%s is not a legal path", path));
errorCode = TSStatusCode.PATH_ILLEGAL.getStatusCode();
}
-}
\ No newline at end of file
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/exception/metadata/MetadataException.java b/server/src/main/java/org/apache/iotdb/db/exception/metadata/MetadataException.java
index dfca520..5d8921c 100644
--- a/server/src/main/java/org/apache/iotdb/db/exception/metadata/MetadataException.java
+++ b/server/src/main/java/org/apache/iotdb/db/exception/metadata/MetadataException.java
@@ -29,6 +29,10 @@ public class MetadataException extends IoTDBException {
private static final long serialVersionUID = 3415275599091623570L;
+ public MetadataException(Throwable cause) {
+ super(cause, TSStatusCode.METADATA_ERROR.getStatusCode());
+ }
+
public MetadataException(Throwable cause, int errorCode) {
super(cause, TSStatusCode.METADATA_ERROR.getStatusCode());
this.errorCode = errorCode;
diff --git a/server/src/main/java/org/apache/iotdb/db/exception/metadata/PathNotExistException.java b/server/src/main/java/org/apache/iotdb/db/exception/metadata/PathNotExistException.java
index fe8e476..7d86ade 100644
--- a/server/src/main/java/org/apache/iotdb/db/exception/metadata/PathNotExistException.java
+++ b/server/src/main/java/org/apache/iotdb/db/exception/metadata/PathNotExistException.java
@@ -15,15 +15,12 @@
* 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.metadata;
public class PathNotExistException extends MetadataException {
- private static final long serialVersionUID = 2693272249167539978L;
-
public PathNotExistException(String path) {
super(String.format("Path [%s] does not exist", path));
}
diff --git a/server/src/main/java/org/apache/iotdb/db/exception/metadata/StorageGroupAlreadySetException.java b/server/src/main/java/org/apache/iotdb/db/exception/metadata/StorageGroupAlreadySetException.java
index c43884a..b196309 100644
--- a/server/src/main/java/org/apache/iotdb/db/exception/metadata/StorageGroupAlreadySetException.java
+++ b/server/src/main/java/org/apache/iotdb/db/exception/metadata/StorageGroupAlreadySetException.java
@@ -15,7 +15,6 @@
* 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.metadata;
@@ -27,4 +26,4 @@ public class StorageGroupAlreadySetException extends MetadataException {
public StorageGroupAlreadySetException(String path) {
super(String.format("%s has already been set to storage group", path));
}
-}
\ No newline at end of file
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/exception/query/QueryProcessException.java b/server/src/main/java/org/apache/iotdb/db/exception/query/QueryProcessException.java
index 9372d15..f8c007c 100644
--- a/server/src/main/java/org/apache/iotdb/db/exception/query/QueryProcessException.java
+++ b/server/src/main/java/org/apache/iotdb/db/exception/query/QueryProcessException.java
@@ -35,6 +35,6 @@ public class QueryProcessException extends IoTDBException {
}
public QueryProcessException(IoTDBException e) {
- super(e.getMessage(), e.getErrorCode());
+ super(e, e.getErrorCode());
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/exception/runtime/StorageEngineFailureException.java b/server/src/main/java/org/apache/iotdb/db/exception/runtime/StorageEngineFailureException.java
index 934ece6..aab283c 100644
--- a/server/src/main/java/org/apache/iotdb/db/exception/runtime/StorageEngineFailureException.java
+++ b/server/src/main/java/org/apache/iotdb/db/exception/runtime/StorageEngineFailureException.java
@@ -24,10 +24,10 @@ public class StorageEngineFailureException extends RuntimeException {
private static final long serialVersionUID = -1197701024139022020L;
public StorageEngineFailureException(Throwable e) {
- super("Create system directory failed! " + e.getMessage());
+ super("Create system directory failed! ", e);
}
public StorageEngineFailureException(String message, Throwable e) {
- super(message + e.getMessage());
+ super(message, e);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/exception/storageGroup/StorageGroupProcessorException.java b/server/src/main/java/org/apache/iotdb/db/exception/storageGroup/StorageGroupProcessorException.java
index 249be85..9bae99d 100644
--- a/server/src/main/java/org/apache/iotdb/db/exception/storageGroup/StorageGroupProcessorException.java
+++ b/server/src/main/java/org/apache/iotdb/db/exception/storageGroup/StorageGroupProcessorException.java
@@ -26,7 +26,7 @@ public class StorageGroupProcessorException extends IoTDBException {
private static final long serialVersionUID = 7373978140952977661L;
public StorageGroupProcessorException(Exception exception) {
- super(exception.getMessage(), TSStatusCode.STORAGE_GROUP_PROCESSOR_ERROR.getStatusCode());
+ super(exception, TSStatusCode.STORAGE_GROUP_PROCESSOR_ERROR.getStatusCode());
}
public StorageGroupProcessorException(String message) {
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 0ef29eb..b34a6a7 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
@@ -24,9 +24,12 @@ import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
+import java.util.Deque;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -39,9 +42,11 @@ import org.apache.iotdb.db.conf.adapter.ActiveTimeSeriesCounter;
import org.apache.iotdb.db.conf.adapter.IoTDBConfigDynamicAdapter;
import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
import org.apache.iotdb.db.exception.ConfigAdjusterException;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.metadata.PathNotExistException;
import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
+import org.apache.iotdb.db.metadata.mnode.LeafMNode;
import org.apache.iotdb.db.metadata.mnode.MNode;
import org.apache.iotdb.db.metadata.mnode.StorageGroupMNode;
import org.apache.iotdb.db.monitor.MonitorConstants;
@@ -841,4 +846,67 @@ public class MManager {
lock.readLock().unlock();
}
}
+
+ public void collectSeries(MNode startingNode, Collection<MeasurementSchema> timeseriesSchemas) {
+ Deque<MNode> nodeDeque = new ArrayDeque<>();
+ nodeDeque.addLast(startingNode);
+ while (!nodeDeque.isEmpty()) {
+ MNode node = nodeDeque.removeFirst();
+ if (node instanceof LeafMNode) {
+ MeasurementSchema nodeSchema = node.getSchema();
+ timeseriesSchemas.add(new MeasurementSchema(node.getFullPath(),
+ nodeSchema.getType(), nodeSchema.getEncodingType(), nodeSchema.getCompressor()));
+ } else if (!node.getChildren().isEmpty()) {
+ nodeDeque.addAll(node.getChildren().values());
+ }
+ }
+ }
+
+ public void collectSeries(String startingPath, List<MeasurementSchema> timeseriesSchemas) {
+ MNode mNode;
+ try {
+ mNode = getNodeByPath(startingPath);
+ } catch (MetadataException e) {
+ return;
+ }
+ collectSeries(mNode, timeseriesSchemas);
+ }
+
+ /**
+ * For a path, infer all storage groups it may belong to.
+ * The path can have wildcards.
+ * Consider the path into two parts: (1) the sub path which can not contain a storage group name and
+ * (2) the sub path which is substring that begin after the storage group name.
+ * (1) Suppose the part of the path can not contain a storage group name (e.g.,
+ * "root".contains("root.sg") == false), then:
+ * If the wildcard is not at the tail, then for each wildcard, only one level will be inferred
+ * and the wildcard will be removed.
+ * If the wildcard is at the tail, then the inference will go on until the storage groups are found
+ * and the wildcard will be kept.
+ * (2) Suppose the path of the path is a substring that begin after the storage group name. (e.g.,
+ * For "root.*.sg1.a.*.b.*" and "root.x.sg1" is a storage group, then this part is "a.*.b.*").
+ * For this part, keep what it is.
+ *
+ * Assuming we have three SGs: root.group1, root.group2, root.area1.group3
+ * Eg1:
+ * for input "root.*", returns ("root.group1", "root.group1.*"), ("root.group2", "root.group2.*")
+ * ("root.area1.group3", "root.area1.group3.*")
+ * Eg2:
+ * for input "root.*.s1", returns ("root.group1", "root.group1.s1"), ("root.group2", "root.group2.s1")
+ *
+ * Eg3:
+ * for input "root.area1.*", returns ("root.area1.group3", "root.area1.group3.*")
+ *
+ *
+ * @param path can be a prefix or a full path.
+ * @return StorageGroupName-FullPath pairs
+ */
+ public Map<String, String> determineStorageGroup(String path) throws IllegalPathException {
+ lock.readLock().lock();
+ try {
+ return mtree.determineStorageGroup(path);
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
}
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 9a07c20..87b6afd 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
@@ -32,6 +32,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeSet;
import java.util.regex.Pattern;
@@ -231,6 +232,7 @@ public class MTree implements Serializable {
*
* @param path Format: root.node(.node)+
*/
+
String deleteTimeseriesAndReturnEmptyStorageGroup(String path) throws MetadataException {
MNode curNode = getNodeByPath(path);
if (!(curNode instanceof LeafMNode)) {
@@ -755,4 +757,63 @@ public class MTree implements Serializable {
}
return res;
}
+
+ Map<String, String> determineStorageGroup(String path) throws IllegalPathException {
+ Map<String, String> paths = new HashMap<>();
+ String[] nodes = MetaUtils.getNodeNames(path);
+ if (nodes.length == 0 || !nodes[0].equals(root.getName())) {
+ throw new IllegalPathException(path);
+ }
+
+ Deque<MNode> nodeStack = new ArrayDeque<>();
+ Deque<Integer> depthStack = new ArrayDeque<>();
+ if (!root.getChildren().isEmpty()) {
+ nodeStack.push(root);
+ depthStack.push(0);
+ }
+
+ while (!nodeStack.isEmpty()) {
+ MNode mNode = nodeStack.removeFirst();
+ int depth = depthStack.removeFirst();
+
+ determineStorageGroup(depth + 1, nodes, mNode, paths, nodeStack, depthStack);
+ }
+ return paths;
+ }
+
+ /**
+ * Try determining the storage group using the children of a mNode. If one child is a storage
+ * group node, put a storageGroupName-fullPath pair into paths. Otherwise put the children that
+ * match the path into the queue and discard other children.
+ */
+ private void determineStorageGroup(int depth, String[] nodes, MNode mNode,
+ Map<String, String> paths, Deque<MNode> nodeStack, Deque<Integer> depthStack) {
+ String currNode = depth >= nodes.length ? PATH_WILDCARD : nodes[depth];
+ for (Entry<String, MNode> entry : mNode.getChildren().entrySet()) {
+ if (!currNode.equals(PATH_WILDCARD) && !currNode.equals(entry.getKey())) {
+ continue;
+ }
+ // this child is desired
+ MNode child = entry.getValue();
+ if (child instanceof StorageGroupMNode) {
+ // we have found one storage group, record it
+ String sgName = child.getFullPath();
+ // concat the remaining path with the storage group name
+ StringBuilder pathWithKnownSG = new StringBuilder(sgName);
+ for (int i = depth + 1; i < nodes.length; i++) {
+ pathWithKnownSG.append(IoTDBConstant.PATH_SEPARATOR).append(nodes[i]);
+ }
+ if (depth >= nodes.length - 1 && currNode.equals(PATH_WILDCARD)) {
+ // the we find the sg at the last node and the last node is a wildcard (find "root
+ // .group1", for "root.*"), also append the wildcard (to make "root.group1.*")
+ pathWithKnownSG.append(IoTDBConstant.PATH_SEPARATOR).append(PATH_WILDCARD);
+ }
+ paths.put(sgName, pathWithKnownSG.toString());
+ } else if (!child.getChildren().isEmpty()) {
+ // push it back so we can traver its children later
+ nodeStack.push(child);
+ depthStack.push(depth);
+ }
+ }
+ }
}
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 19a4d69..bd1089f 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
@@ -41,7 +41,7 @@ import org.apache.iotdb.db.utils.TestOnly;
*/
public class Planner {
- private ParseDriver parseDriver;
+ protected ParseDriver parseDriver;
public Planner() {
this.parseDriver = new ParseDriver();
@@ -70,7 +70,7 @@ public class Planner {
* @return optimized logical operator
* @throws LogicalOptimizeException exception in logical optimizing
*/
- private Operator logicalOptimize(Operator operator)
+ protected Operator logicalOptimize(Operator operator)
throws LogicalOperatorException {
switch (operator.getType()) {
case AUTHOR:
@@ -111,7 +111,7 @@ public class Planner {
*/
private SFWOperator optimizeSFWOperator(SFWOperator root)
throws LogicalOperatorException {
- ConcatPathOptimizer concatPathOptimizer = new ConcatPathOptimizer();
+ ConcatPathOptimizer concatPathOptimizer = getConcatPathOptimizer();
root = (SFWOperator) concatPathOptimizer.transform(root);
FilterOperator filter = root.getFilterOperator();
if (filter == null) {
@@ -127,4 +127,7 @@ public class Planner {
return root;
}
+ protected ConcatPathOptimizer getConcatPathOptimizer() {
+ return new ConcatPathOptimizer();
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
index ec25529..b2466db 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
@@ -20,11 +20,13 @@ package org.apache.iotdb.db.qp.executor;
import java.io.IOException;
import java.sql.SQLException;
-import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
-import org.apache.iotdb.db.qp.physical.crud.*;
+import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
+import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
import org.apache.iotdb.tsfile.read.common.Path;
@@ -89,5 +91,4 @@ public interface IPlanExecutor {
* @return result of each row
*/
Integer[] insertBatch(BatchInsertPlan batchInsertPlan) throws QueryProcessException;
-
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index 01c72a7..25020d3 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -34,6 +34,7 @@ import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TIMESERIES_ENCODING;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TTL;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_USER;
import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_VALUE;
+import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR;
import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX;
import java.io.File;
@@ -62,7 +63,7 @@ import org.apache.iotdb.db.engine.flush.pool.FlushTaskPoolManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
-import org.apache.iotdb.db.exception.query.PathException;
+import org.apache.iotdb.db.exception.metadata.PathNotExistException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.metadata.mnode.InternalMNode;
@@ -127,9 +128,7 @@ import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;
public class PlanExecutor implements IPlanExecutor {
// for data query
- private IQueryRouter queryRouter;
- // for insert
- private StorageEngine storageEngine;
+ protected IQueryRouter queryRouter;
// for system schema
private MManager mManager;
// for administration
@@ -137,7 +136,6 @@ public class PlanExecutor implements IPlanExecutor {
public PlanExecutor() throws QueryProcessException {
queryRouter = new QueryRouter();
- storageEngine = StorageEngine.getInstance();
mManager = MManager.getInstance();
try {
authorizer = LocalFileAuthorizer.getInstance();
@@ -222,7 +220,7 @@ public class PlanExecutor implements IPlanExecutor {
}
}
- private QueryDataSet processDataQuery(QueryPlan queryPlan, QueryContext context)
+ protected QueryDataSet processDataQuery(QueryPlan queryPlan, QueryContext context)
throws StorageEngineException, QueryFilterOptimizationException, QueryProcessException,
IOException {
QueryDataSet queryDataSet;
@@ -316,7 +314,7 @@ public class PlanExecutor implements IPlanExecutor {
return MManager.getInstance().getAllTimeseriesName(path);
}
- private List<String> getNodesList(String schemaPattern, int level) throws MetadataException {
+ protected List<String> getNodesList(String schemaPattern, int level) throws MetadataException {
return MManager.getInstance().getNodesList(schemaPattern, level);
}
@@ -337,7 +335,7 @@ public class PlanExecutor implements IPlanExecutor {
throws MetadataException {
ListDataSet listDataSet = new ListDataSet(Collections.singletonList(new Path(COLUMN_DEVICES)),
Collections.singletonList(TSDataType.TEXT));
- Set<String> devices = MManager.getInstance().getDevices(showDevicesPlan.getPath().toString());
+ Set<String> devices = getDevices(showDevicesPlan.getPath().toString());
for (String s : devices) {
RowRecord record = new RowRecord(0);
Field field = new Field(TSDataType.TEXT);
@@ -348,10 +346,13 @@ public class PlanExecutor implements IPlanExecutor {
return listDataSet;
}
+ protected Set<String> getDevices(String path) throws MetadataException {
+ return MManager.getInstance().getDevices(path);
+ }
+
private QueryDataSet processShowChildPaths(ShowChildPathsPlan showChildPathsPlan)
throws MetadataException {
- Set<String> childPathsList = MManager.getInstance()
- .getChildNodePathInNextLevel(showChildPathsPlan.getPath().toString());
+ Set<String> childPathsList = getPathNextChildren(showChildPathsPlan.getPath().toString());
ListDataSet listDataSet = new ListDataSet(
Collections.singletonList(new Path(COLUMN_CHILD_PATHS)),
Collections.singletonList(TSDataType.TEXT));
@@ -365,6 +366,10 @@ public class PlanExecutor implements IPlanExecutor {
return listDataSet;
}
+ protected Set<String> getPathNextChildren(String path) throws MetadataException {
+ return MManager.getInstance().getChildNodePathInNextLevel(path);
+ }
+
private QueryDataSet processShowStorageGroup() {
ListDataSet listDataSet = new ListDataSet(
Collections.singletonList(new Path(COLUMN_STORAGE_GROUP)),
@@ -390,8 +395,7 @@ public class PlanExecutor implements IPlanExecutor {
new Path(COLUMN_TIMESERIES_COMPRESSION)),
Arrays.asList(TSDataType.TEXT, TSDataType.TEXT, TSDataType.TEXT, TSDataType.TEXT,
TSDataType.TEXT));
- List<String[]> timeseriesList = MManager.getInstance()
- .getAllTimeseriesSchema(timeSeriesPlan.getPath().toString());
+ List<String[]> timeseriesList = getTimeseriesSchemas(timeSeriesPlan.getPath().toString());
for (String[] list : timeseriesList) {
RowRecord record = new RowRecord(0);
for (String s : list) {
@@ -404,6 +408,10 @@ public class PlanExecutor implements IPlanExecutor {
return listDataSet;
}
+ protected List<String[]> getTimeseriesSchemas(String path) throws MetadataException {
+ return MManager.getInstance().getAllTimeseriesSchema(path);
+ }
+
private QueryDataSet processShowTTLQuery(ShowTTLPlan showTTLPlan) {
ListDataSet listDataSet = new ListDataSet(
Arrays.asList(new Path(COLUMN_STORAGE_GROUP), new Path(COLUMN_TTL))
@@ -503,18 +511,12 @@ public class PlanExecutor implements IPlanExecutor {
try {
Set<String> existingPaths = new HashSet<>();
for (Path p : deletePlan.getPaths()) {
- existingPaths.addAll(mManager.getAllTimeseriesName(p.getFullPath()));
+ existingPaths.addAll(getPaths(p.getFullPath()));
}
if (existingPaths.isEmpty()) {
throw new QueryProcessException(
"TimeSeries does not exist and its data cannot be deleted");
}
- for (String onePath : existingPaths) {
- if (!mManager.isPathExist(onePath)) {
- throw new QueryProcessException(String
- .format("TimeSeries %s does not exist and its data cannot be deleted", onePath));
- }
- }
for (String path : existingPaths) {
delete(new Path(path), deletePlan.getDeleteTime());
}
@@ -611,7 +613,7 @@ public class PlanExecutor implements IPlanExecutor {
.createTimeseries(fullPath, schema.getType(), schema.getEncodingType(),
schema.getCompressor(), Collections.emptyMap());
if (result) {
- storageEngine
+ StorageEngine.getInstance()
.addTimeSeries(new Path(fullPath), schema.getType(), schema.getEncodingType(),
schema.getCompressor(), Collections.emptyMap());
}
@@ -684,7 +686,7 @@ public class PlanExecutor implements IPlanExecutor {
String.format("Time series %s does not exist.", path.getFullPath()));
}
mManager.getStorageGroupName(path.getFullPath());
- storageEngine.delete(deviceId, measurementId, timestamp);
+ StorageEngine.getInstance().delete(deviceId, measurementId, timestamp);
} catch (MetadataException | StorageEngineException e) {
throw new QueryProcessException(e);
}
@@ -704,9 +706,7 @@ public class PlanExecutor implements IPlanExecutor {
String measurement = measurementList[i];
if (!node.hasChild(measurement)) {
if (!IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()) {
- throw new QueryProcessException(
- String.format("Current deviceId[%s] does not contain measurement:%s",
- deviceId, measurement));
+ throw new PathNotExistException(deviceId + PATH_SEPARATOR + measurement);
}
TSDataType dataType = TypeInferenceUtils.getPredictedDataType(strValues[i]);
Path path = new Path(deviceId, measurement);
@@ -716,7 +716,7 @@ public class PlanExecutor implements IPlanExecutor {
TSFileDescriptor.getInstance().getConfig().getCompressor(),
Collections.emptyMap());
if (result) {
- storageEngine.addTimeSeries(path, dataType, getDefaultEncoding(dataType));
+ StorageEngine.getInstance().addTimeSeries(path, dataType, getDefaultEncoding(dataType));
}
}
MNode measurementNode = node.getChild(measurement);
@@ -728,8 +728,8 @@ public class PlanExecutor implements IPlanExecutor {
dataTypes[i] = measurementNode.getSchema().getType();
}
insertPlan.setDataTypes(dataTypes);
- storageEngine.insert(insertPlan);
- } catch (PathException | StorageEngineException | MetadataException e) {
+ StorageEngine.getInstance().insert(insertPlan);
+ } catch (StorageEngineException | MetadataException e) {
throw new QueryProcessException(e);
}
}
@@ -782,7 +782,7 @@ public class PlanExecutor implements IPlanExecutor {
TSFileDescriptor.getInstance().getConfig().getCompressor(),
Collections.emptyMap());
if (result) {
- storageEngine
+ StorageEngine.getInstance()
.addTimeSeries(path, dataType, getDefaultEncoding(dataType));
}
}
@@ -800,8 +800,7 @@ public class PlanExecutor implements IPlanExecutor {
measurementNode.getSchema().getType()));
}
}
- return storageEngine.insertBatch(batchInsertPlan);
-
+ return StorageEngine.getInstance().insertBatch(batchInsertPlan);
} catch (StorageEngineException | MetadataException e) {
throw new QueryProcessException(e);
}
@@ -890,7 +889,7 @@ public class PlanExecutor implements IPlanExecutor {
boolean result = mManager
.createTimeseries(path.getFullPath(), dataType, encoding, compressor, props);
if (result) {
- storageEngine.addTimeSeries(path, dataType, encoding, compressor, props);
+ StorageEngine.getInstance().addTimeSeries(path, dataType, encoding, compressor, props);
}
} catch (StorageEngineException | MetadataException e) {
throw new QueryProcessException(e);
@@ -908,7 +907,7 @@ public class PlanExecutor implements IPlanExecutor {
emptyStorageGroups.addAll(mManager.deleteTimeseries(path.toString()));
}
for (String deleteStorageGroup : emptyStorageGroups) {
- storageEngine.deleteAllDataFilesInOneStorageGroup(deleteStorageGroup);
+ StorageEngine.getInstance().deleteAllDataFilesInOneStorageGroup(deleteStorageGroup);
}
} catch (MetadataException e) {
throw new QueryProcessException(e);
@@ -916,7 +915,7 @@ public class PlanExecutor implements IPlanExecutor {
return true;
}
- private boolean setStorageGroup(SetStorageGroupPlan setStorageGroupPlan)
+ public boolean setStorageGroup(SetStorageGroupPlan setStorageGroupPlan)
throws QueryProcessException {
Path path = setStorageGroupPlan.getPath();
try {
@@ -932,7 +931,7 @@ public class PlanExecutor implements IPlanExecutor {
List<String> deletePathList = new ArrayList<>();
try {
for (Path storageGroupPath : deleteStorageGroupPlan.getPaths()) {
- storageEngine.deleteStorageGroup(storageGroupPath.getFullPath());
+ StorageEngine.getInstance().deleteStorageGroup(storageGroupPath.getFullPath());
deletePathList.add(storageGroupPath.getFullPath());
}
mManager.deleteStorageGroups(deletePathList);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/FilterOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/FilterOperator.java
index 67f5946..bc22acb 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/FilterOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/FilterOperator.java
@@ -44,16 +44,16 @@ import org.apache.iotdb.tsfile.utils.StringContainer;
public class FilterOperator extends Operator implements Comparable<FilterOperator> {
// it is the symbol of token. e.g. AND is & and OR is |
- protected String tokenSymbol;
+ String tokenSymbol;
- protected List<FilterOperator> childOperators;
+ private List<FilterOperator> childOperators;
// leaf filter operator means it doesn't have left and right child filterOperator. Leaf filter
// should set FunctionOperator.
- protected boolean isLeaf = false;
+ protected boolean isLeaf;
// isSingle being true means all recursive children of this filter belong to one seriesPath.
- protected boolean isSingle = false;
+ boolean isSingle = false;
// if isSingle = false, singlePath must be null
- protected Path singlePath = null;
+ Path singlePath = null;
public FilterOperator(int tokenType) {
super(tokenType);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
index c86a3e0..91abbb4 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.db.qp.physical;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collections;
@@ -26,6 +27,8 @@ import org.apache.iotdb.db.qp.logical.Operator;
import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
@@ -38,6 +41,17 @@ public abstract class PhysicalPlan {
private Operator.OperatorType operatorType;
private static final int NULL_VALUE_LEN = -1;
+ //for cluster mode, whether the plan may be splitted into several sub plans
+ protected boolean canbeSplit = true;
+
+ /**
+ * whether the plan can be split into more than one Plans.
+ * Only used in the cluster mode.
+ */
+ public boolean canbeSplit() {
+ return canbeSplit;
+ }
+
protected PhysicalPlan(boolean isQuery) {
this.isQuery = isQuery;
}
@@ -73,6 +87,10 @@ public abstract class PhysicalPlan {
isQuery = query;
}
+ public void serializeTo(DataOutputStream stream) throws IOException {
+ throw new UnsupportedOperationException("serialize of unimplemented");
+ }
+
public void serializeTo(ByteBuffer buffer) {
throw new UnsupportedOperationException("serialize of unimplemented");
}
@@ -89,6 +107,14 @@ public abstract class PhysicalPlan {
}
}
+ protected void putString(DataOutputStream stream, String value) throws IOException {
+ if (value == null) {
+ stream.writeInt(NULL_VALUE_LEN);
+ } else {
+ ReadWriteIOUtils.write(value, stream);
+ }
+ }
+
protected String readString(ByteBuffer buffer) {
int valueLen = buffer.getInt();
if (valueLen == NULL_VALUE_LEN) {
@@ -110,6 +136,7 @@ public abstract class PhysicalPlan {
}
PhysicalPlanType type = PhysicalPlanType.values()[typeNum];
PhysicalPlan plan;
+ // TODO-Cluster: support more plans
switch (type) {
case INSERT:
plan = new InsertPlan();
@@ -123,6 +150,14 @@ public abstract class PhysicalPlan {
plan = new BatchInsertPlan();
plan.deserializeFrom(buffer);
break;
+ case SET_STORAGE_GROUP:
+ plan = new SetStorageGroupPlan();
+ plan.deserializeFrom(buffer);
+ break;
+ case CREATE_TIMESERIES:
+ plan = new CreateTimeSeriesPlan();
+ plan.deserializeFrom(buffer);
+ break;
default:
throw new IOException("unrecognized log type " + type);
}
@@ -131,7 +166,7 @@ public abstract class PhysicalPlan {
}
public enum PhysicalPlanType {
- INSERT, DELETE, BATCHINSERT
+ INSERT, DELETE, BATCHINSERT, SET_STORAGE_GROUP, CREATE_TIMESERIES
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
index fe1ee39..2510143 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
@@ -18,6 +18,8 @@
*/
package org.apache.iotdb.db.qp.physical.crud;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -37,7 +39,7 @@ public class BatchInsertPlan extends PhysicalPlan {
private String[] measurements;
private TSDataType[] dataTypes;
- private long[] times;
+ private long[] times; // times should be sorted. It is done in the session API.
private ByteBuffer timeBuffer;
private Object[] columns;
@@ -60,6 +62,11 @@ public class BatchInsertPlan extends PhysicalPlan {
this.deviceId = deviceId;
setMeasurements(measurements);
}
+ public BatchInsertPlan(String deviceId, String[] measurements) {
+ super(false, OperatorType.BATCHINSERT);
+ this.deviceId = deviceId;
+ setMeasurements(measurements);
+ }
public BatchInsertPlan(String deviceId, String[] measurements, List<Integer> dataTypes) {
super(false, OperatorType.BATCHINSERT);
@@ -106,6 +113,85 @@ public class BatchInsertPlan extends PhysicalPlan {
}
@Override
+ public void serializeTo(DataOutputStream stream) throws IOException {
+ int type = PhysicalPlanType.BATCHINSERT.ordinal();
+ stream.writeByte((byte) type);
+
+ putString(stream, deviceId);
+
+ stream.writeInt(measurements.length);
+ for (String m : measurements) {
+ putString(stream, m);
+ }
+
+ for (TSDataType dataType : dataTypes) {
+ stream.writeShort(dataType.serialize());
+ }
+
+ stream.writeInt(index.size());
+
+ if (timeBuffer == null) {
+ for(int loc : index){
+ stream.writeLong(times[loc]);
+ }
+ } else {
+ stream.write(timeBuffer.array());
+ timeBuffer = null;
+ }
+
+ if (valueBuffer == null) {
+ for (int i = 0; i < measurements.length; i++) {
+ TSDataType dataType = dataTypes[i];
+ switch (dataType) {
+ case INT32:
+ int[] intValues = (int[]) columns[i];
+ for(int loc : index){
+ stream.writeInt(intValues[loc]);
+ }
+ break;
+ case INT64:
+ long[] longValues = (long[]) columns[i];
+ for(int loc : index){
+ stream.writeLong(longValues[loc]);
+ }
+ break;
+ case FLOAT:
+ float[] floatValues = (float[]) columns[i];
+ for(int loc : index){
+ stream.writeFloat(floatValues[loc]);
+ }
+ break;
+ case DOUBLE:
+ double[] doubleValues = (double[]) columns[i];
+ for(int loc : index){
+ stream.writeDouble(doubleValues[loc]);
+ }
+ break;
+ case BOOLEAN:
+ boolean[] boolValues = (boolean[]) columns[i];
+ for(int loc : index){
+ stream.write(BytesUtils.boolToByte(boolValues[loc]));
+ }
+ break;
+ case TEXT:
+ Binary[] binaryValues = (Binary[]) columns[i];
+ for(int loc : index){
+ stream.writeInt(binaryValues[loc].getLength());
+ stream.write(binaryValues[loc].getValues());
+ }
+ break;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Data type %s is not supported.", dataType));
+ }
+ }
+ } else {
+ stream.write(valueBuffer.array());
+ valueBuffer = null;
+ }
+ }
+
+ @Override
public void serializeTo(ByteBuffer buffer) {
int type = PhysicalPlanType.BATCHINSERT.ordinal();
buffer.put((byte) type);
@@ -250,6 +336,10 @@ public class BatchInsertPlan extends PhysicalPlan {
}
}
+ public void setDataTypes(TSDataType[] dataTypes) {
+ this.dataTypes = dataTypes;
+ }
+
public Object[] getColumns() {
return columns;
}
@@ -303,4 +393,5 @@ public class BatchInsertPlan extends PhysicalPlan {
public void setRowCount(int size) {
this.rowCount = size;
}
+
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java
index a46aaea..d2f1236 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java
@@ -18,6 +18,8 @@
*/
package org.apache.iotdb.db.qp.physical.crud;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -102,6 +104,14 @@ public class DeletePlan extends PhysicalPlan {
}
@Override
+ public void serializeTo(DataOutputStream stream) throws IOException {
+ int type = PhysicalPlanType.DELETE.ordinal();
+ stream.writeByte((byte) type);
+ stream.writeLong(deleteTime);
+ putString(stream, paths.get(0).getFullPath());
+ }
+
+ @Override
public void serializeTo(ByteBuffer buffer) {
int type = PhysicalPlanType.DELETE.ordinal();
buffer.put((byte) type);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
index 048370b..d2e272b 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
@@ -18,6 +18,8 @@
*/
package org.apache.iotdb.db.qp.physical.crud;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
@@ -41,6 +43,7 @@ public class InsertPlan extends PhysicalPlan {
public InsertPlan() {
super(false, OperatorType.INSERT);
+ canbeSplit = false;
}
@TestOnly
@@ -50,6 +53,7 @@ public class InsertPlan extends PhysicalPlan {
this.deviceId = deviceId;
this.measurements = new String[] {measurement};
this.values = new String[] {insertValue};
+ canbeSplit = false;
}
public InsertPlan(TSRecord tsRecord) {
@@ -64,6 +68,7 @@ public class InsertPlan extends PhysicalPlan {
dataTypes[i] = tsRecord.dataPointList.get(i).getType();
values[i] = tsRecord.dataPointList.get(i).getValue().toString();
}
+ canbeSplit = false;
}
public InsertPlan(String deviceId, long insertTime, String[] measurementList,
@@ -73,6 +78,7 @@ public class InsertPlan extends PhysicalPlan {
this.deviceId = deviceId;
this.measurements = measurementList;
this.values = insertValues;
+ canbeSplit = false;
}
public long getTime() {
@@ -145,6 +151,25 @@ public class InsertPlan extends PhysicalPlan {
}
@Override
+ public void serializeTo(DataOutputStream stream) throws IOException {
+ int type = PhysicalPlanType.INSERT.ordinal();
+ stream.writeByte((byte) type);
+ stream.writeLong(time);
+
+ putString(stream, deviceId);
+
+ stream.writeInt(measurements.length);
+ for (String m : measurements) {
+ putString(stream, m);
+ }
+
+ stream.writeInt(values.length);
+ for (String m : values) {
+ putString(stream, m);
+ }
+ }
+
+ @Override
public void serializeTo(ByteBuffer buffer) {
int type = PhysicalPlanType.INSERT.ordinal();
buffer.put((byte) type);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CountPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CountPlan.java
index 044d838..b1851ad 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CountPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CountPlan.java
@@ -22,6 +22,9 @@ import org.apache.iotdb.tsfile.read.common.Path;
/**
* CountPlan is used to count time-series and count nodes.
+ * COUNT_TIMESERIES if using "COUNT TIMESERIES <Path>" and only this command supports wildcard.
+ * COUNT_NODE_TIMESERIES if using "COUNT TIMESERIES <Path> GROUP BY LEVEL=<INTEGER>"
+ * COUNT_NODE if using "COUNT NODES <Path> LEVEL=<INTEGER>"
*/
public class CountPlan extends ShowPlan {
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateTimeSeriesPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateTimeSeriesPlan.java
index a6ce3cd..90cb602 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateTimeSeriesPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateTimeSeriesPlan.java
@@ -18,10 +18,13 @@
*/
package org.apache.iotdb.db.qp.physical.sys;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.List;
import java.util.Map;
-
+import java.util.Objects;
import org.apache.iotdb.db.qp.logical.Operator;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
@@ -36,8 +39,13 @@ public class CreateTimeSeriesPlan extends PhysicalPlan {
private TSEncoding encoding;
private CompressionType compressor;
private Map<String, String> props;
-
- public CreateTimeSeriesPlan(Path path, TSDataType dataType, TSEncoding encoding,
+
+ public CreateTimeSeriesPlan() {
+ super(false, Operator.OperatorType.CREATE_TIMESERIES);
+ canbeSplit = false;
+ }
+
+ public CreateTimeSeriesPlan(Path path, TSDataType dataType, TSEncoding encoding,
CompressionType compressor, Map<String, String> props) {
super(false, Operator.OperatorType.CREATE_TIMESERIES);
this.path = path;
@@ -45,6 +53,7 @@ public class CreateTimeSeriesPlan extends PhysicalPlan {
this.encoding = encoding;
this.compressor = compressor;
this.props = props;
+ canbeSplit = false;
}
public Path getPath() {
@@ -89,14 +98,8 @@ public class CreateTimeSeriesPlan extends PhysicalPlan {
@Override
public String toString() {
- String ret = String.format("seriesPath: %s%nresultDataType: %s%nencoding: %s%nnamespace type:"
- + " ADD_PATH%nargs: ", path, dataType, encoding);
- StringBuilder stringBuilder = new StringBuilder(ret.length()+50);
- stringBuilder.append(ret);
- for (Map.Entry<String, String> prop : props.entrySet()) {
- stringBuilder.append(prop.getKey()).append("=").append(prop.getValue()).append(",");
- }
- return stringBuilder.toString();
+ return String.format("seriesPath: %s, resultDataType: %s, encoding: %s, compression: %s", path,
+ dataType, encoding, compressor);
}
@Override
@@ -104,4 +107,45 @@ public class CreateTimeSeriesPlan extends PhysicalPlan {
return Collections.singletonList(path);
}
+ @Override
+ public void serializeTo(DataOutputStream stream) throws IOException {
+ stream.write(PhysicalPlanType.CREATE_TIMESERIES.ordinal());
+ byte[] pathBytes = path.getFullPath().getBytes();
+ stream.writeInt(pathBytes.length);
+ stream.write(pathBytes);
+ stream.write(dataType.ordinal());
+ stream.write(encoding.ordinal());
+ stream.write(compressor.ordinal());
+ }
+
+ @Override
+ public void deserializeFrom(ByteBuffer buffer) {
+ int length = buffer.getInt();
+ byte[] pathBytes = new byte[length];
+ buffer.get(pathBytes);
+ path = new Path(new String(pathBytes));
+ dataType = TSDataType.values()[buffer.get()];
+ encoding = TSEncoding.values()[buffer.get()];
+ compressor = CompressionType.values()[buffer.get()];
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ CreateTimeSeriesPlan that = (CreateTimeSeriesPlan) o;
+ return Objects.equals(path, that.path) &&
+ dataType == that.dataType &&
+ encoding == that.encoding &&
+ compressor == that.compressor;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(path, dataType, encoding, compressor);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/SetStorageGroupPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/SetStorageGroupPlan.java
index 72e9ebc..28d19cd 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/SetStorageGroupPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/SetStorageGroupPlan.java
@@ -18,16 +18,23 @@
*/
package org.apache.iotdb.db.qp.physical.sys;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
-
+import java.util.Objects;
import org.apache.iotdb.db.qp.logical.Operator;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.tsfile.read.common.Path;
public class SetStorageGroupPlan extends PhysicalPlan {
private Path path;
-
+
+ public SetStorageGroupPlan() {
+ super(false, Operator.OperatorType.SET_STORAGE_GROUP);
+ }
+
public SetStorageGroupPlan(Path path) {
super(false, Operator.OperatorType.SET_STORAGE_GROUP);
this.path = path;
@@ -50,4 +57,41 @@ public class SetStorageGroupPlan extends PhysicalPlan {
return ret;
}
+ @Override
+ public void serializeTo(DataOutputStream stream) throws IOException {
+ stream.write((byte) PhysicalPlanType.SET_STORAGE_GROUP.ordinal());
+ byte[] fullPathBytes = path.getFullPath().getBytes();
+ stream.writeInt(fullPathBytes.length);
+ stream.write(fullPathBytes);
+ }
+
+ @Override
+ public void deserializeFrom(ByteBuffer buffer) {
+ int length = buffer.getInt();
+ byte[] fullPathBytes = new byte[length];
+ buffer.get(fullPathBytes);
+ path = new Path(new String(fullPathBytes));
+ }
+
+ @Override
+ public String toString() {
+ return "SetStorageGroup{" + path + '}';
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SetStorageGroupPlan that = (SetStorageGroupPlan) o;
+ return Objects.equals(path, that.path);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(path);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowChildPathsPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowChildPathsPlan.java
index 9f6c199..a1171ae 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowChildPathsPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowChildPathsPlan.java
@@ -22,11 +22,13 @@ package org.apache.iotdb.db.qp.physical.sys;
import org.apache.iotdb.tsfile.read.common.Path;
public class ShowChildPathsPlan extends ShowPlan {
+ //show child paths SQL does not support path with wildcard
private Path path;
public ShowChildPathsPlan(ShowContentType showContentType, Path path) {
super(showContentType);
this.path = path;
+ canbeSplit = false;
}
public Path getPath() {
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 49a21c2..64805fa 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
@@ -82,7 +82,7 @@ import org.apache.iotdb.db.qp.physical.sys.ShowPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowPlan.ShowContentType;
import org.apache.iotdb.db.qp.physical.sys.ShowTTLPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowTimeSeriesPlan;
-import org.apache.iotdb.db.service.TSServiceImpl;
+import org.apache.iotdb.db.utils.SchemaUtils;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.read.expression.IExpression;
@@ -210,6 +210,9 @@ public class PhysicalGenerator {
}
}
+ protected TSDataType getSeriesType(String path) throws MetadataException {
+ return SchemaUtils.getSeriesType(path);
+ }
private PhysicalPlan transformQuery(QueryOperator queryOperator)
throws QueryProcessException {
@@ -282,8 +285,8 @@ public class PhysicalGenerator {
Path fullPath = Path.addPrefixPath(suffixPath, device);
try {
// remove stars in SELECT to get actual paths
- List<String> actualPaths = MManager.getInstance()
- .getAllTimeseriesName(fullPath.getFullPath());
+ List<String> actualPaths = getMatchedTimeseries(fullPath.getFullPath()); // remove stars in SELECT to
+ // get actual paths
// for actual non exist path
if (actualPaths.isEmpty() && originAggregations.isEmpty()) {
@@ -305,7 +308,7 @@ public class PhysicalGenerator {
pathForDataType = path.getFullPath();
measurementChecked = path.getMeasurement();
}
- TSDataType dataType = TSServiceImpl.getSeriesType(pathForDataType);
+ TSDataType dataType = getSeriesType(pathForDataType);
if (dataTypeConsistencyChecker.containsKey(measurementChecked)) {
if (!dataType.equals(dataTypeConsistencyChecker.get(measurementChecked))) {
throw new QueryProcessException(
@@ -421,7 +424,7 @@ public class PhysicalGenerator {
Set<String> deviceSet = new LinkedHashSet<>();
try {
for (Path path : paths) {
- Set<String> tempDS = MManager.getInstance().getDevices(path.getFullPath());
+ Set<String> tempDS = getMatchedDevices(path.getFullPath());
deviceSet.addAll(tempDS);
}
retDevices = new ArrayList<>(deviceSet);
@@ -454,7 +457,7 @@ public class PhysicalGenerator {
List<Path> paths = queryPlan.getPaths();
List<TSDataType> dataTypes = new ArrayList<>(paths.size());
for (Path path : paths) {
- TSDataType seriesType = MManager.getInstance().getSeriesType(path.toString());
+ TSDataType seriesType = getSeriesType(path.toString());
dataTypes.add(seriesType);
queryPlan.addTypeMapping(path, seriesType);
}
@@ -522,5 +525,13 @@ public class PhysicalGenerator {
}
}
}
+
+ protected List<String> getMatchedTimeseries(String path) throws MetadataException {
+ return MManager.getInstance().getAllTimeseriesName(path);
+ }
+
+ protected Set<String> getMatchedDevices(String path) throws MetadataException {
+ return MManager.getInstance().getDevices(path);
+ }
}
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 bd36fa4..91c2e0b 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,19 +18,26 @@
*/
package org.apache.iotdb.db.qp.strategy.optimizer;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.LogicalOptimizeException;
import org.apache.iotdb.db.exception.runtime.SQLParserException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.qp.constant.SQLConstant;
import org.apache.iotdb.db.qp.logical.Operator;
-import org.apache.iotdb.db.qp.logical.crud.*;
+import org.apache.iotdb.db.qp.logical.crud.BasicFunctionOperator;
+import org.apache.iotdb.db.qp.logical.crud.FilterOperator;
+import org.apache.iotdb.db.qp.logical.crud.FromOperator;
+import org.apache.iotdb.db.qp.logical.crud.FunctionOperator;
+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.tsfile.read.common.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.*;
-
/**
* concat paths in select and from clause.
*/
@@ -174,7 +181,7 @@ public class ConcatPathOptimizer implements ILogicalOptimizer {
* @param seriesLimit is ensured to be positive integer
* @param seriesOffset is ensured to be non-negative integer
*/
- public void slimitTrim(SelectOperator select, int seriesLimit, int seriesOffset)
+ private void slimitTrim(SelectOperator select, int seriesLimit, int seriesOffset)
throws LogicalOptimizeException {
List<Path> suffixList = select.getSuffixPaths();
List<String> aggregations = select.getAggregations();
@@ -268,7 +275,7 @@ public class ConcatPathOptimizer implements ILogicalOptimizer {
LinkedHashMap<String, Integer> pathMap = new LinkedHashMap<>();
try {
for (Path path : paths) {
- List<String> all = MManager.getInstance().getAllTimeseriesName(path.getFullPath());
+ List<String> all = removeWildcard(path.getFullPath());
for (String subPath : all) {
if (!pathMap.containsKey(subPath)) {
pathMap.put(subPath, 1);
@@ -290,7 +297,7 @@ public class ConcatPathOptimizer implements ILogicalOptimizer {
List<String> newAggregations = new ArrayList<>();
for (int i = 0; i < paths.size(); i++) {
try {
- List<String> actualPaths = MManager.getInstance().getAllTimeseriesName(paths.get(i).getFullPath());
+ List<String> actualPaths = removeWildcard(paths.get(i).getFullPath());
for (String actualPath : actualPaths) {
retPaths.add(new Path(actualPath));
if (afterConcatAggregations != null && !afterConcatAggregations.isEmpty()) {
@@ -304,4 +311,8 @@ public class ConcatPathOptimizer implements ILogicalOptimizer {
selectOperator.setSuffixPathList(retPaths);
selectOperator.setAggregations(newAggregations);
}
+
+ protected List<String> removeWildcard(String path) throws MetadataException {
+ return MManager.getInstance().getAllTimeseriesName(path);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateResult.java b/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateResult.java
index 4b88bca..70019f1 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateResult.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateResult.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.query.aggregation;
import java.io.OutputStream;
import java.nio.ByteBuffer;
+import java.util.Objects;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.query.factory.AggregateResultFactory;
import org.apache.iotdb.db.query.reader.series.IReaderByTimestamp;
@@ -284,4 +285,8 @@ public abstract class AggregateResult {
return hasResult;
}
+ @Override
+ public String toString() {
+ return String.valueOf(getResult());
+ }
}
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 03cbecc..a6d4653 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
@@ -105,7 +105,7 @@ public class QueryResourceManager {
try {
deserializer.close();
} catch (IOException e) {
- throw new StorageEngineException(e.getMessage());
+ throw new StorageEngineException(e);
}
}
externalSortFileMap.remove(queryId);
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java
index 20abcd9..40a1226 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java
@@ -201,7 +201,7 @@ public class AlignByDeviceDataSet extends QueryDataSet {
default:
throw new IOException("unsupported DataSetType");
}
- } catch (QueryProcessException | QueryFilterOptimizationException | StorageEngineException | IOException e) {
+ } catch (QueryProcessException | QueryFilterOptimizationException | StorageEngineException e) {
throw new IOException(e);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithValueFilter.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithValueFilter.java
index 9e94262..208e3f2 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithValueFilter.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithValueFilter.java
@@ -21,16 +21,16 @@ package org.apache.iotdb.db.query.dataset;
import java.io.IOException;
import java.util.List;
import org.apache.iotdb.db.query.reader.series.IReaderByTimestamp;
-import org.apache.iotdb.db.query.timegenerator.ServerTimeGenerator;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.read.query.timegenerator.TimeGenerator;
public class RawQueryDataSetWithValueFilter extends QueryDataSet {
- private ServerTimeGenerator timeGenerator;
+ private TimeGenerator timeGenerator;
private List<IReaderByTimestamp> seriesReaderByTimestampList;
private boolean hasCachedRowRecord;
private RowRecord cachedRowRecord;
@@ -44,7 +44,7 @@ public class RawQueryDataSetWithValueFilter extends QueryDataSet {
* @param readers readers in List(IReaderByTimeStamp) structure
*/
public RawQueryDataSetWithValueFilter(List<Path> paths, List<TSDataType> dataTypes,
- ServerTimeGenerator timeGenerator, List<IReaderByTimestamp> readers) {
+ TimeGenerator timeGenerator, List<IReaderByTimestamp> readers) {
super(paths, dataTypes);
this.timeGenerator = timeGenerator;
this.seriesReaderByTimestampList = readers;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
index 398ef9e..8ceff22 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.query.dataset.groupby;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.PathException;
import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
import org.apache.iotdb.db.query.aggregation.AggregateResult;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
index bf9c5e6..0062e22 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.query.dataset.groupby;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
import org.apache.iotdb.db.query.aggregation.AggregateResult;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
index c39a392..d21b2d5 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
@@ -36,11 +36,13 @@ import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.QueryResourceManager;
import org.apache.iotdb.db.query.dataset.SingleDataSet;
import org.apache.iotdb.db.query.factory.AggregateResultFactory;
+import org.apache.iotdb.db.query.filter.TsFileFilter;
import org.apache.iotdb.db.query.reader.series.IReaderByTimestamp;
import org.apache.iotdb.db.query.reader.series.IAggregateReader;
import org.apache.iotdb.db.query.reader.series.SeriesAggregateReader;
import org.apache.iotdb.db.query.reader.series.SeriesReaderByTimestamp;
import org.apache.iotdb.db.query.timegenerator.ServerTimeGenerator;
+import org.apache.iotdb.db.utils.QueryUtils;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
import org.apache.iotdb.tsfile.read.common.BatchData;
@@ -50,13 +52,14 @@ import org.apache.iotdb.tsfile.read.expression.IExpression;
import org.apache.iotdb.tsfile.read.expression.impl.GlobalTimeExpression;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.read.query.timegenerator.TimeGenerator;
public class AggregationExecutor {
private List<Path> selectedSeries;
- private List<TSDataType> dataTypes;
- private List<String> aggregations;
- private IExpression expression;
+ protected List<TSDataType> dataTypes;
+ protected List<String> aggregations;
+ protected IExpression expression;
/**
* aggregation batch calculation size.
@@ -107,32 +110,45 @@ public class AggregationExecutor {
* @param context query context
* @return AggregateResult list
*/
- private List<AggregateResult> aggregateOneSeries(
+ protected List<AggregateResult> aggregateOneSeries(
Map.Entry<Path, List<Integer>> pathToAggrIndexes,
Filter timeFilter, QueryContext context)
throws IOException, QueryProcessException, StorageEngineException {
List<AggregateResult> aggregateResultList = new ArrayList<>();
- List<Boolean> isCalculatedList = new ArrayList<>();
+
Path seriesPath = pathToAggrIndexes.getKey();
TSDataType tsDataType = dataTypes.get(pathToAggrIndexes.getValue().get(0));
+ for (int i : pathToAggrIndexes.getValue()) {
+ // construct AggregateResult
+ AggregateResult aggregateResult = AggregateResultFactory
+ .getAggrResultByName(aggregations.get(i), tsDataType);
+ aggregateResultList.add(aggregateResult);
+ }
+ aggregateOneSeries(seriesPath, context, timeFilter, tsDataType, aggregateResultList, null);
+ return aggregateResultList;
+ }
+
+ public static void aggregateOneSeries(Path seriesPath, QueryContext context, Filter timeFilter,
+ TSDataType tsDataType, List<AggregateResult> aggregateResultList, TsFileFilter fileFilter)
+ throws StorageEngineException, IOException, QueryProcessException {
+ List<Boolean> isCalculatedList = new ArrayList<>();
+ for (int i = 0; i < aggregateResultList.size(); i++) {
+ isCalculatedList.add(false);
+ }
// construct series reader without value filter
QueryDataSource queryDataSource = QueryResourceManager.getInstance()
.getQueryDataSource(seriesPath, context, timeFilter);
+ if (fileFilter != null) {
+ QueryUtils.filterQueryDataSource(queryDataSource, fileFilter);
+ }
// update filter by TTL
timeFilter = queryDataSource.updateFilterUsingTTL(timeFilter);
- IAggregateReader seriesReader = new SeriesAggregateReader(pathToAggrIndexes.getKey(),
+ IAggregateReader seriesReader = new SeriesAggregateReader(seriesPath,
tsDataType, context, queryDataSource, timeFilter, null, null);
- for (int i : pathToAggrIndexes.getValue()) {
- // construct AggregateResult
- AggregateResult aggregateResult = AggregateResultFactory
- .getAggrResultByName(aggregations.get(i), tsDataType);
- aggregateResultList.add(aggregateResult);
- isCalculatedList.add(false);
- }
- int remainingToCalculate = pathToAggrIndexes.getValue().size();
+ int remainingToCalculate = aggregateResultList.size();
while (seriesReader.hasNextChunk()) {
// cal by chunk statistics
@@ -146,7 +162,7 @@ public class AggregationExecutor {
isCalculatedList.set(i, true);
remainingToCalculate--;
if (remainingToCalculate == 0) {
- return aggregateResultList;
+ return;
}
}
}
@@ -166,7 +182,7 @@ public class AggregationExecutor {
isCalculatedList.set(i, true);
remainingToCalculate--;
if (remainingToCalculate == 0) {
- return aggregateResultList;
+ return;
}
}
}
@@ -186,7 +202,7 @@ public class AggregationExecutor {
isCalculatedList.set(i, true);
remainingToCalculate--;
if (remainingToCalculate == 0) {
- return aggregateResultList;
+ return;
}
}
}
@@ -194,7 +210,6 @@ public class AggregationExecutor {
}
}
}
- return aggregateResultList;
}
/**
@@ -203,15 +218,14 @@ public class AggregationExecutor {
* @param context query context.
*/
public QueryDataSet executeWithValueFilter(QueryContext context)
- throws StorageEngineException, PathException, IOException {
+ throws StorageEngineException, IOException {
- ServerTimeGenerator timestampGenerator = new ServerTimeGenerator(expression, context);
+ TimeGenerator timestampGenerator = getTimeGenerator(context);
List<IReaderByTimestamp> readersOfSelectedSeries = new ArrayList<>();
for (int i = 0; i < selectedSeries.size(); i++) {
Path path = selectedSeries.get(i);
- SeriesReaderByTimestamp seriesReaderByTimestamp = new SeriesReaderByTimestamp(path,
- dataTypes.get(i), context,
- QueryResourceManager.getInstance().getQueryDataSource(path, context, null), null);
+ IReaderByTimestamp seriesReaderByTimestamp = getReaderByTime(path,
+ dataTypes.get(i), context);
readersOfSelectedSeries.add(seriesReaderByTimestamp);
}
@@ -225,11 +239,22 @@ public class AggregationExecutor {
return constructDataSet(aggregateResults);
}
+ protected TimeGenerator getTimeGenerator(QueryContext context) throws StorageEngineException {
+ return new ServerTimeGenerator(expression, context);
+ }
+
+ protected IReaderByTimestamp getReaderByTime(Path path, TSDataType dataType,
+ QueryContext context) throws StorageEngineException {
+ return new SeriesReaderByTimestamp(path,
+ dataType, context,
+ QueryResourceManager.getInstance().getQueryDataSource(path, context, null), null);
+ }
+
/**
* calculate aggregation result with value filter.
*/
private void aggregateWithValueFilter(List<AggregateResult> aggregateResults,
- ServerTimeGenerator timestampGenerator, List<IReaderByTimestamp> readersOfSelectedSeries)
+ TimeGenerator timestampGenerator, List<IReaderByTimestamp> readersOfSelectedSeries)
throws IOException {
while (timestampGenerator.hasNext()) {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java b/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java
index 3f27bb4..f8e1297 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/QueryRouter.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
import org.apache.iotdb.db.qp.physical.crud.FillQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
+import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.dataset.groupby.GroupByWithValueFilterDataSet;
@@ -64,7 +65,7 @@ public class QueryRouter implements IQueryRouter {
}
queryPlan.setExpression(optimizedExpression);
- RawDataQueryExecutor rawDataQueryExecutor = new RawDataQueryExecutor(queryPlan);
+ RawDataQueryExecutor rawDataQueryExecutor = getRawDataQueryExecutor(queryPlan);
if (!queryPlan.isAlignByTime()) {
return rawDataQueryExecutor.executeNonAlign(context);
@@ -78,6 +79,10 @@ public class QueryRouter implements IQueryRouter {
return rawDataQueryExecutor.executeWithoutValueFilter(context);
}
+ protected RawDataQueryExecutor getRawDataQueryExecutor(RawDataQueryPlan queryPlan) {
+ return new RawDataQueryExecutor(queryPlan);
+ }
+
@Override
public QueryDataSet aggregate(AggregationPlan aggregationPlan, QueryContext context)
throws QueryFilterOptimizationException, StorageEngineException, QueryProcessException,
@@ -93,7 +98,7 @@ public class QueryRouter implements IQueryRouter {
aggregationPlan.setExpression(optimizedExpression);
- AggregationExecutor engineExecutor = new AggregationExecutor(aggregationPlan);
+ AggregationExecutor engineExecutor = getAggregationExecutor(aggregationPlan);
if (optimizedExpression != null
&& optimizedExpression.getType() != ExpressionType.GLOBAL_TIME) {
@@ -103,6 +108,9 @@ public class QueryRouter implements IQueryRouter {
return engineExecutor.executeWithoutValueFilter(context);
}
+ protected AggregationExecutor getAggregationExecutor(AggregationPlan aggregationPlan) {
+ return new AggregationExecutor(aggregationPlan);
+ }
@Override
public QueryDataSet groupBy(GroupByPlan groupByPlan, QueryContext context)
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
index 8785f54..4f9ea11 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
@@ -25,8 +25,8 @@ import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.QueryResourceManager;
-import org.apache.iotdb.db.query.dataset.RawQueryDataSetWithValueFilter;
import org.apache.iotdb.db.query.dataset.NonAlignEngineDataSet;
+import org.apache.iotdb.db.query.dataset.RawQueryDataSetWithValueFilter;
import org.apache.iotdb.db.query.dataset.RawQueryDataSetWithoutValueFilter;
import org.apache.iotdb.db.query.reader.series.IReaderByTimestamp;
import org.apache.iotdb.db.query.reader.series.ManagedSeriesReader;
@@ -39,15 +39,16 @@ import org.apache.iotdb.tsfile.read.expression.IExpression;
import org.apache.iotdb.tsfile.read.expression.impl.GlobalTimeExpression;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.read.query.timegenerator.TimeGenerator;
/**
* IoTDB query executor.
*/
public class RawDataQueryExecutor {
- private List<Path> deduplicatedPaths;
- private List<TSDataType> deduplicatedDataTypes;
- private IExpression optimizedExpression;
+ protected List<Path> deduplicatedPaths;
+ protected List<TSDataType> deduplicatedDataTypes;
+ protected IExpression optimizedExpression;
public RawDataQueryExecutor(RawDataQueryPlan queryPlan) {
this.deduplicatedPaths = queryPlan.getDeduplicatedPaths();
@@ -77,7 +78,7 @@ public class RawDataQueryExecutor {
readersOfSelectedSeries);
}
- private List<ManagedSeriesReader> initManagedSeriesReader(QueryContext context)
+ protected List<ManagedSeriesReader> initManagedSeriesReader(QueryContext context)
throws StorageEngineException {
Filter timeFilter = null;
if (optimizedExpression != null) {
@@ -108,19 +109,30 @@ public class RawDataQueryExecutor {
*/
public QueryDataSet executeWithValueFilter(QueryContext context) throws StorageEngineException {
- ServerTimeGenerator timestampGenerator = new ServerTimeGenerator(
+ TimeGenerator timestampGenerator = getTimeGenerator(
optimizedExpression, context);
List<IReaderByTimestamp> readersOfSelectedSeries = new ArrayList<>();
for (int i = 0; i < deduplicatedPaths.size(); i++) {
Path path = deduplicatedPaths.get(i);
- SeriesReaderByTimestamp seriesReaderByTimestamp = new SeriesReaderByTimestamp(path,
- deduplicatedDataTypes.get(i), context,
- QueryResourceManager.getInstance().getQueryDataSource(path, context, null), null);
+ IReaderByTimestamp seriesReaderByTimestamp = getReaderByTimestamp(path,
+ deduplicatedDataTypes.get(i), context);
readersOfSelectedSeries.add(seriesReaderByTimestamp);
}
return new RawQueryDataSetWithValueFilter(deduplicatedPaths, deduplicatedDataTypes,
timestampGenerator, readersOfSelectedSeries);
}
+ protected IReaderByTimestamp getReaderByTimestamp(Path path, TSDataType dataType,
+ QueryContext context)
+ throws StorageEngineException {
+ return new SeriesReaderByTimestamp(path,
+ dataType, context,
+ QueryResourceManager.getInstance().getQueryDataSource(path, context, null), null);
+ }
+
+ protected TimeGenerator getTimeGenerator(IExpression expression,
+ QueryContext context) throws StorageEngineException {
+ return new ServerTimeGenerator(expression, context);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/externalsort/ExternalSortJob.java b/server/src/main/java/org/apache/iotdb/db/query/externalsort/ExternalSortJob.java
index 4d91289..bf05fba 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/externalsort/ExternalSortJob.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/externalsort/ExternalSortJob.java
@@ -26,7 +26,7 @@ import org.apache.iotdb.tsfile.read.reader.IPointReader;
/**
* This class represents an external sort job. Every job will use a separated directory.
*/
-public class ExternalSortJob {
+class ExternalSortJob {
private List<ExternalSortJobPart> partList;
@@ -41,5 +41,4 @@ public class ExternalSortJob {
}
return readers;
}
-
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/externalsort/ExternalSortJobScheduler.java b/server/src/main/java/org/apache/iotdb/db/query/externalsort/ExternalSortJobScheduler.java
index 1db8cd3..c1e7071 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/externalsort/ExternalSortJobScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/externalsort/ExternalSortJobScheduler.java
@@ -16,29 +16,28 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.query.externalsort;
+ package org.apache.iotdb.db.query.externalsort;
-import java.util.concurrent.atomic.AtomicLong;
+ public class ExternalSortJobScheduler {
-public class ExternalSortJobScheduler {
-
- private AtomicLong jobId = new AtomicLong(0);
+ private long queryId = 0;
private ExternalSortJobScheduler() {
}
- long genJobId() {
- return jobId.getAndIncrement();
+ synchronized long genJobId() {
+ queryId++;
+ return queryId;
}
private static class ExternalSortJobSchedulerHelper {
- private static final ExternalSortJobScheduler INSTANCE = new ExternalSortJobScheduler();
+ private static final ExternalSortJobScheduler INSTANCE = new ExternalSortJobScheduler();
}
public static ExternalSortJobScheduler getInstance() {
- return ExternalSortJobSchedulerHelper.INSTANCE;
+ return ExternalSortJobSchedulerHelper.INSTANCE;
}
-}
+ }
diff --git a/server/src/main/java/org/apache/iotdb/db/query/externalsort/SimpleExternalSortEngine.java b/server/src/main/java/org/apache/iotdb/db/query/externalsort/SimpleExternalSortEngine.java
index 5fba08b..00097a8 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/externalsort/SimpleExternalSortEngine.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/externalsort/SimpleExternalSortEngine.java
@@ -103,10 +103,10 @@ public class SimpleExternalSortEngine implements ExternalSortJobEngine {
int toIndex = Math.min(i + minExternalSortSourceCount, ret.size());
List<ExternalSortJobPart> partGroup = ret.subList(i, toIndex);
i = toIndex;
- StringBuilder tmpFilePath = new StringBuilder(queryDir).append(jobId).append("_")
- .append(partId);
+ String tmpFilePath = queryDir + jobId + "_"
+ + partId;
MultiSourceExternalSortJobPart part = new MultiSourceExternalSortJobPart(queryId,
- tmpFilePath.toString(), partGroup);
+ tmpFilePath, partGroup);
tmpPartList.add(part);
partId++;
}
@@ -115,19 +115,19 @@ public class SimpleExternalSortEngine implements ExternalSortJobEngine {
return new ExternalSortJob(ret);
}
- public String getQueryDir() {
+ String getQueryDir() {
return queryDir;
}
- public void setQueryDir(String queryDir) {
+ void setQueryDir(String queryDir) {
this.queryDir = queryDir;
}
- public int getMinExternalSortSourceCount() {
+ int getMinExternalSortSourceCount() {
return minExternalSortSourceCount;
}
- public void setMinExternalSortSourceCount(int minExternalSortSourceCount) {
+ void setMinExternalSortSourceCount(int minExternalSortSourceCount) {
this.minExternalSortSourceCount = minExternalSortSourceCount;
}
@@ -171,10 +171,10 @@ public class SimpleExternalSortEngine implements ExternalSortJobEngine {
private static class SimpleExternalSortJobEngineHelper {
- private static SimpleExternalSortEngine INSTANCE = new SimpleExternalSortEngine();
+ private static final SimpleExternalSortEngine INSTANCE = new SimpleExternalSortEngine();
}
public static SimpleExternalSortEngine getInstance() {
return SimpleExternalSortJobEngineHelper.INSTANCE;
}
-}
\ No newline at end of file
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesReaderByTimestamp.java b/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesReaderByTimestamp.java
index dd1273f..071974b 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesReaderByTimestamp.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesReaderByTimestamp.java
@@ -42,6 +42,10 @@ public class SeriesReaderByTimestamp implements IReaderByTimestamp {
dataSource, TimeFilter.gtEq(Long.MIN_VALUE), null, fileFilter);
}
+ public SeriesReaderByTimestamp(SeriesReader seriesReader) {
+ this.seriesReader = seriesReader;
+ }
+
@Override
public Object[] getValuesInTimestamps(long[] timestamps) throws IOException {
Object[] result = new Object[timestamps.length];
diff --git a/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java b/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java
index 6901250..ee5baff 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java
@@ -130,7 +130,7 @@ public class PriorityMergeReader implements IPointReader {
}
}
- class Element {
+ static class Element {
IPointReader reader;
TimeValuePair timeValuePair;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java b/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java
index 28fcd4e..5695e3f 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java
@@ -39,7 +39,11 @@ import org.apache.iotdb.tsfile.read.reader.IBatchReader;
*/
public class ServerTimeGenerator extends TimeGenerator {
- private QueryContext context;
+ protected QueryContext context;
+
+ public ServerTimeGenerator(QueryContext context) {
+ this.context = context;
+ }
/**
* Constructor of EngineTimeGenerator.
@@ -50,7 +54,7 @@ public class ServerTimeGenerator extends TimeGenerator {
try {
super.constructNode(expression);
} catch (IOException e) {
- throw new StorageEngineException(e.getMessage());
+ throw new StorageEngineException(e);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
index 0d2c4c5..072e2cb 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
@@ -96,7 +96,9 @@ public class IoTDB implements IoTDBMBean {
registerManager.register(UpgradeSevice.getINSTANCE());
registerManager.register(MergeManager.getINSTANCE());
registerManager.register(CacheHitRatioMonitor.getInstance());
- registerManager.register(MetricsService.getInstance());
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableMetricService()) {
+ registerManager.register(MetricsService.getInstance());
+ }
registerManager.register(StorageEngine.getInstance());
JMXService.registerMBean(getInstance(), mbeanName);
diff --git a/server/src/main/java/org/apache/iotdb/db/service/ServiceType.java b/server/src/main/java/org/apache/iotdb/db/service/ServiceType.java
index 670ea32..0b0dea5 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/ServiceType.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/ServiceType.java
@@ -43,7 +43,8 @@ public enum ServiceType {
generateJmxName(IoTDBConstant.IOTDB_PACKAGE, "Cache Hit Ratio")),
FLUSH_SERVICE("Flush ServerService",
- generateJmxName("org.apache.iotdb.db.engine.pool", "Flush Manager"));
+ generateJmxName("org.apache.iotdb.db.engine.pool", "Flush Manager")),
+ CLUSTER_MONITOR_SERVICE("Cluster Monitor ServerService", "Cluster Monitor");
private String name;
private String jmxName;
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 2a7a614..8c5f056 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
@@ -83,6 +83,7 @@ import org.apache.iotdb.db.query.dataset.RawQueryDataSetWithoutValueFilter;
import org.apache.iotdb.db.tools.watermark.GroupedLSBWatermarkEncoder;
import org.apache.iotdb.db.tools.watermark.WatermarkEncoder;
import org.apache.iotdb.db.utils.QueryDataSetUtils;
+import org.apache.iotdb.db.utils.SchemaUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.service.rpc.thrift.ServerProperties;
import org.apache.iotdb.service.rpc.thrift.TSBatchInsertionReq;
@@ -138,8 +139,8 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
"meet error while parsing SQL to physical plan: {}";
public static Vector<SqlArgument> sqlArgumentsList = new Vector<>();
- private Planner processor;
- private IPlanExecutor executor;
+ protected Planner processor;
+ protected IPlanExecutor executor;
// Record the username for every rpc connection (session).
private Map<Long, String> sessionIdUsernameMap = new ConcurrentHashMap<>();
@@ -168,52 +169,6 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
executor = new PlanExecutor();
}
- public static TSDataType getSeriesType(String path) throws QueryProcessException {
- switch (path.toLowerCase()) {
- // authorization queries
- case COLUMN_ROLE:
- case COLUMN_USER:
- case COLUMN_PRIVILEGE:
- case COLUMN_STORAGE_GROUP:
- return TSDataType.TEXT;
- case COLUMN_TTL:
- return TSDataType.INT64;
- default:
- // do nothing
- }
-
- if (path.contains("(") && !path.startsWith("(") && path.endsWith(")")) {
- // aggregation
- int leftBracketIndex = path.indexOf('(');
- String aggrType = path.substring(0, leftBracketIndex);
- String innerPath = path.substring(leftBracketIndex + 1, path.length() - 1);
- switch (aggrType.toLowerCase()) {
- case SQLConstant.MIN_TIME:
- case SQLConstant.MAX_TIME:
- case SQLConstant.COUNT:
- return TSDataType.INT64;
- case SQLConstant.LAST_VALUE:
- case SQLConstant.FIRST_VALUE:
- case SQLConstant.MIN_VALUE:
- case SQLConstant.MAX_VALUE:
- return getSeriesType(innerPath);
- case SQLConstant.AVG:
- case SQLConstant.SUM:
- return TSDataType.DOUBLE;
- default:
- throw new QueryProcessException("aggregate does not support " + aggrType + " function.");
- }
- }
- TSDataType dataType;
- try {
- dataType = MManager.getInstance().getSeriesType(path);
- } catch (MetadataException e) {
- throw new QueryProcessException(e);
- }
-
- return dataType;
- }
-
@Override
public TSOpenSessionResp openSession(TSOpenSessionReq req) throws TException {
logger.info(
@@ -350,7 +305,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
/**
* release single operation resource
*/
- private void releaseQueryResource(long queryId) throws StorageEngineException {
+ protected void releaseQueryResource(long queryId) throws StorageEngineException {
// remove the corresponding Physical Plan
queryId2DataSet.remove(queryId);
QueryResourceManager.getInstance().endQuery(queryId);
@@ -377,6 +332,14 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
return new TSStatus(statusCodeAndMessage);
}
+ protected TSDataType getSeriesType(String path) throws QueryProcessException, MetadataException {
+ try {
+ return SchemaUtils.getSeriesType(path);
+ } catch (MetadataException e) {
+ throw new QueryProcessException(e);
+ }
+ }
+
@Override
public TSFetchMetadataResp fetchMetadata(TSFetchMetadataReq req) {
TSStatus status;
@@ -406,7 +369,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
status = getStatus(TSStatusCode.METADATA_ERROR, req.getType());
break;
}
- } catch (QueryProcessException | MetadataException | OutOfMemoryError e) {
+ } catch (MetadataException | OutOfMemoryError | QueryProcessException e) {
logger.error(
String.format("Failed to fetch timeseries %s's metadata", req.getColumnPath()), e);
status = getStatus(TSStatusCode.METADATA_ERROR, e.getMessage());
@@ -814,7 +777,11 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
}
for (String column : respColumns) {
- columnTypes.add(getSeriesType(column).toString());
+ try {
+ columnTypes.add(getSeriesType(column).toString());
+ } catch (MetadataException e) {
+ throw new QueryProcessException(e);
+ }
}
}
@@ -1047,12 +1014,16 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
throws QueryProcessException, QueryFilterOptimizationException, StorageEngineException,
IOException, MetadataException, SQLException {
- QueryContext context = new QueryContext(queryId);
+ QueryContext context = genQueryContext(queryId);
QueryDataSet queryDataSet = executor.processQuery(physicalPlan, context);
queryId2DataSet.put(queryId, queryDataSet);
return queryDataSet;
}
+ protected QueryContext genQueryContext(long queryId) {
+ return new QueryContext(queryId);
+ }
+
@Override
public TSExecuteStatementResp executeUpdateStatement(TSExecuteStatementReq req) {
try {
@@ -1149,7 +1120,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
return resp;
}
- void handleClientExit() {
+ protected void handleClientExit() {
Long sessionId = currSessionId.get();
if (sessionId != null) {
TSCloseSessionReq req = new TSCloseSessionReq(sessionId);
@@ -1435,7 +1406,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
return null;
}
- private TSStatus executePlan(PhysicalPlan plan) {
+ protected TSStatus executePlan(PhysicalPlan plan) {
boolean execRet;
try {
execRet = executeNonQuery(plan);
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManager.java b/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManager.java
index 6c4c569..14fbdc5 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManager.java
@@ -35,6 +35,7 @@ import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.engine.merge.task.MergeTask;
import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.sync.conf.SyncSenderDescriptor;
diff --git a/server/src/main/java/org/apache/iotdb/db/tools/IoTDBDataDirViewer.java b/server/src/main/java/org/apache/iotdb/db/tools/IoTDBDataDirViewer.java
index 780603f..262de5f 100644
--- a/server/src/main/java/org/apache/iotdb/db/tools/IoTDBDataDirViewer.java
+++ b/server/src/main/java/org/apache/iotdb/db/tools/IoTDBDataDirViewer.java
@@ -118,7 +118,7 @@ public class IoTDBDataDirViewer {
private static void printResource(String filename, PrintWriter pw) throws IOException {
filename = filename.substring(0, filename.length() - 9);
TsFileResource resource = new TsFileResource(SystemFileFactory.INSTANCE.getFile(filename));
- resource.deSerialize();
+ resource.deserialize();
// sort device strings
SortedSet<String> keys = new TreeSet<>(resource.getStartTimeMap().keySet());
for (String device : keys) {
diff --git a/server/src/main/java/org/apache/iotdb/db/tools/TsFileResourcePrinter.java b/server/src/main/java/org/apache/iotdb/db/tools/TsFileResourcePrinter.java
index d52a220..03e97d1 100644
--- a/server/src/main/java/org/apache/iotdb/db/tools/TsFileResourcePrinter.java
+++ b/server/src/main/java/org/apache/iotdb/db/tools/TsFileResourcePrinter.java
@@ -54,7 +54,7 @@ public class TsFileResourcePrinter {
filename = filename.substring(0, filename.length() - 9);
TsFileResource resource = new TsFileResource(SystemFileFactory.INSTANCE.getFile(filename));
System.err.println(String.format("analyzing %s ...", filename));
- resource.deSerialize();
+ resource.deserialize();
System.out.println("historicalVersions: " + resource.getHistoricalVersions());
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
index cd011ac..98b9311 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
@@ -44,7 +44,7 @@ public class FileLoaderUtils {
// write .resource file
tsFileResource.serialize();
} else {
- tsFileResource.deSerialize();
+ tsFileResource.deserialize();
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
index c18ce58..8ae97d8 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
@@ -18,15 +18,34 @@
*/
package org.apache.iotdb.db.utils;
+import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_PRIVILEGE;
+import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_ROLE;
+import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_STORAGE_GROUP;
+import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TTL;
+import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_USER;
+
+import java.util.Collections;
import java.util.List;
+import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.exception.metadata.PathAlreadyExistException;
import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.qp.constant.SQLConstant;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+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.write.schema.MeasurementSchema;
import org.apache.iotdb.tsfile.write.schema.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class SchemaUtils {
+ private static final Logger logger = LoggerFactory.getLogger(SchemaUtils.class);
+
private SchemaUtils() {
}
@@ -48,11 +67,75 @@ public class SchemaUtils {
* @param schemaList the schema of the columns in this file.
* @return a Schema contains the provided schemas.
*/
- public static Schema getSchemaFromColumnSchema(List<MeasurementSchema> schemaList) {
+ private static Schema getSchemaFromColumnSchema(List<MeasurementSchema> schemaList) {
Schema schema = new Schema();
for (MeasurementSchema measurementSchema : schemaList) {
schema.registerMeasurement(measurementSchema);
}
return schema;
}
+
+ public static void registerTimeseries(MeasurementSchema schema) {
+ try {
+ logger.debug("Registering timeseries {}", schema);
+ String path = schema.getMeasurementId();
+ TSDataType dataType = schema.getType();
+ TSEncoding encoding = schema.getEncodingType();
+ CompressionType compressionType = schema.getCompressor();
+ boolean result = MManager.getInstance().createTimeseries(path, dataType, encoding,
+ compressionType, Collections.emptyMap());
+ if (result) {
+ StorageEngine.getInstance().addTimeSeries(new Path(path), dataType, encoding,
+ compressionType, Collections.emptyMap());
+ }
+ } catch (PathAlreadyExistException ignored) {
+ // ignore added timeseries
+ } catch (MetadataException | StorageEngineException e) {
+ logger.error("Cannot create timeseries {} in snapshot, ignored", schema.getMeasurementId(),
+ e);
+ }
+
+ }
+
+ public static TSDataType getSeriesType(String path)
+ throws MetadataException {
+ switch (path.toLowerCase()) {
+ // authorization queries
+ case COLUMN_ROLE:
+ case COLUMN_USER:
+ case COLUMN_PRIVILEGE:
+ case COLUMN_STORAGE_GROUP:
+ return TSDataType.TEXT;
+ case SQLConstant.RESERVED_TIME:
+ case COLUMN_TTL:
+ return TSDataType.INT64;
+ default:
+ // do nothing
+ }
+
+ if (path.contains("(") && !path.startsWith("(") && path.endsWith(")")) {
+ // aggregation
+ int leftBracketIndex = path.indexOf('(');
+ String aggrType = path.substring(0, leftBracketIndex);
+ String innerPath = path.substring(leftBracketIndex + 1, path.length() - 1);
+ switch (aggrType.toLowerCase()) {
+ case SQLConstant.MIN_TIME:
+ case SQLConstant.MAX_TIME:
+ case SQLConstant.COUNT:
+ return TSDataType.INT64;
+ case SQLConstant.LAST_VALUE:
+ case SQLConstant.FIRST_VALUE:
+ case SQLConstant.MIN_VALUE:
+ case SQLConstant.MAX_VALUE:
+ return getSeriesType(innerPath);
+ case SQLConstant.AVG:
+ case SQLConstant.SUM:
+ return TSDataType.DOUBLE;
+ default:
+ throw new MetadataException(
+ "aggregate does not support " + aggrType + " function.");
+ }
+ }
+ return MManager.getInstance().getSeriesType(path);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/TestOnly.java b/server/src/main/java/org/apache/iotdb/db/utils/TestOnly.java
index d4058ba..fa1acc2 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/TestOnly.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/TestOnly.java
@@ -27,5 +27,9 @@ import java.lang.annotation.Target;
@Target({ElementType.METHOD, ElementType.CONSTRUCTOR})
@Retention(RetentionPolicy.SOURCE)
+/**
+ * TestOnly implies that the method should only be used in the tests, otherwise its functionality
+ * is not guaranteed and may interfere with the normal code.
+ */
public @interface TestOnly {
}
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java b/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java
index 941b494..bd0c2d6 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java
@@ -150,7 +150,7 @@ public class TsFileRecoverPerformer {
private void recoverResourceFromFile() throws IOException {
try {
- tsFileResource.deSerialize();
+ tsFileResource.deserialize();
} catch (IOException e) {
logger.warn("Cannot deserialize TsFileResource {}, construct it using "
+ "TsFileSequenceReader", tsFileResource.getFile(), e);
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeLogTest.java b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeLogTest.java
index cf4bd93..4ad0991 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeLogTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeLogTest.java
@@ -33,7 +33,6 @@ import org.apache.iotdb.db.engine.merge.manage.MergeResource;
import org.apache.iotdb.db.engine.merge.task.MergeTask;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.metadata.MetadataException;
-import org.apache.iotdb.db.exception.query.PathException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
import org.junit.After;
@@ -45,7 +44,7 @@ public class MergeLogTest extends MergeTest {
File tempSGDir;
@Before
- public void setUp() throws IOException, WriteProcessException, MetadataException, PathException {
+ public void setUp() throws IOException, WriteProcessException, MetadataException {
super.setUp();
tempSGDir = new File(TestConstant.BASE_OUTPUT_PATH.concat("tempSG"));
tempSGDir.mkdirs();
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeOverLapTest.java b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeOverLapTest.java
index a4b5e7c..656a215 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeOverLapTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeOverLapTest.java
@@ -26,9 +26,8 @@ import org.apache.iotdb.db.constant.TestConstant;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
import org.apache.iotdb.db.engine.merge.task.MergeTask;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
-import org.apache.iotdb.db.exception.query.PathException;
+import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.reader.series.SeriesRawDataBatchReader;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
@@ -57,7 +56,7 @@ public class MergeOverLapTest extends MergeTest {
@Before
public void setUp()
- throws IOException, WriteProcessException, MetadataException, PathException {
+ throws IOException, WriteProcessException, MetadataException {
ptNum = 1000;
super.setUp();
tempSGDir = new File(TestConstant.BASE_OUTPUT_PATH.concat("tempSG"));
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTaskTest.java b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTaskTest.java
index e333c66..7b61725 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTaskTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTaskTest.java
@@ -25,10 +25,9 @@ import org.apache.iotdb.db.constant.TestConstant;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
import org.apache.iotdb.db.engine.merge.task.MergeTask;
import org.apache.iotdb.db.engine.modification.Deletion;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.StorageEngineException;
-import org.apache.iotdb.db.exception.metadata.MetadataException;
-import org.apache.iotdb.db.exception.query.PathException;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.reader.series.SeriesRawDataBatchReader;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
@@ -51,7 +50,7 @@ public class MergeTaskTest extends MergeTest {
private File tempSGDir;
@Before
- public void setUp() throws IOException, WriteProcessException, MetadataException, PathException {
+ public void setUp() throws IOException, WriteProcessException, MetadataException, MetadataException {
super.setUp();
tempSGDir = new File(TestConstant.BASE_OUTPUT_PATH.concat("tempSG"));
tempSGDir.mkdirs();
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTest.java b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTest.java
index 21a4683..028f0d0 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTest.java
@@ -34,7 +34,6 @@ import org.apache.iotdb.db.engine.cache.TsFileMetaDataCache;
import org.apache.iotdb.db.engine.merge.manage.MergeManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.metadata.MetadataException;
-import org.apache.iotdb.db.exception.query.PathException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.query.control.FileReaderManager;
@@ -71,7 +70,7 @@ abstract class MergeTest {
private int prevMergeChunkThreshold;
@Before
- public void setUp() throws IOException, WriteProcessException, MetadataException, PathException {
+ public void setUp() throws IOException, WriteProcessException, MetadataException, MetadataException {
MManager.getInstance().init();
prevMergeChunkThreshold =
IoTDBDescriptor.getInstance().getConfig().getChunkMergePointThreshold();
@@ -94,7 +93,7 @@ abstract class MergeTest {
MergeManager.getINSTANCE().stop();
}
- private void prepareSeries() throws MetadataException, PathException {
+ private void prepareSeries() throws MetadataException, MetadataException {
measurementSchemas = new MeasurementSchema[measurementNum];
for (int i = 0; i < measurementNum; i++) {
measurementSchemas[i] = new MeasurementSchema("sensor" + i, TSDataType.DOUBLE,
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java b/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java
index 5be94ea..1ad0d22 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java
@@ -35,19 +35,16 @@ import org.apache.iotdb.db.conf.directories.DirectoryManager;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.modification.io.LocalTextModificationAccessor;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
-import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.exception.StorageEngineException;
-import org.apache.iotdb.db.exception.metadata.MetadataException;
-import org.apache.iotdb.db.exception.query.PathException;
+import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.db.exception.storageGroup.StorageGroupException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.control.QueryResourceManager;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
-import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.read.reader.IPointReader;
@@ -232,7 +229,7 @@ public class DeletionFileNodeTest {
@Test
public void testDeleteInOverflowFile()
- throws StorageEngineException, QueryProcessException, IOException {
+ throws StorageEngineException, QueryProcessException {
// insert into BufferWrite
for (int i = 101; i <= 200; i++) {
TSRecord record = new TSRecord(i, processorName);
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionQueryTest.java b/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionQueryTest.java
index 9e4f16a..ce79ceb 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionQueryTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionQueryTest.java
@@ -30,7 +30,6 @@ import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
-import org.apache.iotdb.db.exception.query.PathException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.exception.storageGroup.StorageGroupException;
import org.apache.iotdb.db.metadata.MManager;
@@ -39,7 +38,6 @@ import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
import org.apache.iotdb.db.query.executor.QueryRouter;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
-import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
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;
@@ -66,8 +64,8 @@ public class DeletionQueryTest {
}
@Before
- public void setup() throws MetadataException,
- PathException, IOException, StorageEngineException, StartupException, StorageGroupException {
+ public void setup() throws
+ MetadataException, StorageEngineException, StartupException {
EnvironmentUtils.envSetUp();
MManager.getInstance().setStorageGroup(processorName);
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/FileNodeManagerBenchmark.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/FileNodeManagerBenchmark.java
index 7fea61d..fc8d93a 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/FileNodeManagerBenchmark.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/FileNodeManagerBenchmark.java
@@ -24,7 +24,6 @@ import java.util.concurrent.atomic.AtomicLong;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.StorageEngineException;
-import org.apache.iotdb.db.exception.query.PathException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.exception.storageGroup.StorageGroupException;
import org.apache.iotdb.db.metadata.MManager;
@@ -65,7 +64,7 @@ public class FileNodeManagerBenchmark {
}
private static void prepare()
- throws MetadataException, PathException, IOException, StorageGroupException {
+ throws MetadataException {
MManager manager = MManager.getInstance();
manager.setStorageGroup(prefix);
for (String device : devices) {
@@ -81,8 +80,8 @@ public class FileNodeManagerBenchmark {
}
public static void main(String[] args)
- throws InterruptedException, IOException, MetadataException,
- PathException, StorageEngineException, StorageGroupException {
+ throws InterruptedException, IOException,
+ MetadataException, StorageEngineException {
tearDown();
prepare();
long startTime = System.currentTimeMillis();
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
index 15ce293..81fd0ea 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
@@ -37,9 +37,7 @@ import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.OutOfTTLException;
-import org.apache.iotdb.db.exception.query.PathException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.db.exception.storageGroup.StorageGroupException;
import org.apache.iotdb.db.exception.storageGroup.StorageGroupProcessorException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.metadata.mnode.StorageGroupMNode;
@@ -76,7 +74,7 @@ public class TTLTest {
@Before
public void setUp()
- throws MetadataException, IOException, StartupException, PathException, StorageGroupProcessorException {
+ throws MetadataException, IOException, StartupException, StorageGroupProcessorException {
IoTDBDescriptor.getInstance().getConfig().setPartitionInterval(86400);
EnvironmentUtils.envSetUp();
createSchemas();
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBMultiSeriesIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBMultiSeriesIT.java
index a814340..5d98103 100644
--- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBMultiSeriesIT.java
+++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBMultiSeriesIT.java
@@ -408,7 +408,7 @@ public class IoTDBMultiSeriesIT {
fail("not throw exception when unknown time series in where clause");
} catch (SQLException e) {
assertEquals(
- "401: Statement format is not right: Filter has some time series don't correspond to any known time series",
+ "Statement format is not right: Filter has some time series don't correspond to any known time series",
e.getMessage());
}
}
@@ -426,7 +426,7 @@ public class IoTDBMultiSeriesIT {
} catch (SQLException e) {
e.printStackTrace();
assertEquals(
- "401: Statement format is not right: Path [root.vehicle.d0.s10] does not exist",
+ "org.apache.iotdb.db.exception.metadata.PathNotExistException: Path [root.vehicle.d0.s10] does not exist",
e.getMessage());
}
}
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 04c9cf2..2fa9efe 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
@@ -26,7 +26,6 @@ import static org.junit.Assert.fail;
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
-import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
@@ -98,7 +97,7 @@ public class IoTDBSequenceDataQueryIT {
EnvironmentUtils.cleanEnv();
}
- private static void insertData() throws ClassNotFoundException, SQLException {
+ private static void insertData() throws ClassNotFoundException {
Class.forName(Config.JDBC_DRIVER_NAME);
try (Connection connection = DriverManager
.getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
@@ -133,7 +132,7 @@ public class IoTDBSequenceDataQueryIT {
// insert data (time from 1200-1499)
for (long time = 1200; time < 1500; time++) {
- String sql = null;
+ String sql;
if (time % 2 == 0) {
sql = String
.format("insert into root.vehicle.d0(timestamp,s0) values(%s,%s)", time, time % 17);
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 9d16ee0..232d9ad 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
@@ -104,7 +104,7 @@ public class IoTDBSeriesReaderIT {
EnvironmentUtils.cleanEnv();
}
- private static void insertData() throws ClassNotFoundException, SQLException {
+ private static void insertData() throws ClassNotFoundException {
Class.forName(Config.JDBC_DRIVER_NAME);
try (Connection connection = DriverManager
.getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
@@ -341,6 +341,7 @@ public class IoTDBSeriesReaderIT {
Path path2 = new Path(Constant.d0s1);
SingleSeriesExpression singleSeriesExpression = new SingleSeriesExpression(path1,
ValueFilter.lt(111));
+
List<Path> pathList = new ArrayList<>();
List<TSDataType> dataTypes = new ArrayList<>();
pathList.add(path1);
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBTtlIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBTtlIT.java
index 341627a..4fd5f13 100644
--- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBTtlIT.java
+++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBTtlIT.java
@@ -142,7 +142,6 @@ public class IoTDBTtlIT {
Statement statement = connection.createStatement()) {
statement.execute("SET STORAGE GROUP TO root.group1");
statement.execute("SET STORAGE GROUP TO root.group2");
-
String result = doQuery(statement, "SHOW ALL TTL");
assertEquals("root.group1,null\n"
+ "root.group2,null\n", result);
diff --git a/server/src/test/java/org/apache/iotdb/db/qp/plan/ConcatOptimizerTest.java b/server/src/test/java/org/apache/iotdb/db/qp/plan/ConcatOptimizerTest.java
index f6d6b67..015a094 100644
--- a/server/src/test/java/org/apache/iotdb/db/qp/plan/ConcatOptimizerTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/qp/plan/ConcatOptimizerTest.java
@@ -23,7 +23,6 @@ import static org.junit.Assert.assertEquals;
import java.io.IOException;
import org.antlr.v4.runtime.RecognitionException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
-import org.apache.iotdb.db.exception.query.PathException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.qp.Planner;
@@ -49,7 +48,7 @@ public class ConcatOptimizerTest {
private Planner processor;
@Before
- public void before() throws MetadataException, PathException {
+ public void before() throws MetadataException {
processor = new Planner();
MManager.getInstance().init();
MManager.getInstance().setStorageGroup("root.laptop");
diff --git a/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java b/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java
index c2371c9..4a5be46 100644
--- a/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java
@@ -92,8 +92,7 @@ public class PhysicalPlanTest {
String metadata = "create timeseries root.vehicle.d1.s2 with datatype=INT32,encoding=RLE";
Planner processor = new Planner();
CreateTimeSeriesPlan plan = (CreateTimeSeriesPlan) processor.parseSQLToPhysicalPlan(metadata);
- assertEquals(String.format("seriesPath: root.vehicle.d1.s2%n" + "resultDataType: INT32%n" +
- "encoding: RLE%nnamespace type: ADD_PATH%n" + "args: "), plan.toString());
+ assertEquals("seriesPath: root.vehicle.d1.s2, resultDataType: INT32, encoding: RLE, compression: SNAPPY", plan.toString());
}
@Test
@@ -101,8 +100,7 @@ public class PhysicalPlanTest {
String metadata = "create timeseries root.vehicle.d1.s2 with datatype=int32,encoding=rle";
Planner processor = new Planner();
CreateTimeSeriesPlan plan = (CreateTimeSeriesPlan) processor.parseSQLToPhysicalPlan(metadata);
- assertEquals(String.format("seriesPath: root.vehicle.d1.s2%n" + "resultDataType: INT32%n" +
- "encoding: RLE%nnamespace type: ADD_PATH%n" + "args: "), plan.toString());
+ assertEquals("seriesPath: root.vehicle.d1.s2, resultDataType: INT32, encoding: RLE, compression: SNAPPY", plan.toString());
}
@Test
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TimeValuePair.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TimeValuePair.java
index c52ee5b..20ebd90 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TimeValuePair.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TimeValuePair.java
@@ -78,4 +78,6 @@ public class TimeValuePair implements Serializable, Comparable<TimeValuePair> {
public int compareTo(TimeValuePair o) {
return Long.compare(this.getTimestamp(), o.getTimestamp());
}
+
+
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/BatchData.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/BatchData.java
index eb483ac..a4c316f 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/BatchData.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/BatchData.java
@@ -18,16 +18,20 @@
*/
package org.apache.iotdb.tsfile.read.common;
+import java.io.Serializable;
+import java.util.ArrayList;
import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.reader.BatchDataIterator;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
-import org.apache.iotdb.tsfile.utils.TsPrimitiveType.*;
-
-import java.io.Serializable;
-import java.util.ArrayList;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsBinary;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsBoolean;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsDouble;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsFloat;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsInt;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsLong;
/**
* <code>BatchData</code> is a self-defined data structure which is optimized for different type of
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/GroupByFilter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/GroupByFilter.java
index e8cdb81..f2d82bf 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/GroupByFilter.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/GroupByFilter.java
@@ -146,4 +146,12 @@ public class GroupByFilter implements Filter, Serializable {
public int hashCode() {
return Objects.hash(interval, slidingStep, startTime, endTime);
}
+
+ public long getStartTime() {
+ return startTime;
+ }
+
+ public long getEndTime() {
+ return endTime;
+ }
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/In.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/In.java
index 1bb76d4..3767f01 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/In.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/In.java
@@ -118,4 +118,8 @@ public class In<T extends Comparable<T>> implements Filter {
public FilterSerializeId getSerializeId() {
return FilterSerializeId.IN;
}
+
+ public Set<T> getValues() {
+ return values;
+ }
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Murmur128Hash.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Murmur128Hash.java
index 2bb5118..80ebf0a 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Murmur128Hash.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Murmur128Hash.java
@@ -23,29 +23,44 @@ import java.nio.ByteBuffer;
public class Murmur128Hash {
/**
- * get hashcode of value by seed
+ * get hashcode of value by seed
+ *
* @param value value
* @param seed seend
* @return hashcode of value
*/
- public static int hash(String value, int seed){
- return (int) hash3_x64_128(ByteBuffer.wrap(value.getBytes()), 0, value.length(), seed);
+ public static int hash(String value, int seed) {
+ return (int) hash3_x64_128(value.getBytes(), 0, value.getBytes().length, seed);
+ }
+
+ /**
+ * get hashcode of two values by seed
+ *
+ * @param value1 the first value
+ * @param value2 the second value
+ * @param seed seend
+ * @return hashcode of value
+ */
+ public static int hash(String value1, long value2, int seed) {
+ return (int) hash3_x64_128(
+ BytesUtils.concatByteArray(value1.getBytes(), BytesUtils.longToBytes(value2)), 0,
+ value1.length() + 8, seed);
}
/**************************************
* Methods to perform murmur 128 hash.
**************************************/
- private static long getBlock(ByteBuffer key, int offset, int index) {
+ private static long getBlock(byte[] key, int offset, int index) {
int i_8 = index << 3;
int blockOffset = offset + i_8;
- return ((long) key.get(blockOffset) & 0xff) + (((long) key.get(blockOffset + 1) & 0xff) << 8)
+ return ((long) key[blockOffset] & 0xff) + (((long) key[blockOffset + 1] & 0xff) << 8)
+
- (((long) key.get(blockOffset + 2) & 0xff) << 16) + (
- ((long) key.get(blockOffset + 3) & 0xff) << 24) +
- (((long) key.get(blockOffset + 4) & 0xff) << 32) + (
- ((long) key.get(blockOffset + 5) & 0xff) << 40) +
- (((long) key.get(blockOffset + 6) & 0xff) << 48) + (
- ((long) key.get(blockOffset + 7) & 0xff) << 56);
+ (((long) key[blockOffset + 2] & 0xff) << 16) + (
+ ((long) key[blockOffset + 3] & 0xff) << 24) +
+ (((long) key[blockOffset + 4] & 0xff) << 32) + (
+ ((long) key[blockOffset + 5] & 0xff) << 40) +
+ (((long) key[blockOffset + 6] & 0xff) << 48) + (
+ ((long) key[blockOffset + 7] & 0xff) << 56);
}
private static long rotl64(long v, int n) {
@@ -61,7 +76,7 @@ public class Murmur128Hash {
return k;
}
- private static long hash3_x64_128(ByteBuffer key, int offset, int length, long seed) {
+ private static long hash3_x64_128(byte[] key, int offset, int length, long seed) {
final int nblocks = length >> 4; // Process as 128-bit blocks.
long h1 = seed;
long h2 = seed;
@@ -95,53 +110,53 @@ public class Murmur128Hash {
long k2 = 0;
switch (length & 15) {
case 15:
- k2 ^= ((long) key.get(offset + 14)) << 48;
+ k2 ^= ((long) key[offset + 14]) << 48;
// fallthrough
case 14:
- k2 ^= ((long) key.get(offset + 13)) << 40;
+ k2 ^= ((long) key[offset + 13]) << 40;
// fallthrough
case 13:
- k2 ^= ((long) key.get(offset + 12)) << 32;
+ k2 ^= ((long) key[offset + 12]) << 32;
// fallthrough
case 12:
- k2 ^= ((long) key.get(offset + 11)) << 24;
+ k2 ^= ((long) key[offset + 11]) << 24;
// fallthrough
case 11:
- k2 ^= ((long) key.get(offset + 10)) << 16;
+ k2 ^= ((long) key[offset + 10]) << 16;
// fallthrough
case 10:
- k2 ^= ((long) key.get(offset + 9)) << 8;
+ k2 ^= ((long) key[offset + 9]) << 8;
// fallthrough
case 9:
- k2 ^= ((long) key.get(offset + 8));
+ k2 ^= ((long) key[offset + 8]);
k2 *= c2;
k2 = rotl64(k2, 33);
k2 *= c1;
h2 ^= k2;
// fallthrough
case 8:
- k1 ^= ((long) key.get(offset + 7)) << 56;
+ k1 ^= ((long) key[offset + 7]) << 56;
// fallthrough
case 7:
- k1 ^= ((long) key.get(offset + 6)) << 48;
+ k1 ^= ((long) key[offset + 6]) << 48;
// fallthrough
case 6:
- k1 ^= ((long) key.get(offset + 5)) << 40;
+ k1 ^= ((long) key[offset + 5]) << 40;
// fallthrough
case 5:
- k1 ^= ((long) key.get(offset + 4)) << 32;
+ k1 ^= ((long) key[offset + 4]) << 32;
// fallthrough
case 4:
- k1 ^= ((long) key.get(offset + 3)) << 24;
+ k1 ^= ((long) key[offset + 3]) << 24;
// fallthrough
case 3:
- k1 ^= ((long) key.get(offset + 2)) << 16;
+ k1 ^= ((long) key[offset + 2]) << 16;
// fallthrough
case 2:
- k1 ^= ((long) key.get(offset + 1)) << 8;
+ k1 ^= ((long) key[offset + 1]) << 8;
// fallthrough
case 1:
- k1 ^= (key.get(offset));
+ k1 ^= (key[offset]);
k1 *= c1;
k1 = rotl64(k1, 31);
k1 *= c2;
@@ -164,4 +179,5 @@ public class Murmur128Hash {
return h1 + h2;
}
+
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java
index 9d89621..2e305d9 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java
@@ -25,6 +25,7 @@ import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.DO
import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.FLOAT;
import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.INTEGER;
import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.LONG;
+import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.NULL;
import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.STRING;
import java.io.DataOutputStream;
@@ -781,7 +782,7 @@ public class ReadWriteIOUtils {
}
enum ClassSerializeId {
- LONG, DOUBLE, INTEGER, FLOAT, BINARY, BOOLEAN, STRING
+ LONG, DOUBLE, INTEGER, FLOAT, BINARY, BOOLEAN, STRING, NULL
}
public static void writeObject(Object value, DataOutputStream outputStream) {
@@ -806,6 +807,8 @@ public class ReadWriteIOUtils {
} else if (value instanceof Boolean) {
outputStream.write(BOOLEAN.ordinal());
outputStream.write(((Boolean) value) ? 1 : 0);
+ } else if (value == null) {
+ outputStream.write(NULL.ordinal());
} else {
outputStream.write(STRING.ordinal());
byte[] bytes = value.toString().getBytes();
@@ -835,6 +838,8 @@ public class ReadWriteIOUtils {
byte[] bytes = new byte[length];
buffer.get(bytes);
return new Binary(bytes);
+ case NULL:
+ return null;
case STRING:
default:
length = buffer.getInt();
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schema/MeasurementSchema.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schema/MeasurementSchema.java
index b0c4a34..91d64f6 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schema/MeasurementSchema.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schema/MeasurementSchema.java
@@ -318,4 +318,5 @@ public class MeasurementSchema implements Comparable<MeasurementSchema>, Seriali
sc.addTail("]");
return sc.toString();
}
+
}