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 2019/06/06 10:26:05 UTC
[incubator-iotdb] branch refactor_overflow updated (b34cb2a ->
458c20f)
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a change to branch refactor_overflow
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git.
from b34cb2a fix tests
add f8754f9 add the print of information about chunkgroups for debug (#175)
add 451e87d change print to log (#176)
add 0043172 Update README.md
add e7886ca Refactor iotdb code for cluster (#172)
add 8fc0c7e fix unstable unit test and polish ReadOnlyTsFile query logic (#177)
add 5b26774 modify error statement in doc (#182)
add 84bdd5a [IOTDB-100]Fix return error message while executing sum aggregation query (#181)
add 2fbd949 [IOTDB-95]keep stack trace when logging or throwing an Exception (#180)
add 000891f fix bug where IOTDB_HOME is misused (#184)
add 5cc46a9 [IOTDB-96]Refactor author query (#179)
add 343c7bd [IOTDB-108]Fix mistakes in doc (#187)
add e2fb430 fix that ignoreTimeStamp in response is incorrectly set and used (#191)
add 5c15ac3 Organize properties (#189)
add 68844b2 fix missing wal log node after recovery
add 6f40c05 Merge pull request #186 from apache/fix_missing_wal_node
new 458c20f Merge branch 'master' into refactor_overflow
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
README.md | 9 +
docs/Documentation/QuickStart.md | 10 +
.../UserGuideV0.7.0/4-Deployment and Management.md | 71 ++-
.../UserGuideV0.7.0/5-SQL Documentation.md | 17 +-
.../iotdb/tsfile/hadoop/TSFRecordWriter.java | 11 +-
.../iotdb/tsfile/hadoop/example/TsFileHelper.java | 6 +-
.../cn/edu/thu/tsfile/hadoop/TsFileTestHelper.java | 6 +-
iotdb-cli/cli/bin/export-csv.bat | 10 +-
iotdb-cli/cli/bin/export-csv.sh | 8 +-
iotdb-cli/cli/bin/import-csv.bat | 10 +-
iotdb-cli/cli/bin/import-csv.sh | 8 +-
iotdb-cli/cli/bin/start-client.bat | 8 +-
iotdb-cli/cli/bin/start-client.sh | 6 +-
.../apache/iotdb/cli/client/AbstractClient.java | 4 +
.../java/org/apache/iotdb/cli/tool/ImportCsv.java | 4 +-
iotdb/iotdb/conf/iotdb-engine.properties | 68 ++-
iotdb/iotdb/conf/iotdb-sync-client.properties | 4 +-
.../iotdb/db/auth/authorizer/BasicAuthorizer.java | 3 +-
.../apache/iotdb/db/auth/entity/PathPrivilege.java | 1 -
.../db/concurrent/IoTDBThreadPoolFactory.java | 8 +-
.../org/apache/iotdb/db/concurrent/ThreadName.java | 2 +-
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 185 +++----
.../org/apache/iotdb/db/conf/IoTDBConstant.java | 10 +-
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 17 +-
.../java/org/apache/iotdb/db/engine/Processor.java | 2 +-
.../apache/iotdb/db/engine/pool/FlushManager.java | 6 +-
.../apache/iotdb/db/engine/pool/MergeManager.java | 2 -
.../db/engine/sgmanager/StorageGroupProcessor.java | 2 +-
.../db/engine/tsfiledata/TsFileProcessor.java | 182 +++----
...eckException.java => ConsistencyException.java} | 11 +-
.../db/exception/qp/IllegalASTFormatException.java | 7 +
.../db/exception/qp/LogicalOperatorException.java | 7 +
.../db/exception/qp/LogicalOptimizeException.java | 7 +
.../db/exception/qp/QueryProcessorException.java | 4 +
.../java/org/apache/iotdb/db/metadata/MGraph.java | 12 +-
.../org/apache/iotdb/db/metadata/MManager.java | 13 -
.../java/org/apache/iotdb/db/metadata/MTree.java | 32 --
.../org/apache/iotdb/db/metadata/Metadata.java | 106 +++-
.../org/apache/iotdb/db/qp/QueryProcessor.java | 13 +-
.../db/qp/executor/IQueryProcessExecutor.java | 134 +++++
.../iotdb/db/qp/executor/OverflowQPExecutor.java | 266 ++++++----
.../iotdb/db/qp/executor/QueryProcessExecutor.java | 56 +-
.../org/apache/iotdb/db/qp/logical/Operator.java | 2 +-
.../db/qp/logical/crud/BasicFunctionOperator.java | 5 +-
.../iotdb/db/qp/logical/crud/FilterOperator.java | 5 +-
.../apache/iotdb/db/qp/physical/PhysicalPlan.java | 8 +-
.../iotdb/db/qp/physical/crud/AggregationPlan.java | 1 +
.../iotdb/db/qp/physical/crud/DeletePlan.java | 1 +
.../iotdb/db/qp/physical/crud/FillQueryPlan.java | 1 +
.../iotdb/db/qp/physical/crud/GroupByPlan.java | 1 +
.../iotdb/db/qp/physical/crud/InsertPlan.java | 6 +
.../iotdb/db/qp/physical/crud/QueryPlan.java | 5 +-
.../iotdb/db/qp/physical/crud/UpdatePlan.java | 1 +
.../iotdb/db/qp/physical/sys/AuthorPlan.java | 7 +
.../iotdb/db/qp/physical/sys/LoadDataPlan.java | 1 +
.../iotdb/db/qp/physical/sys/MetadataPlan.java | 1 +
.../iotdb/db/qp/physical/sys/PropertyPlan.java | 1 +
.../{writelog => qp/physical}/transfer/Codec.java | 3 +-
.../physical}/transfer/CodecInstances.java | 5 +-
.../physical}/transfer/PhysicalPlanCodec.java | 14 +-
.../transfer/PhysicalPlanLogTransfer.java | 2 +-
.../physical}/transfer/SystemLogOperator.java | 2 +-
.../iotdb/db/qp/strategy/PhysicalGenerator.java | 8 +-
.../qp/strategy/optimizer/ConcatPathOptimizer.java | 14 +-
.../db/query/aggregation/AggreResultData.java | 2 +-
.../db/query/aggregation/AggregateFunction.java | 1 -
.../db/query/aggregation/impl/CountAggrFunc.java | 2 +-
.../db/query/aggregation/impl/FirstAggrFunc.java | 2 +-
.../db/query/aggregation/impl/LastAggrFunc.java | 2 +-
.../db/query/aggregation/impl/MaxTimeAggrFunc.java | 2 +-
.../query/aggregation/impl/MaxValueAggrFunc.java | 2 +-
.../db/query/aggregation/impl/MeanAggrFunc.java | 13 +-
.../db/query/aggregation/impl/MinTimeAggrFunc.java | 2 +-
.../query/aggregation/impl/MinValueAggrFunc.java | 2 +-
.../db/query/aggregation/impl/SumAggrFunc.java | 10 +
.../db/query/control/QueryResourceManager.java | 21 +-
.../query/dataset/AggreResultDataPointReader.java | 2 +-
.../apache/iotdb/db/query/dataset/AuthDataSet.java | 45 +-
.../dataset/EngineDataSetWithTimeGenerator.java | 35 +-
.../dataset/EngineDataSetWithoutTimeGenerator.java | 4 +
.../groupby/GroupByEngineDataSet.java | 11 +-
.../groupby/GroupByWithOnlyTimeFilterDataSet.java} | 8 +-
.../groupby/GroupByWithValueFilterDataSet.java} | 8 +-
.../AbstractExecutorWithoutTimeGenerator.java | 86 +++
.../db/query/executor/AggregateEngineExecutor.java | 2 +-
.../EngineExecutorWithoutTimeGenerator.java | 46 +-
.../iotdb/db/query/executor/EngineQueryRouter.java | 44 +-
.../db/query/executor/IEngineQueryRouter.java | 78 +++
.../{aggregation => factory}/AggreFuncFactory.java | 3 +-
...nstructor.java => AbstractNodeConstructor.java} | 62 +--
.../query/timegenerator/EngineNodeConstructor.java | 65 +--
.../query/timegenerator/EngineTimeGenerator.java | 1 -
.../apache/iotdb/db/service/CloseMergeService.java | 6 +-
.../java/org/apache/iotdb/db/service/IoTDB.java | 14 +-
.../org/apache/iotdb/db/service/JDBCService.java | 17 +-
.../iotdb/db/service/JDBCServiceEventHandler.java | 6 +-
.../org/apache/iotdb/db/service/JMXService.java | 5 +-
.../java/org/apache/iotdb/db/service/Monitor.java | 3 +-
.../apache/iotdb/db/service/RegisterManager.java | 2 +-
.../org/apache/iotdb/db/service/TSServiceImpl.java | 591 +++++++++++----------
.../org/apache/iotdb/db/sql/parse/ParseDriver.java | 2 +-
.../org/apache/iotdb/db/sync/conf/Constans.java | 10 +
.../iotdb/db/sync/conf/SyncSenderConfig.java | 24 +-
.../iotdb/db/sync/conf/SyncSenderDescriptor.java | 7 +-
.../iotdb/db/sync/receiver/SyncServiceImpl.java | 2 +-
.../iotdb/db/sync/receiver/SyncServiceManager.java | 5 +-
.../apache/iotdb/db/sync/sender/SyncSender.java | 10 +
.../iotdb/db/sync/sender/SyncSenderImpl.java | 106 ++--
.../org/apache/iotdb/db/utils/RecordUtils.java | 12 +-
.../apache/iotdb/db/writelog/io/RAFLogReader.java | 2 +-
.../writelog/manager/MultiFileLogNodeManager.java | 9 +-
.../db/writelog/node/ExclusiveWriteLogNode.java | 4 +-
.../recover/ExclusiveLogRecoverPerformer.java | 15 +-
.../db/concurrent/IoTDBThreadPoolFactoryTest.java | 2 +-
.../db/engine/tsfiledata/TsFileProcessorTest.java | 2 +-
.../iotdb/db/integration/IoTDBAggregationIT.java | 86 ++-
.../iotdb/db/integration/IoTDBAuthorizationIT.java | 416 ++++++++-------
.../iotdb/db/integration/IoTDBSeriesReaderIT.java | 2 +-
.../iotdb/db/metadata/MManagerBasicTest.java | 29 +-
.../org/apache/iotdb/db/metadata/MTreeTest.java | 28 +-
.../org/apache/iotdb/db/metadata/MetadataTest.java | 93 ++++
.../org/apache/iotdb/db/qp/QueryProcessorTest.java | 88 ++-
.../transfer/PhysicalPlanLogTransferTest.java | 2 +-
.../org/apache/iotdb/db/qp/plan/QPUpdateTest.java | 2 +-
.../apache/iotdb/db/qp/utils/MemIntQpExecutor.java | 13 +-
.../EngineDataSetWithTimeGeneratorTest.java | 2 +-
.../query/executor/GroupByEngineDataSetTest.java | 14 +-
.../org/apache/iotdb/db/tools/WalCheckerTest.java | 2 +-
.../apache/iotdb/db/utils/EnvironmentUtils.java | 2 -
.../apache/iotdb/db/writelog/PerformanceTest.java | 2 +-
.../apache/iotdb/db/writelog/WriteLogNodeTest.java | 2 +-
.../iotdb/db/writelog/io/LogWriterReaderTest.java | 3 +-
jdbc/README.md | 4 +-
.../org/apache/iotdb/jdbc/IoTDBConnection.java | 6 +-
.../org/apache/iotdb/jdbc/IoTDBQueryResultSet.java | 9 +
.../java/org/apache/iotdb/jdbc/IoTDBStatement.java | 15 +-
service-rpc/src/main/thrift/rpc.thrift | 1 +
.../org/apache/iotdb/tsfile/DefaultSource.scala | 9 +-
.../tsfile/common/constant/QueryConstant.java | 6 -
.../tsfile/exception/write/PageException.java | 8 +
.../exception/write/WriteProcessException.java | 16 +-
.../apache/iotdb/tsfile/read/ReadOnlyTsFile.java | 17 +-
.../apache/iotdb/tsfile/read/common/BatchData.java | 4 +-
.../org/apache/iotdb/tsfile/read/common/Field.java | 22 +
.../org/apache/iotdb/tsfile/read/common/Path.java | 4 +-
.../apache/iotdb/tsfile/read/common/TimeRange.java | 225 ++++----
.../tsfile/read/controller/MetadataQuerier.java | 39 +-
.../read/controller/MetadataQuerierByFileImpl.java | 189 +++----
.../tsfile/read/expression/ExpressionType.java | 30 +-
.../tsfile/read/expression/IBinaryExpression.java | 4 +
.../iotdb/tsfile/read/expression/IExpression.java | 4 +-
.../read/expression/impl/BinaryExpression.java | 38 +-
.../read/expression/impl/GlobalTimeExpression.java | 10 +-
.../expression/impl/SingleSeriesExpression.java | 10 +-
.../tsfile/read/filter/basic/BinaryFilter.java | 3 +
.../iotdb/tsfile/read/filter/basic/Filter.java | 2 +
.../tsfile/read/filter/basic/UnaryFilter.java | 3 +
.../tsfile/read/filter/operator/AndFilter.java | 7 +-
.../iotdb/tsfile/read/filter/operator/Eq.java | 6 +
.../iotdb/tsfile/read/filter/operator/Gt.java | 6 +
.../iotdb/tsfile/read/filter/operator/GtEq.java | 6 +
.../iotdb/tsfile/read/filter/operator/Lt.java | 6 +
.../iotdb/tsfile/read/filter/operator/LtEq.java | 6 +
.../iotdb/tsfile/read/filter/operator/NotEq.java | 6 +
.../tsfile/read/filter/operator/NotFilter.java | 5 +
.../tsfile/read/filter/operator/OrFilter.java | 5 +
.../query/dataset/DataSetWithTimeGenerator.java | 32 --
.../tsfile/read/query/dataset/QueryDataSet.java | 36 ++
.../tsfile/read/query/executor/TsFileExecutor.java | 100 ++--
.../java/org/apache/iotdb/tsfile/utils/Pair.java | 5 +-
.../iotdb/tsfile/write/chunk/ChunkBuffer.java | 7 +-
.../iotdb/tsfile/write/chunk/ChunkWriterImpl.java | 7 +-
.../iotdb/tsfile/read/ReadInPartitionTest.java | 228 +++++---
.../iotdb/tsfile/read/common/TimeRangeTest.java | 42 +-
.../controller/MetadataQuerierByFileImplTest.java | 156 ++++--
175 files changed, 2970 insertions(+), 2019 deletions(-)
copy iotdb/src/main/java/org/apache/iotdb/db/exception/{SysCheckException.java => ConsistencyException.java} (79%)
create mode 100644 iotdb/src/main/java/org/apache/iotdb/db/qp/executor/IQueryProcessExecutor.java
rename iotdb/src/main/java/org/apache/iotdb/db/{writelog => qp/physical}/transfer/Codec.java (91%)
rename iotdb/src/main/java/org/apache/iotdb/db/{writelog => qp/physical}/transfer/CodecInstances.java (99%)
rename iotdb/src/main/java/org/apache/iotdb/db/{writelog => qp/physical}/transfer/PhysicalPlanCodec.java (78%)
rename iotdb/src/main/java/org/apache/iotdb/db/{writelog => qp/physical}/transfer/PhysicalPlanLogTransfer.java (98%)
rename iotdb/src/main/java/org/apache/iotdb/db/{writelog => qp/physical}/transfer/SystemLogOperator.java (96%)
copy tsfile/src/main/java/org/apache/iotdb/tsfile/read/query/dataset/QueryDataSet.java => iotdb/src/main/java/org/apache/iotdb/db/query/dataset/AuthDataSet.java (56%)
rename iotdb/src/main/java/org/apache/iotdb/db/query/{executor => dataset}/groupby/GroupByEngineDataSet.java (96%)
rename iotdb/src/main/java/org/apache/iotdb/db/query/{executor/groupby/GroupByWithOnlyTimeFilterDataSetDataSet.java => dataset/groupby/GroupByWithOnlyTimeFilterDataSet.java} (97%)
rename iotdb/src/main/java/org/apache/iotdb/db/query/{executor/groupby/GroupByWithValueFilterDataSetDataSet.java => dataset/groupby/GroupByWithValueFilterDataSet.java} (94%)
create mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/executor/AbstractExecutorWithoutTimeGenerator.java
create mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/executor/IEngineQueryRouter.java
rename iotdb/src/main/java/org/apache/iotdb/db/query/{aggregation => factory}/AggreFuncFactory.java (96%)
copy iotdb/src/main/java/org/apache/iotdb/db/query/timegenerator/{EngineNodeConstructor.java => AbstractNodeConstructor.java} (70%)
create mode 100644 iotdb/src/test/java/org/apache/iotdb/db/metadata/MetadataTest.java
rename iotdb/src/test/java/org/apache/iotdb/db/{writelog => qp/physical}/transfer/PhysicalPlanLogTransferTest.java (99%)
[incubator-iotdb] 01/01: Merge branch 'master' into
refactor_overflow
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch refactor_overflow
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 458c20f42f25953990d486203189e11618065431
Merge: b34cb2a 6f40c05
Author: 江天 <jt...@163.com>
AuthorDate: Thu Jun 6 18:24:06 2019 +0800
Merge branch 'master' into refactor_overflow
# Conflicts:
# iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
# iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessor.java
# iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeManager.java
# iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java
# iotdb/src/main/java/org/apache/iotdb/db/engine/pool/MergeManager.java
# iotdb/src/main/java/org/apache/iotdb/db/exception/BufferWriteProcessorException.java
# iotdb/src/main/java/org/apache/iotdb/db/exception/FileNodeProcessorException.java
# iotdb/src/main/java/org/apache/iotdb/db/exception/StorageGroupProcessorException.java
# iotdb/src/main/java/org/apache/iotdb/db/qp/executor/QueryProcessExecutor.java
# iotdb/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java
# iotdb/src/main/java/org/apache/iotdb/db/query/executor/EngineExecutorWithoutTimeGenerator.java
# iotdb/src/main/java/org/apache/iotdb/db/query/timegenerator/EngineNodeConstructor.java
# iotdb/src/main/java/org/apache/iotdb/db/service/IoTDB.java
# iotdb/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
# iotdb/src/main/java/org/apache/iotdb/db/utils/RecordUtils.java
# iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/ExclusiveLogRecoverPerformer.java
# iotdb/src/test/java/org/apache/iotdb/db/qp/utils/MemIntQpExecutor.java
# iotdb/src/test/java/org/apache/iotdb/db/writelog/PerformanceTest.java
README.md | 9 +
docs/Documentation/QuickStart.md | 10 +
.../UserGuideV0.7.0/4-Deployment and Management.md | 71 ++-
.../UserGuideV0.7.0/5-SQL Documentation.md | 17 +-
.../iotdb/tsfile/hadoop/TSFRecordWriter.java | 11 +-
.../iotdb/tsfile/hadoop/example/TsFileHelper.java | 6 +-
.../cn/edu/thu/tsfile/hadoop/TsFileTestHelper.java | 6 +-
iotdb-cli/cli/bin/export-csv.bat | 10 +-
iotdb-cli/cli/bin/export-csv.sh | 8 +-
iotdb-cli/cli/bin/import-csv.bat | 10 +-
iotdb-cli/cli/bin/import-csv.sh | 8 +-
iotdb-cli/cli/bin/start-client.bat | 8 +-
iotdb-cli/cli/bin/start-client.sh | 6 +-
.../apache/iotdb/cli/client/AbstractClient.java | 4 +
.../java/org/apache/iotdb/cli/tool/ImportCsv.java | 4 +-
iotdb/iotdb/conf/iotdb-engine.properties | 68 ++-
iotdb/iotdb/conf/iotdb-sync-client.properties | 4 +-
.../iotdb/db/auth/authorizer/BasicAuthorizer.java | 3 +-
.../apache/iotdb/db/auth/entity/PathPrivilege.java | 1 -
.../db/concurrent/IoTDBThreadPoolFactory.java | 8 +-
.../org/apache/iotdb/db/concurrent/ThreadName.java | 2 +-
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 185 +++----
.../org/apache/iotdb/db/conf/IoTDBConstant.java | 10 +-
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 17 +-
.../java/org/apache/iotdb/db/engine/Processor.java | 2 +-
.../apache/iotdb/db/engine/pool/FlushManager.java | 6 +-
.../apache/iotdb/db/engine/pool/MergeManager.java | 2 -
.../db/engine/sgmanager/StorageGroupProcessor.java | 2 +-
.../db/engine/tsfiledata/TsFileProcessor.java | 182 +++----
...matException.java => ConsistencyException.java} | 21 +-
.../db/exception/qp/IllegalASTFormatException.java | 7 +
.../db/exception/qp/LogicalOperatorException.java | 7 +
.../db/exception/qp/LogicalOptimizeException.java | 7 +
.../db/exception/qp/QueryProcessorException.java | 4 +
.../java/org/apache/iotdb/db/metadata/MGraph.java | 12 +-
.../org/apache/iotdb/db/metadata/MManager.java | 13 -
.../java/org/apache/iotdb/db/metadata/MTree.java | 32 --
.../org/apache/iotdb/db/metadata/Metadata.java | 106 +++-
.../org/apache/iotdb/db/qp/QueryProcessor.java | 13 +-
.../db/qp/executor/IQueryProcessExecutor.java | 134 +++++
.../iotdb/db/qp/executor/OverflowQPExecutor.java | 266 ++++++----
.../iotdb/db/qp/executor/QueryProcessExecutor.java | 56 +-
.../org/apache/iotdb/db/qp/logical/Operator.java | 2 +-
.../db/qp/logical/crud/BasicFunctionOperator.java | 5 +-
.../iotdb/db/qp/logical/crud/FilterOperator.java | 5 +-
.../apache/iotdb/db/qp/physical/PhysicalPlan.java | 8 +-
.../iotdb/db/qp/physical/crud/AggregationPlan.java | 1 +
.../iotdb/db/qp/physical/crud/DeletePlan.java | 1 +
.../iotdb/db/qp/physical/crud/FillQueryPlan.java | 1 +
.../iotdb/db/qp/physical/crud/GroupByPlan.java | 1 +
.../iotdb/db/qp/physical/crud/InsertPlan.java | 6 +
.../iotdb/db/qp/physical/crud/QueryPlan.java | 5 +-
.../iotdb/db/qp/physical/crud/UpdatePlan.java | 1 +
.../iotdb/db/qp/physical/sys/AuthorPlan.java | 7 +
.../iotdb/db/qp/physical/sys/LoadDataPlan.java | 1 +
.../iotdb/db/qp/physical/sys/MetadataPlan.java | 1 +
.../iotdb/db/qp/physical/sys/PropertyPlan.java | 1 +
.../{writelog => qp/physical}/transfer/Codec.java | 3 +-
.../physical}/transfer/CodecInstances.java | 5 +-
.../physical}/transfer/PhysicalPlanCodec.java | 14 +-
.../transfer/PhysicalPlanLogTransfer.java | 2 +-
.../physical}/transfer/SystemLogOperator.java | 2 +-
.../iotdb/db/qp/strategy/PhysicalGenerator.java | 8 +-
.../qp/strategy/optimizer/ConcatPathOptimizer.java | 14 +-
.../db/query/aggregation/AggreResultData.java | 2 +-
.../db/query/aggregation/AggregateFunction.java | 1 -
.../db/query/aggregation/impl/CountAggrFunc.java | 2 +-
.../db/query/aggregation/impl/FirstAggrFunc.java | 2 +-
.../db/query/aggregation/impl/LastAggrFunc.java | 2 +-
.../db/query/aggregation/impl/MaxTimeAggrFunc.java | 2 +-
.../query/aggregation/impl/MaxValueAggrFunc.java | 2 +-
.../db/query/aggregation/impl/MeanAggrFunc.java | 13 +-
.../db/query/aggregation/impl/MinTimeAggrFunc.java | 2 +-
.../query/aggregation/impl/MinValueAggrFunc.java | 2 +-
.../db/query/aggregation/impl/SumAggrFunc.java | 10 +
.../db/query/control/QueryResourceManager.java | 21 +-
.../query/dataset/AggreResultDataPointReader.java | 2 +-
.../apache/iotdb/db/query/dataset/AuthDataSet.java | 45 +-
.../dataset/EngineDataSetWithTimeGenerator.java | 35 +-
.../dataset/EngineDataSetWithoutTimeGenerator.java | 4 +
.../groupby/GroupByEngineDataSet.java | 11 +-
.../groupby/GroupByWithOnlyTimeFilterDataSet.java} | 8 +-
.../groupby/GroupByWithValueFilterDataSet.java} | 8 +-
.../AbstractExecutorWithoutTimeGenerator.java | 86 +++
.../db/query/executor/AggregateEngineExecutor.java | 2 +-
.../EngineExecutorWithoutTimeGenerator.java | 46 +-
.../iotdb/db/query/executor/EngineQueryRouter.java | 44 +-
.../db/query/executor/IEngineQueryRouter.java | 78 +++
.../{aggregation => factory}/AggreFuncFactory.java | 3 +-
...nstructor.java => AbstractNodeConstructor.java} | 62 +--
.../query/timegenerator/EngineNodeConstructor.java | 65 +--
.../query/timegenerator/EngineTimeGenerator.java | 1 -
.../apache/iotdb/db/service/CloseMergeService.java | 6 +-
.../java/org/apache/iotdb/db/service/IoTDB.java | 14 +-
.../org/apache/iotdb/db/service/JDBCService.java | 17 +-
.../iotdb/db/service/JDBCServiceEventHandler.java | 6 +-
.../org/apache/iotdb/db/service/JMXService.java | 5 +-
.../java/org/apache/iotdb/db/service/Monitor.java | 3 +-
.../apache/iotdb/db/service/RegisterManager.java | 2 +-
.../org/apache/iotdb/db/service/TSServiceImpl.java | 591 +++++++++++----------
.../org/apache/iotdb/db/sql/parse/ParseDriver.java | 2 +-
.../org/apache/iotdb/db/sync/conf/Constans.java | 10 +
.../iotdb/db/sync/conf/SyncSenderConfig.java | 24 +-
.../iotdb/db/sync/conf/SyncSenderDescriptor.java | 7 +-
.../iotdb/db/sync/receiver/SyncServiceImpl.java | 2 +-
.../iotdb/db/sync/receiver/SyncServiceManager.java | 5 +-
.../apache/iotdb/db/sync/sender/SyncSender.java | 10 +
.../iotdb/db/sync/sender/SyncSenderImpl.java | 106 ++--
.../org/apache/iotdb/db/utils/RecordUtils.java | 12 +-
.../apache/iotdb/db/writelog/io/RAFLogReader.java | 2 +-
.../writelog/manager/MultiFileLogNodeManager.java | 9 +-
.../db/writelog/node/ExclusiveWriteLogNode.java | 4 +-
.../recover/ExclusiveLogRecoverPerformer.java | 15 +-
.../db/concurrent/IoTDBThreadPoolFactoryTest.java | 2 +-
.../db/engine/tsfiledata/TsFileProcessorTest.java | 2 +-
.../iotdb/db/integration/IoTDBAggregationIT.java | 86 ++-
.../iotdb/db/integration/IoTDBAuthorizationIT.java | 416 ++++++++-------
.../iotdb/db/integration/IoTDBSeriesReaderIT.java | 2 +-
.../iotdb/db/metadata/MManagerBasicTest.java | 29 +-
.../org/apache/iotdb/db/metadata/MTreeTest.java | 28 +-
.../org/apache/iotdb/db/metadata/MetadataTest.java | 93 ++++
.../org/apache/iotdb/db/qp/QueryProcessorTest.java | 88 ++-
.../transfer/PhysicalPlanLogTransferTest.java | 2 +-
.../org/apache/iotdb/db/qp/plan/QPUpdateTest.java | 2 +-
.../apache/iotdb/db/qp/utils/MemIntQpExecutor.java | 13 +-
.../EngineDataSetWithTimeGeneratorTest.java | 2 +-
.../query/executor/GroupByEngineDataSetTest.java | 14 +-
.../org/apache/iotdb/db/tools/WalCheckerTest.java | 2 +-
.../apache/iotdb/db/utils/EnvironmentUtils.java | 2 -
.../apache/iotdb/db/writelog/PerformanceTest.java | 2 +-
.../apache/iotdb/db/writelog/WriteLogNodeTest.java | 2 +-
.../iotdb/db/writelog/io/LogWriterReaderTest.java | 3 +-
jdbc/README.md | 4 +-
.../org/apache/iotdb/jdbc/IoTDBConnection.java | 6 +-
.../org/apache/iotdb/jdbc/IoTDBQueryResultSet.java | 9 +
.../java/org/apache/iotdb/jdbc/IoTDBStatement.java | 15 +-
service-rpc/src/main/thrift/rpc.thrift | 1 +
.../org/apache/iotdb/tsfile/DefaultSource.scala | 9 +-
.../tsfile/common/constant/QueryConstant.java | 6 -
.../tsfile/exception/write/PageException.java | 8 +
.../exception/write/WriteProcessException.java | 16 +-
.../apache/iotdb/tsfile/read/ReadOnlyTsFile.java | 17 +-
.../apache/iotdb/tsfile/read/common/BatchData.java | 4 +-
.../org/apache/iotdb/tsfile/read/common/Field.java | 22 +
.../org/apache/iotdb/tsfile/read/common/Path.java | 4 +-
.../apache/iotdb/tsfile/read/common/TimeRange.java | 225 ++++----
.../tsfile/read/controller/MetadataQuerier.java | 39 +-
.../read/controller/MetadataQuerierByFileImpl.java | 189 +++----
.../tsfile/read/expression/ExpressionType.java | 30 +-
.../tsfile/read/expression/IBinaryExpression.java | 4 +
.../iotdb/tsfile/read/expression/IExpression.java | 4 +-
.../read/expression/impl/BinaryExpression.java | 38 +-
.../read/expression/impl/GlobalTimeExpression.java | 10 +-
.../expression/impl/SingleSeriesExpression.java | 10 +-
.../tsfile/read/filter/basic/BinaryFilter.java | 3 +
.../iotdb/tsfile/read/filter/basic/Filter.java | 2 +
.../tsfile/read/filter/basic/UnaryFilter.java | 3 +
.../tsfile/read/filter/operator/AndFilter.java | 7 +-
.../iotdb/tsfile/read/filter/operator/Eq.java | 6 +
.../iotdb/tsfile/read/filter/operator/Gt.java | 6 +
.../iotdb/tsfile/read/filter/operator/GtEq.java | 6 +
.../iotdb/tsfile/read/filter/operator/Lt.java | 6 +
.../iotdb/tsfile/read/filter/operator/LtEq.java | 6 +
.../iotdb/tsfile/read/filter/operator/NotEq.java | 6 +
.../tsfile/read/filter/operator/NotFilter.java | 5 +
.../tsfile/read/filter/operator/OrFilter.java | 5 +
.../query/dataset/DataSetWithTimeGenerator.java | 32 --
.../tsfile/read/query/dataset/QueryDataSet.java | 36 ++
.../tsfile/read/query/executor/TsFileExecutor.java | 100 ++--
.../java/org/apache/iotdb/tsfile/utils/Pair.java | 5 +-
.../iotdb/tsfile/write/chunk/ChunkBuffer.java | 7 +-
.../iotdb/tsfile/write/chunk/ChunkWriterImpl.java | 7 +-
.../iotdb/tsfile/read/ReadInPartitionTest.java | 228 +++++---
.../iotdb/tsfile/read/common/TimeRangeTest.java | 42 +-
.../controller/MetadataQuerierByFileImplTest.java | 156 ++++--
175 files changed, 2978 insertions(+), 2021 deletions(-)
diff --cc iotdb/iotdb/conf/iotdb-engine.properties
index 5031068,848696c..ff27ab1
--- a/iotdb/iotdb/conf/iotdb-engine.properties
+++ b/iotdb/iotdb/conf/iotdb-engine.properties
@@@ -228,12 -237,9 +250,14 @@@ sync_server_port=555
# The default is to allow all IP to sync
IP_white_list=0.0.0.0/0
- # Choose a sync strategy of loading historical data:
- #1. It's more likely to update historical data, please choose "true".
- #2. It's more likely not to update historical data or you don't know exactly, please choose "false".
+ # The processing strategy chosen by the sync server when merging the sync data.
+ # 1. If the sync data accounts for more than 50% of the update of the historical data (compared with the latest timestamp of the local storage group data),then it is recommended to select strategy 1.
+ # Setting the parameter to true, which has a greater impact on the write performance of the IoTDB system and occupies less CPU of the machine.
+ # 2. If the sync data accounts for less than 50% of the update of the historical data (compared with the latest timestamp of the local storage group data),then it is recommended to select strategy 2.
+ # Setting the parameter to false, which has little impact on the write performance of IoTDB system and takes up a large amount of CPU power.
update_historical_data_possibility=false
+
+# the memory limit (Bytes) for each query;
+# by default, it is 5MB
+# if there are too many overflow files, you may need to enlarge it.
+query_memory_limit = 5242880
diff --cc iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 8ab170c,4bf5ac8..2cb2adf
--- a/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@@ -30,13 -31,12 +31,13 @@@ public class IoTDBConfig
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBConfig.class);
public static final String CONFIG_NAME = "iotdb-engine.properties";
-- public static final String DEFAULT_DATA_DIR = "data";
-- public static final String DEFAULT_SYS_DIR = "system";
-- public static final String DEFAULT_TSFILE_DIR = "settled";
- public static final String DEFAULT_OVERFLOW_DIR = "overflow";
-- public static final String MULT_DIR_STRATEGY_PREFIX =
++ private static final String DEFAULT_DATA_DIR = "data";
++ private static final String DEFAULT_SYS_DIR = "system";
++ static final String DEFAULT_TSFILE_DIR = "settled";
++ static final String DEFAULT_OVERFLOW_DIR = "overflow";
++ private static final String MULT_DIR_STRATEGY_PREFIX =
"org.apache.iotdb.db.conf.directories.strategy.";
-- public static final String DEFAULT_MULT_DIR_STRATEGY = "MaxDiskUsableSpaceFirstStrategy";
++ private static final String DEFAULT_MULT_DIR_STRATEGY = "MaxDiskUsableSpaceFirstStrategy";
private String rpcAddress = "0.0.0.0";
/**
@@@ -109,11 -109,11 +110,6 @@@
private String metadataDir = "schema";
/**
-- * Data directory of derby data.
-- */
-- private String derbyHome = "derby";
--
-- /**
* Data directory of Write ahead log folder.
*/
private String walFolder = "wal";
@@@ -124,11 -124,11 +120,6 @@@
private String indexFileDir = "index";
/**
-- * Temporary directory for temporary files of read (External Sort). TODO: unused field
-- */
-- private String readTmpFileDir = "readTmp";
--
-- /**
* The maximum concurrent thread number for merging overflow. When the value <=0 or > CPU core
* number, use the CPU core number.
*/
@@@ -278,12 -278,10 +269,17 @@@
private long cacheFileReaderClearPeriod = 100000;
/**
+ * the memory limit (Bytes) for each query;
+ * by default, it is 5MB
+ * if there are too many overflow files, you may need to enlarge it.
+ */
+ private long queryMemoryLimit = 5242880;
+
++ /**
+ * Replace implementation class of JDBC service
+ */
+ private String rpcImplClassName = TSServiceImpl.class.getName();
+
public IoTDBConfig() {
// empty constructor
}
@@@ -292,59 -290,49 +288,59 @@@
return zoneID;
}
-- public String getZoneIDString() {
-- return getZoneID().toString();
-- }
--
-- public void updatePath() {
++ void updatePath() {
confirmMultiDirStrategy();
preUpdatePath();
// update the paths of subdirectories in the dataDir
- if (getDataDir().length() > 0 && !getDataDir().endsWith(File.separator)) {
- setDataDir(getDataDir() + File.separatorChar);
- }
- setOverflowDataDir(getDataDir() + getOverflowDataDir());
++ updateDataDir();
+
- if (getBufferWriteDirs() == null || getBufferWriteDirs().length == 0) {
- setBufferWriteDirs(new String[]{DEFAULT_TSFILE_DIR});
- }
- for (int i = 0; i < getBufferWriteDirs().length; i++) {
- if (new File(getBufferWriteDirs()[i]).isAbsolute()) {
- continue;
- }
-
- getBufferWriteDirs()[i] = getDataDir() + getBufferWriteDirs()[i];
- }
+
+ // update the paths of subdirectories in the sysDir
+ if (getSysDir().length() > 0 && !getSysDir().endsWith(File.separator)) {
+ setSysDir(getSysDir() + File.separatorChar);
+ }
- setFileNodeDir(getSysDir() + getFileNodeDir());
++ setStorageGroupDir(getSysDir() + getStorageGroupDir());
+ setMetadataDir(getSysDir() + getMetadataDir());
+
+ // update the paths of subdirectories in the walDir
+ if (getWalDir().length() > 0 && !getWalDir().endsWith(File.separator)) {
+ setWalDir(getWalDir() + File.separatorChar);
+ }
+ setWalFolder(getWalDir() + getWalFolder());
+
- setDerbyHome(getSysDir() + getDerbyHome());
+ setIndexFileDir(getDataDir() + getIndexFileDir());
+ }
+
++ private void updateDataDir() {
+ if (getDataDir().length() > 0 && !getDataDir().endsWith(File.separator)) {
+ setDataDir(getDataDir() + File.separatorChar);
+ }
+
+ if (getOverflowDataDirs() == null || getOverflowDataDirs().length == 0) {
+ setOverflowDataDirs(new String[]{DEFAULT_OVERFLOW_DIR});
+ }
+ for (int i = 0; i < getOverflowDataDirs().length; i++) {
+ if (new File(getOverflowDataDirs()[i]).isAbsolute()) {
+ continue;
+ }
+
+ getOverflowDataDirs()[i] = getDataDir() + getOverflowDataDirs()[i];
+ }
+
+ if (getBufferWriteDirs() == null || getBufferWriteDirs().length == 0) {
+ setBufferWriteDirs(new String[]{DEFAULT_TSFILE_DIR});
+ }
+ for (int i = 0; i < getBufferWriteDirs().length; i++) {
+ if (new File(getBufferWriteDirs()[i]).isAbsolute()) {
+ continue;
+ }
+
+ getBufferWriteDirs()[i] = getDataDir() + getBufferWriteDirs()[i];
+ }
-
- // update the paths of subdirectories in the sysDir
- if (getSysDir().length() > 0 && !getSysDir().endsWith(File.separator)) {
- setSysDir(getSysDir() + File.separatorChar);
- }
- setStorageGroupDir(getSysDir() + getStorageGroupDir());
- setMetadataDir(getSysDir() + getMetadataDir());
-
- // update the paths of subdirectories in the walDir
- if (getWalDir().length() > 0 && !getWalDir().endsWith(File.separator)) {
- setWalDir(getWalDir() + File.separatorChar);
- }
- setWalFolder(getWalDir() + getWalFolder());
-
- setDerbyHome(getSysDir() + getDerbyHome());
- setIndexFileDir(getDataDir() + getIndexFileDir());
+ }
+
/*
* First, if dataDir is null, dataDir will be assigned the default
* value(i.e.,"data"+File.separatorChar+"data".
@@@ -438,7 -426,7 +434,7 @@@
return rpcAddress;
}
-- public void setRpcAddress(String rpcAddress) {
++ void setRpcAddress(String rpcAddress) {
this.rpcAddress = rpcAddress;
}
@@@ -446,7 -434,7 +442,7 @@@
return rpcPort;
}
-- public void setRpcPort(int rpcPort) {
++ void setRpcPort(int rpcPort) {
this.rpcPort = rpcPort;
}
@@@ -486,43 -474,43 +482,43 @@@
return dataDir;
}
-- public void setDataDir(String dataDir) {
++ void setDataDir(String dataDir) {
this.dataDir = dataDir;
}
-- public String getSysDir() {
++ String getSysDir() {
return sysDir;
}
-- public void setSysDir(String sysDir) {
++ void setSysDir(String sysDir) {
this.sysDir = sysDir;
}
-- public String getWalDir() {
++ String getWalDir() {
return walDir;
}
-- public void setWalDir(String walDir) {
++ void setWalDir(String walDir) {
this.walDir = walDir;
}
- public String getOverflowDataDir() {
- return overflowDataDir;
+ public String[] getOverflowDataDirs() {
+ return overflowDataDirs;
}
- public void setOverflowDataDirs(String[] overflowDataDirs) {
- public void setOverflowDataDir(String overflowDataDir) {
- this.overflowDataDir = overflowDataDir;
++ void setOverflowDataDirs(String[] overflowDataDirs) {
+ this.overflowDataDirs = overflowDataDirs;
}
- public String getFileNodeDir() {
- return fileNodeDir;
+ public String getStorageGroupDir() {
+ return storageGroupDir;
}
- public void setStorageGroupDir(String storageGroupDir) {
- public void setFileNodeDir(String fileNodeDir) {
- this.fileNodeDir = fileNodeDir;
++ private void setStorageGroupDir(String storageGroupDir) {
+ this.storageGroupDir = storageGroupDir;
}
-- public void setBufferWriteDirs(String[] bufferWriteDirs) {
++ void setBufferWriteDirs(String[] bufferWriteDirs) {
this.bufferWriteDirs = bufferWriteDirs;
}
@@@ -530,7 -518,7 +526,7 @@@
return multDirStrategyClassName;
}
-- public void setMultDirStrategyClassName(String multDirStrategyClassName) {
++ void setMultDirStrategyClassName(String multDirStrategyClassName) {
this.multDirStrategyClassName = multDirStrategyClassName;
}
@@@ -538,23 -526,23 +534,15 @@@
return metadataDir;
}
-- public void setMetadataDir(String metadataDir) {
++ private void setMetadataDir(String metadataDir) {
this.metadataDir = metadataDir;
}
-- public String getDerbyHome() {
-- return derbyHome;
-- }
--
-- public void setDerbyHome(String derbyHome) {
-- this.derbyHome = derbyHome;
-- }
--
public String getWalFolder() {
return walFolder;
}
-- public void setWalFolder(String walFolder) {
++ private void setWalFolder(String walFolder) {
this.walFolder = walFolder;
}
@@@ -562,31 -550,31 +550,23 @@@
return indexFileDir;
}
-- public void setIndexFileDir(String indexFileDir) {
++ private void setIndexFileDir(String indexFileDir) {
this.indexFileDir = indexFileDir;
}
-- public String getReadTmpFileDir() {
-- return readTmpFileDir;
-- }
--
-- public void setReadTmpFileDir(String readTmpFileDir) {
-- this.readTmpFileDir = readTmpFileDir;
-- }
--
public int getMergeConcurrentThreads() {
return mergeConcurrentThreads;
}
-- public void setMergeConcurrentThreads(int mergeConcurrentThreads) {
++ void setMergeConcurrentThreads(int mergeConcurrentThreads) {
this.mergeConcurrentThreads = mergeConcurrentThreads;
}
-- public int getMaxOpenFolder() {
++ int getMaxOpenFolder() {
return maxOpenFolder;
}
-- public void setMaxOpenFolder(int maxOpenFolder) {
++ void setMaxOpenFolder(int maxOpenFolder) {
this.maxOpenFolder = maxOpenFolder;
}
@@@ -594,7 -582,7 +574,7 @@@
return fetchSize;
}
-- public void setFetchSize(int fetchSize) {
++ void setFetchSize(int fetchSize) {
this.fetchSize = fetchSize;
}
@@@ -602,15 -590,15 +582,11 @@@
return writeInstanceThreshold;
}
-- public void setWriteInstanceThreshold(int writeInstanceThreshold) {
-- this.writeInstanceThreshold = writeInstanceThreshold;
-- }
--
public long getPeriodTimeForFlush() {
return periodTimeForFlush;
}
-- public void setPeriodTimeForFlush(long periodTimeForFlush) {
++ void setPeriodTimeForFlush(long periodTimeForFlush) {
this.periodTimeForFlush = periodTimeForFlush;
}
@@@ -618,7 -606,7 +594,7 @@@
return periodTimeForMerge;
}
-- public void setPeriodTimeForMerge(long periodTimeForMerge) {
++ void setPeriodTimeForMerge(long periodTimeForMerge) {
this.periodTimeForMerge = periodTimeForMerge;
}
@@@ -626,7 -614,7 +602,7 @@@
return enableTimingCloseAndMerge;
}
-- public void setEnableTimingCloseAndMerge(boolean enableTimingCloseAndMerge) {
++ void setEnableTimingCloseAndMerge(boolean enableTimingCloseAndMerge) {
this.enableTimingCloseAndMerge = enableTimingCloseAndMerge;
}
@@@ -634,11 -622,11 +610,11 @@@
return concurrentFlushThread;
}
-- public void setConcurrentFlushThread(int concurrentFlushThread) {
++ void setConcurrentFlushThread(int concurrentFlushThread) {
this.concurrentFlushThread = concurrentFlushThread;
}
-- public void setZoneID(ZoneId zoneID) {
++ void setZoneID(ZoneId zoneID) {
this.zoneID = zoneID;
}
@@@ -662,7 -650,7 +638,7 @@@
return memMonitorInterval;
}
-- public void setMemMonitorInterval(long memMonitorInterval) {
++ void setMemMonitorInterval(long memMonitorInterval) {
this.memMonitorInterval = memMonitorInterval;
}
@@@ -670,11 -658,11 +646,11 @@@
return memControllerType;
}
-- public void setMemControllerType(int memControllerType) {
++ void setMemControllerType(int memControllerType) {
this.memControllerType = memControllerType;
}
-- public long getBufferwriteMetaSizeThreshold() {
++ long getBufferwriteMetaSizeThreshold() {
return bufferwriteMetaSizeThreshold;
}
@@@ -694,7 -682,7 +670,7 @@@
return overflowMetaSizeThreshold;
}
-- public void setOverflowMetaSizeThreshold(long overflowMetaSizeThreshold) {
++ void setOverflowMetaSizeThreshold(long overflowMetaSizeThreshold) {
this.overflowMetaSizeThreshold = overflowMetaSizeThreshold;
}
@@@ -718,7 -706,7 +694,7 @@@
return enableSmallFlush;
}
-- public void setEnableSmallFlush(boolean enableSmallFlush) {
++ void setEnableSmallFlush(boolean enableSmallFlush) {
this.enableSmallFlush = enableSmallFlush;
}
@@@ -726,7 -714,7 +702,7 @@@
return smallFlushInterval;
}
-- public void setSmallFlushInterval(long smallFlushInterval) {
++ void setSmallFlushInterval(long smallFlushInterval) {
this.smallFlushInterval = smallFlushInterval;
}
@@@ -750,7 -738,7 +726,7 @@@
return statMonitorDetectFreqSec;
}
-- public void setStatMonitorDetectFreqSec(int statMonitorDetectFreqSec) {
++ void setStatMonitorDetectFreqSec(int statMonitorDetectFreqSec) {
this.statMonitorDetectFreqSec = statMonitorDetectFreqSec;
}
@@@ -758,15 -746,15 +734,15 @@@
return statMonitorRetainIntervalSec;
}
-- public void setStatMonitorRetainIntervalSec(int statMonitorRetainIntervalSec) {
++ void setStatMonitorRetainIntervalSec(int statMonitorRetainIntervalSec) {
this.statMonitorRetainIntervalSec = statMonitorRetainIntervalSec;
}
-- public int getExternalSortThreshold() {
++ int getExternalSortThreshold() {
return externalSortThreshold;
}
-- public void setExternalSortThreshold(int externalSortThreshold) {
++ void setExternalSortThreshold(int externalSortThreshold) {
this.externalSortThreshold = externalSortThreshold;
}
@@@ -774,7 -762,7 +750,7 @@@
return mManagerCacheSize;
}
-- public void setmManagerCacheSize(int mManagerCacheSize) {
++ void setmManagerCacheSize(int mManagerCacheSize) {
this.mManagerCacheSize = mManagerCacheSize;
}
@@@ -782,7 -770,7 +758,7 @@@
return maxLogEntrySize;
}
-- public void setMaxLogEntrySize(int maxLogEntrySize) {
++ void setMaxLogEntrySize(int maxLogEntrySize) {
this.maxLogEntrySize = maxLogEntrySize;
}
@@@ -790,7 -778,7 +766,7 @@@
return isSyncEnable;
}
-- public void setSyncEnable(boolean syncEnable) {
++ void setSyncEnable(boolean syncEnable) {
isSyncEnable = syncEnable;
}
@@@ -798,7 -786,7 +774,7 @@@
return syncServerPort;
}
-- public void setSyncServerPort(int syncServerPort) {
++ void setSyncServerPort(int syncServerPort) {
this.syncServerPort = syncServerPort;
}
@@@ -806,16 -794,16 +782,16 @@@
return languageVersion;
}
-- public void setLanguageVersion(String languageVersion) {
++ void setLanguageVersion(String languageVersion) {
this.languageVersion = languageVersion;
}
- public boolean isUpdate_historical_data_possibility() {
- return update_historical_data_possibility;
+ public boolean isUpdateHistoricalDataPossibility() {
+ return updateHistoricalDataPossibility;
}
- public void setUpdate_historical_data_possibility(boolean update_historical_data_possibility) {
- this.update_historical_data_possibility = update_historical_data_possibility;
- public void setUpdateHistoricalDataPossibility(boolean updateHistoricalDataPossibility) {
++ void setUpdateHistoricalDataPossibility(boolean updateHistoricalDataPossibility) {
+ this.updateHistoricalDataPossibility = updateHistoricalDataPossibility;
}
public String getIpWhiteList() {
@@@ -834,11 -822,11 +810,16 @@@
this.cacheFileReaderClearPeriod = cacheFileReaderClearPeriod;
}
- public long getQueryMemoryLimit() {
++ long getQueryMemoryLimit() {
+ return queryMemoryLimit;
+ }
+
- public void setQueryMemoryLimit(long queryMemoryLimit) {
++ void setQueryMemoryLimit(long queryMemoryLimit) {
+ this.queryMemoryLimit = queryMemoryLimit;
+ }
++
+ public String getRpcImplClassName() {
+ return rpcImplClassName;
+ }
+
- public void setRpcImplClassName(String rpcImplClassName) {
- this.rpcImplClassName = rpcImplClassName;
- }
}
diff --cc iotdb/src/main/java/org/apache/iotdb/db/engine/Processor.java
index 059b2bc,5c515ff..dcec454
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/Processor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/Processor.java
@@@ -173,7 -176,7 +173,7 @@@ public abstract class Processor
* @return a future that returns true if successfully, otherwise false.
* @throws IOException
*/
-- public abstract Future<Boolean> flush() throws IOException;
++ public abstract Future<Boolean> flush() throws IOException, TsFileProcessorException;
/**
* Close the processor.<br>
diff --cc iotdb/src/main/java/org/apache/iotdb/db/engine/pool/MergeManager.java
index ff795c6,44874bb..d57ddef
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/pool/MergeManager.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/pool/MergeManager.java
@@@ -18,11 -18,10 +18,9 @@@
*/
package org.apache.iotdb.db.engine.pool;
+import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
--import java.util.concurrent.Executors;
import java.util.concurrent.Future;
--import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.db.concurrent.ThreadName;
diff --cc iotdb/src/main/java/org/apache/iotdb/db/engine/sgmanager/StorageGroupProcessor.java
index b5a704e,0000000..13c74a1
mode 100644,000000..100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/sgmanager/StorageGroupProcessor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/sgmanager/StorageGroupProcessor.java
@@@ -1,1066 -1,0 +1,1066 @@@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.engine.sgmanager;
+
+import static java.time.ZonedDateTime.ofInstant;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.ArrayList;
+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.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.iotdb.db.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.directories.Directories;
+import org.apache.iotdb.db.engine.Processor;
+import org.apache.iotdb.db.engine.EngineConstants;
+import org.apache.iotdb.db.engine.merge.MergeTask;
+import org.apache.iotdb.db.engine.modification.Deletion;
+import org.apache.iotdb.db.engine.modification.ModificationFile;
+import org.apache.iotdb.db.engine.overflowdata.OverflowProcessor;
+import org.apache.iotdb.db.engine.pool.MergeManager;
+import org.apache.iotdb.db.engine.datasource.QueryDataSource;
+import org.apache.iotdb.db.engine.datasource.SeriesDataSource;
+import org.apache.iotdb.db.engine.tsfiledata.TsFileProcessor;
+import org.apache.iotdb.db.engine.version.SimpleFileVersionController;
+import org.apache.iotdb.db.engine.version.VersionController;
+import org.apache.iotdb.db.exception.PathErrorException;
+import org.apache.iotdb.db.exception.StorageGroupProcessorException;
+import org.apache.iotdb.db.exception.TsFileProcessorException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.monitor.IStatistic;
+import org.apache.iotdb.db.monitor.MonitorConstants;
+import org.apache.iotdb.db.monitor.MonitorConstants.StorageGroupProcessorStatConstants;
+import org.apache.iotdb.db.monitor.StatMonitor;
+import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
+import org.apache.iotdb.db.qp.physical.crud.UpdatePlan;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.factory.SeriesReaderFactory;
+import org.apache.iotdb.db.query.reader.IReader;
+import org.apache.iotdb.db.utils.ImmediateFuture;
+import org.apache.iotdb.db.utils.MemUtils;
+import org.apache.iotdb.db.utils.TimeValuePair;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.file.footer.ChunkGroupFooter;
+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.read.expression.impl.SingleSeriesExpression;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.write.chunk.ChunkBuffer;
+import org.apache.iotdb.tsfile.write.chunk.ChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.schema.FileSchema;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class StorageGroupProcessor extends Processor implements IStatistic {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(StorageGroupProcessor.class);
+
+ public static final String MERGE_TEMP_SUFFIX = ".merging";
+ private static final String OLD_FILE_RECORD = "MERGE_OLD_FILES";
+
+ private final String statStorageGroupName;
+ private String storageGroupDir;
+
+ private TsFileProcessor tsFileProcessor;
+ private OverflowProcessor overflowProcessor;
+
+ private FileSchema fileSchema;
+ private VersionController versionController;
+ private long lastMergeTime;
+
+ /**
+ * oldMultiPassTokenSet records how many queries are progressing before the last merge starts.
+ */
+ private Set<Integer> oldMultiPassTokenSet = null;
+
+ /**
+ * newMultiPassTokenSet records how many queries are progressing since the last merge starts.
+ */
+ private Set<Integer> newMultiPassTokenSet = new HashSet<>();
+
+ /**
+ * newMultiPassCount assign an Id to each query to track whether this query has ended and its
+ * resource can be released (whether files can be removed).
+ */
+ private AtomicInteger newMultiPassCount = new AtomicInteger(0);
+
+ private boolean isMerging = false;
+
+ /**
+ * This is the modification file of the result of the current merge.
+ */
+ private ModificationFile mergingModification;
+
+ private TsFileIOWriter mergeFileWriter = null;
+ private boolean mergeIsChunkGroupHasData = false;
+ private long mergeStartPos;
+
+ private Map<String, AtomicLong> statisticMap;
+
+ /**
+ * Construct processor using StorageGroup name
+ */
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ StorageGroupProcessor(String processorName)
+ throws TsFileProcessorException {
+ super(processorName);
+
+ fileSchema = constructFileSchema(processorName);
+
+ File systemFolder = new File(IoTDBDescriptor.getInstance().getConfig().getStorageGroupDir(),
+ processorName);
+ if (!systemFolder.exists()) {
+ systemFolder.mkdirs();
+ LOGGER.info("The directory of the storage group processor {} doesn't exist. Create new "
+ + "directory {}",
+ getProcessorName(), systemFolder.getAbsolutePath());
+ }
+ storageGroupDir = systemFolder.getAbsolutePath();
+ try {
+ cleanLastMerge();
+ } catch (IOException e) {
+ throw new TsFileProcessorException(e);
+ }
+
+ //the version controller is shared by tsfile and overflow processor.
+ try {
+ versionController = new SimpleFileVersionController(systemFolder.getAbsolutePath());
+ } catch (IOException e) {
+ throw new TsFileProcessorException(String.format("Unable to create VersionController for %s",
+ processorName), e);
+ }
+ tsFileProcessor = new TsFileProcessor(processorName, versionController, fileSchema);
+ overflowProcessor = new OverflowProcessor(processorName, versionController, fileSchema);
+
+ statStorageGroupName =
+ MonitorConstants.STAT_STORAGE_GROUP_PREFIX + MonitorConstants.MONITOR_PATH_SEPARATOR
+ + MonitorConstants.FILE_NODE_PATH + MonitorConstants.MONITOR_PATH_SEPARATOR
+ + processorName.replaceAll("\\.", "_");
+
+ // RegisterStatService
+ registerStatMetadata();
+ }
+
+ public OperationResult insert(InsertPlan insertPlan) throws TsFileProcessorException {
+ OperationResult result = getTsFileProcessor().insert(insertPlan);
+ if (result == OperationResult.WRITE_REJECT_BY_TIME) {
+ result = getOverflowProcessor().insert(insertPlan);
+ }
+ return result;
+ }
+
+ public void update(UpdatePlan plan) throws TsFileProcessorException {
+ getTsFileProcessor().update(plan);
+ getOverflowProcessor().update(plan);
+ }
+
+ public void delete(String device, String measurementId, long timestamp)
+ throws TsFileProcessorException {
+ deleteInOverflow(device, measurementId, timestamp);
+ deleteInSeqFile(device, measurementId, timestamp);
+ if (mergingModification != null) {
+ try {
+ mergingModification.write(new Deletion(device + IoTDBConstant.PATH_SEPARATOR + measurementId,
+ versionController.nextVersion(), timestamp));
+ } catch (IOException e) {
+ throw new TsFileProcessorException(e);
+ }
+ }
+ }
+
+ /**
+ * query data.
+ */
+ public QueryDataSource query(SingleSeriesExpression seriesExpression, QueryContext context)
+ throws TsFileProcessorException {
+ SeriesDataSource tsfileData;
+ SeriesDataSource overflowData;
+ try {
+ tsfileData = getTsFileProcessor().query(seriesExpression, context);
+ } catch (IOException e) {
+ throw new TsFileProcessorException(e);
+ }
+ try {
+ overflowData = getOverflowProcessor().query(seriesExpression, context);
+ } catch (IOException e) {
+ throw new TsFileProcessorException(e);
+ }
+
+ return new QueryDataSource(tsfileData, overflowData);
+ }
+
+ @Override
+ public boolean canBeClosed() {
+ return tsFileProcessor.canBeClosed() && overflowProcessor.canBeClosed();
+ }
+
+ @Override
- public Future<Boolean> flush() throws IOException {
++ public Future<Boolean> flush() throws IOException, TsFileProcessorException {
+ Future<Boolean> tsFileFuture = tsFileProcessor.flush();
+ Future<Boolean> overflowFuture = overflowProcessor.flush();
+
+ return new StorageGroupFlushFuture(tsFileFuture, overflowFuture);
+ }
+
+ @Override
+ public void close() throws TsFileProcessorException {
+ tsFileProcessor.close();
+ overflowProcessor.close();
+ }
+
+ @Override
+ public long memoryUsage() {
+ return tsFileProcessor.memoryUsage() + overflowProcessor.memoryUsage();
+ }
+
+
+
+ @Override
+ public Map<String, InsertPlan> getAllStatisticsValue() {
+ Long curTime = System.currentTimeMillis();
+ HashMap<String, InsertPlan> tsRecordHashMap = new HashMap<>();
+ Map<String, AtomicLong> hashMap = getStatParamsHashMap();
+ String[] measurements = new String[hashMap.size()];
+ String[] values = new String[hashMap.size()];
+
+ int i = 0;
+ for (Map.Entry<String, AtomicLong> entry : hashMap.entrySet()) {
+ measurements[i] = entry.getKey();
+ values[i++] = entry.getValue().toString();
+ }
+ InsertPlan plan = new InsertPlan(statStorageGroupName, curTime, measurements, values);
+
+ tsRecordHashMap.put(statStorageGroupName, plan);
+ return tsRecordHashMap;
+ }
+
+ @Override
+ public void registerStatMetadata() {
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableStatMonitor()) {
+ String statStorageDeltaName =
+ MonitorConstants.STAT_STORAGE_GROUP_PREFIX + MonitorConstants.MONITOR_PATH_SEPARATOR
+ + MonitorConstants.FILE_NODE_PATH + MonitorConstants.MONITOR_PATH_SEPARATOR
+ + processorName.replaceAll("\\.", "_");
+ StatMonitor statMonitor = StatMonitor.getInstance();
+ statMonitor.registerStatistics(statStorageDeltaName, this);
+ Map<String, String> typeMap = new HashMap<>();
+ for (StorageGroupProcessorStatConstants statConstant :
+ StorageGroupProcessorStatConstants.values()) {
+ typeMap
+ .put(statStorageGroupName + MonitorConstants.MONITOR_PATH_SEPARATOR + statConstant.name(),
+ MonitorConstants.DATA_TYPE_INT64);
+ }
+ StatMonitor.getInstance().registerStatStorageGroup(typeMap);
+ }
+
+ statisticMap = new HashMap<>();
+ for (StorageGroupProcessorStatConstants statConstant :
+ StorageGroupProcessorStatConstants.values()) {
+ statisticMap.put(statConstant.name(), new AtomicLong(0));
+ }
+ }
+
+ @Override
+ public List<String> getAllPathForStatistic() {
+ List<String> list = new ArrayList<>();
+ for (StorageGroupProcessorStatConstants statConstant :
+ StorageGroupProcessorStatConstants.values()) {
+ list.add(
+ statStorageGroupName + MonitorConstants.MONITOR_PATH_SEPARATOR + statConstant.name());
+ }
+ return list;
+ }
+
+ @Override
+ public Map<String, AtomicLong> getStatParamsHashMap() {
+ return statisticMap;
+ }
+
+ private FileSchema constructFileSchema(String processorName) {
+ List<MeasurementSchema> columnSchemaList;
+ columnSchemaList = MManager.getInstance().getSchemaForFileName(processorName);
+
+ FileSchema schema = new FileSchema();
+ for (MeasurementSchema measurementSchema : columnSchemaList) {
+ schema.registerMeasurement(measurementSchema);
+ }
+ return schema;
+
+ }
+
+ private TsFileProcessor getTsFileProcessor() throws TsFileProcessorException {
+ if (tsFileProcessor.isClosed()) {
+ tsFileProcessor.reopen();
+ }
+ return tsFileProcessor;
+ }
+
+ private OverflowProcessor getOverflowProcessor() throws TsFileProcessorException {
+ if (overflowProcessor.isClosed()) {
+ overflowProcessor.reopen();
+ }
+ return overflowProcessor;
+ }
+
+ public class StorageGroupFlushFuture implements Future<Boolean>{
+ private Future<Boolean> tsFileFuture;
+ private Future<Boolean> overflowFuture;
+
+ StorageGroupFlushFuture(Future<Boolean> tsFileFuture, Future<Boolean> overflowFuture) {
+ this.tsFileFuture = tsFileFuture != null ? tsFileFuture : new ImmediateFuture<>(true);
+ this.overflowFuture = overflowFuture != null ? tsFileFuture : new ImmediateFuture<>(true);
+ }
+
+ @Override
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ boolean tsCanceled = tsFileFuture.cancel(mayInterruptIfRunning);
+ boolean overflowCanceled = overflowFuture.cancel(mayInterruptIfRunning);
+ return tsCanceled && overflowCanceled;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return tsFileFuture.isCancelled() && overflowFuture.isCancelled();
+ }
+
+ @Override
+ public boolean isDone() {
+ return tsFileFuture.isDone() && overflowFuture.isDone();
+ }
+
+ @Override
+ public Boolean get() throws InterruptedException, ExecutionException {
+ boolean tsFileOK = tsFileFuture.get();
+ boolean overflowOK = overflowFuture.get();
+ return tsFileOK && overflowOK;
+ }
+
+ @Override
+ public Boolean get(long timeout, TimeUnit unit)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ boolean tsFileOK = tsFileFuture.get(timeout, unit);
+ boolean overflowOK = overflowFuture.get(timeout, unit);
+ return tsFileOK && overflowOK;
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return this == o;
+ }
+
+ @Override
+ public int hashCode() {
+ return super.hashCode();
+ }
+
+ long getLastInsetTime(String deviceId) throws TsFileProcessorException {
+ return getTsFileProcessor().getLastInsertTime(deviceId);
+ }
+
+ void deleteInSeqFile(String deviceId, String measurementId, long timestamp)
+ throws TsFileProcessorException {
+ try {
+ getTsFileProcessor().delete(deviceId, measurementId, timestamp);
+ } catch (IOException e) {
+ throw new TsFileProcessorException(e);
+ }
+ }
+
+ void deleteInOverflow(String deviceId, String measurementId, long timestamp)
+ throws TsFileProcessorException {
+ try {
+ getOverflowProcessor().delete(deviceId, measurementId, timestamp);
+ } catch (IOException e) {
+ throw new TsFileProcessorException(e);
+ }
+ }
+
+ /**
+ * Assign a token to a new query and put it into the newSet so that we know which
+ */
+ int addMultiPassCount() {
+ LOGGER.debug("Add MultiPassCount: read lock newMultiPassCount.");
+ Integer token = newMultiPassCount.incrementAndGet();
+
+ newMultiPassTokenSet.add(token);
+ LOGGER.debug("Add multi token:{}, nsPath:{}.", token, getProcessorName());
+ return token;
+ }
+
+ /**
+ * decrease multiple pass count.
+ */
+ void decreaseMultiPassCount(int token) {
+ if (newMultiPassTokenSet.contains(token)) {
+ newMultiPassTokenSet.remove(token);
+ LOGGER.debug("Remove multi token:{}, nspath:{}, new set:{}, count:{}", token,
+ getProcessorName(),
+ newMultiPassTokenSet, newMultiPassCount);
+ } else if (oldMultiPassTokenSet != null && oldMultiPassTokenSet.contains(token)) {
+ // remove token first, then unlock
+ oldMultiPassTokenSet.remove(token);
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Remove multi token:{}, old set:{}", token, oldMultiPassTokenSet);
+ }
+ } else {
+ LOGGER.error("remove token error:{},new set:{}, old set:{}", token, newMultiPassTokenSet,
+ oldMultiPassTokenSet);
+ // should add throw exception
+ }
+ }
+
+ /**
+ * append one specified TsFile to this StorageGroup.
+ *
+ * @param destFileResource the information of destination file
+ * @param srcFilePath the path of the source file
+ */
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ void appendFile(TsFileResource destFileResource, String srcFilePath)
+ throws TsFileProcessorException {
+ try {
+ if (!destFileResource.getFile().getParentFile().exists()) {
+ destFileResource.getFile().getParentFile().mkdirs();
+ }
+ // move file
+ File srcFile = new File(srcFilePath);
+ File destFile = destFileResource.getFile();
+ if (!srcFile.exists()) {
+ throw new StorageGroupProcessorException(
+ String.format("The source file %s does not exist.", srcFilePath));
+ }
+ if (destFile.exists()) {
+ throw new StorageGroupProcessorException(
+ String.format("The destination file %s already exists.",
+ destFileResource.getFile().getAbsolutePath()));
+ }
+ if (!srcFile.renameTo(destFile)) {
+ LOGGER.warn("File renaming failed when appending new file. Origin: {}, Target: {}",
+ srcFile.getPath(), destFile.getPath());
+ }
+ // append the new tsfile
+ this.tsFileProcessor.appendFile(destFileResource);
+ // reconstruct the inverted index of the new storage groups
+ } catch (Exception e) {
+ LOGGER.error("Failed to append the tsfile {} to storage group processor {}.",
+ destFileResource,
+ getProcessorName());
+ throw new TsFileProcessorException(e);
+ }
+ }
+
+ /**
+ * get overlap tsfiles which are conflict with the appendFile.
+ *
+ * @param appendFile the appended tsfile information
+ */
+ List<String> getOverlapFiles(TsFileResource appendFile, String uuid)
+ throws TsFileProcessorException {
+ return tsFileProcessor.getOverlapFiles(appendFile, uuid);
+ }
+
+ /**
+ * register a timeseries in this StorageGroup
+ * @param measurementId
+ * @param dataType
+ * @param encoding
+ * @param compressor
+ * @param props
+ */
+ void addTimeSeries(String measurementId, TSDataType dataType, TSEncoding encoding,
+ CompressionType compressor, Map<String, String> props) {
+ fileSchema.registerMeasurement(new MeasurementSchema(measurementId, dataType, encoding,
+ compressor, props));
+ }
+
+ /**
+ * submit the merge task to the <code>MergePool</code>.
+ *
+ * @return null -can't submit the merge task, because this storage group is not overflowed or
+ * it is
+ * merging now. Future - submit the merge task successfully.
+ */
+ Future submitToMerge() {
+ ZoneId zoneId = IoTDBDescriptor.getInstance().getConfig().getZoneID();
+ if (lastMergeTime > 0 && LOGGER.isInfoEnabled()) {
+ long thisMergeTime = System.currentTimeMillis();
+ long mergeTimeInterval = thisMergeTime - lastMergeTime;
+ LOGGER.info(
+ "The storage group {} last merge time is {}, this merge time is {}, "
+ + "merge time interval is {}s",
+ processorName, ofInstant(Instant.ofEpochMilli(lastMergeTime),
+ zoneId), ofInstant(Instant.ofEpochMilli(thisMergeTime),
+ zoneId), mergeTimeInterval / 1000);
+ }
+ lastMergeTime = System.currentTimeMillis();
+
+ if (overflowProcessor.totalFileSize() < IoTDBDescriptor.getInstance()
+ .getConfig().getOverflowFileSizeThreshold()) {
+ if (LOGGER.isInfoEnabled()) {
+ LOGGER.info(
+ "Skip this merge taks submission, because the total file size {} of overflow processor {} "
+ + "does not reaches the threshold {}.",
+ MemUtils.bytesCntToStr(overflowProcessor.totalFileSize()), getProcessorName(),
+ MemUtils.bytesCntToStr(
+ IoTDBDescriptor.getInstance().getConfig().getOverflowFileSizeThreshold()));
+ }
+ return null;
+ }
+
+ if (!isMerging) {
+ Callable<Exception> mergeThread;
+ mergeThread = new MergeTask(this);
+ LOGGER.info("Submit the merge task, the merge storage group is {}", processorName);
+ return MergeManager.getInstance().submit(mergeThread);
+ } else {
+ LOGGER.warn(
+ "Skip this merge task submission, because last merge task is not over yet, "
+ + "the merge storage group is {}",
+ processorName);
+ }
+ return null;
+ }
+
+ /**
+ * Divide seqFiles and unseqFiles into k groups separately such that seqFiles in the ith group
+ * overlp some unseqFiles in the ith group but overlap none unseqFiles in other groups.
+ * @param seqFiles
+ * @param unseqFiles
+ */
+ private Pair<List<List<TsFileResource>>, List<List<TsFileResource>>> groupMergeFiles(
+ List<TsFileResource> seqFiles, List<TsFileResource> unseqFiles) {
+ List<List<TsFileResource>> seqGroups = new ArrayList<>();
+ List<List<TsFileResource>> unseqGroups = new ArrayList<>();
+ boolean[] seqVisited = new boolean[seqFiles.size()];
+ boolean[] unseqVisited = new boolean[unseqFiles.size()];
+ for (int i = 0; i < seqFiles.size(); i++) {
+ seqVisited[i] = false;
+ }
+ for (int i = 0; i < unseqFiles.size(); i++) {
+ unseqVisited[i] = false;
+ }
+
+ for (int i = 0; i < seqFiles.size(); i++) {
+ if (seqVisited[i]) {
+ continue;
+ }
+
+ List<TsFileResource> currSeqGroup = new ArrayList<>();
+ List<TsFileResource> currUnseqGroup = new ArrayList<>();
+
+ // we only need to check new files (range [lastIndex:]) added by last iteration,
+ // older files can be skipped
+ int lastSeqIndex = 0;
+ int lastUnseqIndex = 0;
+
+ currSeqGroup.add(seqFiles.get(i));
+ boolean updated = true;
+ while (updated) {
+ // find unseqFiles that overlap current seqFiles
+ updated = searchOverlap(unseqFiles, unseqVisited, currSeqGroup,
+ lastSeqIndex, currUnseqGroup);
+ lastSeqIndex = currSeqGroup.size();
+ // find seqFiles that overlap current unseqFiles
+ updated = updated || searchOverlap(seqFiles, seqVisited, currUnseqGroup,
+ lastUnseqIndex, currSeqGroup);
+ lastUnseqIndex = currUnseqGroup.size();
+ }
+
+ seqGroups.add(currSeqGroup);
+ unseqGroups.add(currUnseqGroup);
+ }
+ return new Pair<>(seqGroups, unseqGroups);
+ }
+
+ /**
+ * For each TsFile in 'candidates', ignoring the ith element if 'candidatesVisited[i]' == true,
+ * find out whether it overlaps another TsFile in 'targets[offset:]' and add the candidate
+ * into result.
+ * @param candidates
+ * @param candidatesVisited
+ * @param targets
+ * @param offset
+ * @param result
+ * @return
+ */
+ private boolean searchOverlap(List<TsFileResource> candidates, boolean[] candidatesVisited,
+ List<TsFileResource> targets, int offset, List<TsFileResource> result) {
+ boolean found = false;
+ for (int j = 0; j < candidates.size(); j++) {
+ if (candidatesVisited[j]) {
+ continue;
+ }
+ TsFileResource candidate = candidates.get(j);
+ for (int k = offset; k < targets.size(); k++) {
+ if (candidate.overlaps(targets.get(k))) {
+ result.add(candidate);
+ candidatesVisited[j] = true;
+ found = true;
+ }
+ }
+ }
+ return found;
+ }
+
+ /**
+ * Merge this storage group, merge the tsfile data with overflow data.
+ */
+ public void merge() throws StorageGroupProcessorException {
+ writeLock();
+ Pair<List<List<TsFileResource>>, List<List<TsFileResource>>> fileGroups;
+ List<TsFileResource> unseqFiles;
+ List<TsFileResource> seqFiles;
+ try {
+ // close seqFile and overflow(unseqFile), prepare for merge
+ LOGGER.info("The StorageGroupProcessor {} begins to merge.", processorName);
+ close();
+ // change status from work to merge
+ isMerging = true;
+
+ // find what files are to be processed in this merge
+ seqFiles = new ArrayList<>(tsFileProcessor.getTsFileResources());
+ unseqFiles = new ArrayList<>(overflowProcessor.getTsFileResources());
+
+ fileGroups = groupMergeFiles(seqFiles, unseqFiles);
+ } catch (TsFileProcessorException e) {
+ throw new StorageGroupProcessorException(e);
+ } finally {
+ writeUnlock();
+ isMerging = false;
+ }
+
+ List<List<TsFileResource>> seqGroups = fileGroups.left;
+ List<List<TsFileResource>> unseqGroups = fileGroups.right;
+
+ LOGGER.info("{}: There are {} groups to be merged", processorName, seqGroups.size());
+
+
+ for (int i = 0; i < seqGroups.size(); i++) {
+ List<TsFileResource> seqGroup = seqGroups.get(i);
+ List<TsFileResource> unseqGroup = unseqGroups.get(i);
+
+ int seqFileSize = seqGroup.size();
+ int unseqFileSize = unseqGroup.size();
+
+ long startTime = System.currentTimeMillis();
+ LOGGER.info("{}: Starting merging the {} group, there are {} seq groups and {} unseq groups",
+ processorName, i, seqFileSize, unseqFileSize);
+ TsFileResource newResource = mergeFileGroup(seqGroup, unseqGroup);
+ try {
+ addMergedFile(newResource, seqGroup, unseqGroup);
+ } catch (IOException e) {
+ LOGGER.error("Cannot merge the {} group because:", e);
+ continue;
+ }
+ long endTime = System.currentTimeMillis();
+ LOGGER.info("{}: End merging the {} group after {}ms", processorName, i, endTime - startTime);
+ }
+ }
+
+ private void waitForQueriesOnOldFile() {
+ long sleepTime = 1000;
+ long maxSleepTime = 60 * 1000L;
+ while (!oldMultiPassTokenSet.isEmpty()) {
+ try {
+ Thread.sleep(sleepTime);
+ sleepTime = sleepTime * 2 > maxSleepTime ? maxSleepTime : sleepTime * 2;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOGGER.warn("{}: unexpected interuption when waiting for queries on old files",
+ processorName, e);
+ }
+ }
+ }
+
+ private void addMergedFile(TsFileResource mergedFile, List<TsFileResource> seqGroup,
+ List<TsFileResource> unseqGroup) throws IOException {
+ writeLock();
+ try {
+ // record merge old files so that they can be deleted after system crush
+ recordOldFiles(mergedFile, seqGroup, unseqGroup);
+ // update file lists of seqProcessor and unseqProcessor
+ tsFileProcessor.replaceFiles(seqGroup, mergedFile);
+ overflowProcessor.replaceFiles(unseqGroup, null);
+ // rename the new file so that it will not be deleted after system crush
+ String newName = mergedFile.getFile().getName().replace(MERGE_TEMP_SUFFIX, "");
+ File newFile = new File(mergedFile.getFile().getParent(), newName);
+ //noinspection ResultOfMethodCallIgnored
+ newFile.delete();
+ if (!mergedFile.getFile().renameTo(newFile)) {
+ throw new IOException(String.format("Cannot rename merged file %s",
+ mergedFile.getFilePath()));
+ }
+ // now new queries will no longer use old files, so record new queries in another token set
+ oldMultiPassTokenSet = newMultiPassTokenSet;
+ newMultiPassTokenSet = new HashSet<>();
+ } finally {
+ writeUnlock();
+ }
+ // wait until all old queries end
+ waitForQueriesOnOldFile();
+ // remove old files
+ for (TsFileResource oldFile : seqGroup) {
+ Files.deleteIfExists(oldFile.getFile().toPath());
+ }
+ for (TsFileResource oldFile : unseqGroup) {
+ Files.deleteIfExists(oldFile.getFile().toPath());
+ }
+ // remove old file records
+ Files.deleteIfExists(getMergeFileRecord().toPath());
+ }
+
+ private File getMergeFileRecord() {
+ return new File(storageGroupDir, OLD_FILE_RECORD);
+ }
+
+ private void recordOldFiles(TsFileResource newFile,
+ List<TsFileResource> seqGroup, List<TsFileResource> unseqGroup)
+ throws IOException {
+ File oldFileRecord = getMergeFileRecord();
+ try (BufferedWriter bufferedWriter = new BufferedWriter(new FileWriter(oldFileRecord, true))) {
+ for (TsFileResource resource : seqGroup) {
+ bufferedWriter.write(resource.getFilePath());
+ bufferedWriter.write("\n");
+ }
+ for (TsFileResource resource : unseqGroup) {
+ bufferedWriter.write(resource.getFilePath());
+ bufferedWriter.write("\n");
+ }
+ bufferedWriter.write(newFile.getFilePath());
+ }
+ }
+
+ private TsFileResource mergeFileGroup(List<TsFileResource> seqGroup, List<TsFileResource> unseqGroup)
+ throws StorageGroupProcessorException {
+ if (unseqGroup.isEmpty()) {
+ return seqGroup.get(0);
+ }
+ // find all timeseries (possibly) of each device (the key) contained in these files.
+ Map<String, List<Path>> pathMap;
+ try {
+ pathMap = resolveMergePaths(seqGroup, unseqGroup);
+ } catch (PathErrorException e) {
+ throw new StorageGroupProcessorException(e);
+ }
+ // min data time of all files
+ long minimumTime = resolveMinimumTime(seqGroup, unseqGroup);
+
+ String mergeBaseDir = Directories.getInstance().getNextFolderForTsfile();
+ String mergeFileName = minimumTime
+ + EngineConstants.TSFILE_NAME_SEPARATOR + System.currentTimeMillis()
+ + MERGE_TEMP_SUFFIX;
+ String mergeOutputPath = constructOutputFilePath(mergeBaseDir, getProcessorName(),
+ mergeFileName);
+ mergeFileName = getProcessorName() + File.separatorChar + mergeFileName;
+ try {
+ mergeFileWriter = new TsFileIOWriter(new File(mergeOutputPath));
+ } catch (IOException e) {
+ throw new StorageGroupProcessorException(e);
+ }
+ mergingModification = new ModificationFile(mergeOutputPath
+ + ModificationFile.FILE_SUFFIX);
+ mergeIsChunkGroupHasData = false;
+ mergeStartPos = -1;
+
+ Map<String, Long> startTimeMap = new HashMap<>();
+ Map<String, Long> endTimeMap = new HashMap<>();
+
+
+ for (Entry<String, List<Path>> entry : pathMap.entrySet()) {
+ mergeDevice(entry.getKey(), entry.getValue(), seqGroup, unseqGroup, startTimeMap, endTimeMap);
+ }
+
+ try {
+ mergeFileWriter.endFile(fileSchema);
+ } catch (IOException e) {
+ throw new StorageGroupProcessorException(e);
+ }
+ TsFileResource newResource;
+ try {
+ newResource = new TsFileResource(
+ new File(mergeBaseDir + File.separator + mergeFileName), false);
+ } catch (IOException e) {
+ throw new StorageGroupProcessorException(e);
+ }
+ newResource.setStartTimeMap(startTimeMap);
+ newResource.setEndTimeMap(endTimeMap);
+ newResource.setModFile(mergingModification);
+ mergingModification = null;
+ mergeFileWriter = null;
+ return newResource;
+ }
+
+ private void mergeDevice(String deviceId, List<Path> paths, List<TsFileResource> seqGroup,
+ List<TsFileResource> unseqGroup, Map<String, Long> startTimeMap,
+ Map<String, Long> endTimeMap) {
+ QueryContext context = new QueryContext();
+ IReader seriesReader = null;
+ ChunkGroupFooter footer;
+ int numOfChunks = 0;
+ // start writing a device
+ for (Path path : paths) {
+ // start writing a series
+ try {
+ // sequence
+ SeriesDataSource seqSource = new SeriesDataSource(path, seqGroup, null, null);
+ SeriesDataSource unseqSource = new SeriesDataSource(path, unseqGroup, null, null);
+
+ seriesReader = SeriesReaderFactory.getInstance().createSeriesReaderForMerge(seqSource,
+ unseqSource, context);
+ TSDataType dataType = MManager.getInstance().getSeriesType(path.getFullPath());
+ // end writing a series
+ numOfChunks += queryAndWriteSeries(seriesReader, path, null, dataType,
+ startTimeMap, endTimeMap);
+ } catch (PathErrorException | IOException e) {
+ LOGGER.error("{}: error occurred when merging {}", processorName, path.getFullPath(), e);
+ } finally {
+ if (seriesReader != null) {
+ try {
+ seriesReader.close();
+ } catch (IOException e) {
+ LOGGER.error("{}: cannot close series reader of {} when merge", processorName,
+ path.getFullPath(), e);
+ }
+ }
+ }
+ }
+ // end writing a device
+ try {
+ if (mergeIsChunkGroupHasData) {
+ // end the new rowGroupMetadata
+ long size = mergeFileWriter.getPos() - mergeStartPos;
+ footer = new ChunkGroupFooter(deviceId, size, numOfChunks);
+ mergeFileWriter.endChunkGroup(footer, 0);
+ }
+ } catch (IOException e) {
+ LOGGER.error("{}: Cannot end ChunkGroup of {} when merge", processorName, deviceId, e);
+ }
+ }
+
+ private Map<String, List<Path>> resolveMergePaths(List<TsFileResource> seqGroup, List<TsFileResource> unseqGroup)
+ throws PathErrorException {
+ Map<String, List<Path>> paths = new HashMap<>();
+
+ resolveMergeGroupPaths(seqGroup, paths);
+ resolveMergeGroupPaths(unseqGroup, paths);
+
+ return paths;
+ }
+
+ private void resolveMergeGroupPaths(List<TsFileResource> fileGroup,
+ Map<String, List<Path>> paths) throws PathErrorException {
+ for (TsFileResource tsFile : fileGroup) {
+ for (String device : tsFile.getDevices()) {
+ if (paths.containsKey(device)) {
+ continue;
+ }
+ List<String> pathStrings = MManager.getInstance().getLeafNodePathInNextLevel(device);
+ for (String string : pathStrings) {
+ paths.computeIfAbsent(device, x -> new ArrayList<>()).add(new Path(string));
+ }
+ }
+ }
+ }
+
+ private long resolveMinimumTime(List<TsFileResource> seqGroup, List<TsFileResource> unseqGroup) {
+ long minTime = Long.MAX_VALUE;
+ for (TsFileResource tsFile : seqGroup) {
+ for (Long startTime : tsFile.getStartTimeMap().values()) {
+ if (startTime < minTime) {
+ minTime = startTime;
+ }
+ }
+ }
+ for (TsFileResource tsFile : unseqGroup) {
+ for (Long startTime : tsFile.getStartTimeMap().values()) {
+ if (startTime < minTime) {
+ minTime = startTime;
+ }
+ }
+ }
+ return minTime;
+ }
+
+ private String constructOutputFilePath(String baseDir, String processorName, String fileName) {
+
+ String localBaseDir = baseDir;
+ if (localBaseDir.charAt(localBaseDir.length() - 1) != File.separatorChar) {
+ localBaseDir = localBaseDir + File.separatorChar + processorName;
+ }
+ File dataDir = new File(localBaseDir);
+ if (!dataDir.exists()) {
+ LOGGER.warn("The bufferwrite processor data dir doesn't exists, create new directory {}",
+ localBaseDir);
+ dataDir.mkdirs();
+ }
+ File outputFile = new File(dataDir, fileName);
+ return outputFile.getPath();
+ }
+
+ private int queryAndWriteSeries(IReader seriesReader, Path path,
+ SingleSeriesExpression seriesFilter, TSDataType dataType,
+ Map<String, Long> startTimeMap, Map<String, Long> endTimeMap)
+ throws IOException {
+ int numOfChunk = 0;
+ if (!seriesReader.hasNext()) {
+ LOGGER.debug(
+ "The time-series {} has no data with the filter {} in the storage group {}",
+ path, seriesFilter, getProcessorName());
+ } else {
+ numOfChunk ++;
+ TimeValuePair timeValuePair = seriesReader.next();
+ if (!mergeIsChunkGroupHasData) {
+ // start a new rowGroupMetadata
+ mergeIsChunkGroupHasData = true;
+ // the datasize and numOfChunk is fake
+ // the accurate datasize and numOfChunk will get after write all this device data.
+ mergeFileWriter.startFlushChunkGroup(path.getDevice());
+ mergeStartPos = mergeFileWriter.getPos();
+ }
+ // init the serieswWriteImpl
+ MeasurementSchema measurementSchema = fileSchema
+ .getMeasurementSchema(path.getMeasurement());
+ ChunkBuffer pageWriter = new ChunkBuffer(measurementSchema);
+ int pageSizeThreshold = TSFileConfig.pageSizeInByte;
+ ChunkWriterImpl seriesWriterImpl = new ChunkWriterImpl(measurementSchema, pageWriter,
+ pageSizeThreshold);
+ // write the series data
+ writeOneSeries(path.getDevice(), seriesWriterImpl, dataType,
+ seriesReader, startTimeMap, endTimeMap, timeValuePair);
+ // flush the series data
+ seriesWriterImpl.writeToFileWriter(mergeFileWriter);
+ }
+ return numOfChunk;
+ }
+
+ private void writeOneSeries(String deviceId, ChunkWriterImpl seriesWriterImpl,
+ TSDataType dataType, IReader seriesReader, Map<String, Long> startTimeMap,
+ Map<String, Long> endTimeMap, TimeValuePair firstTVPair) throws IOException {
+ long startTime;
+ long endTime;
+ TimeValuePair localTV = firstTVPair;
+ writeTVPair(seriesWriterImpl, dataType, localTV);
+ startTime = endTime = localTV.getTimestamp();
+ if (!startTimeMap.containsKey(deviceId) || startTimeMap.get(deviceId) > startTime) {
+ startTimeMap.put(deviceId, startTime);
+ }
+ if (!endTimeMap.containsKey(deviceId) || endTimeMap.get(deviceId) < endTime) {
+ endTimeMap.put(deviceId, endTime);
+ }
+ while (seriesReader.hasNext()) {
+ localTV = seriesReader.next();
+ endTime = localTV.getTimestamp();
+ writeTVPair(seriesWriterImpl, dataType, localTV);
+ }
+ if (!endTimeMap.containsKey(deviceId) || endTimeMap.get(deviceId) < endTime) {
+ endTimeMap.put(deviceId, endTime);
+ }
+ }
+
+ private void writeTVPair(ChunkWriterImpl seriesWriterImpl, TSDataType dataType,
+ TimeValuePair timeValuePair) throws IOException {
+ switch (dataType) {
+ case BOOLEAN:
+ seriesWriterImpl.write(timeValuePair.getTimestamp(), timeValuePair.getValue().getBoolean());
+ break;
+ case INT32:
+ seriesWriterImpl.write(timeValuePair.getTimestamp(), timeValuePair.getValue().getInt());
+ break;
+ case INT64:
+ seriesWriterImpl.write(timeValuePair.getTimestamp(), timeValuePair.getValue().getLong());
+ break;
+ case FLOAT:
+ seriesWriterImpl.write(timeValuePair.getTimestamp(), timeValuePair.getValue().getFloat());
+ break;
+ case DOUBLE:
+ seriesWriterImpl.write(timeValuePair.getTimestamp(), timeValuePair.getValue().getDouble());
+ break;
+ case TEXT:
+ seriesWriterImpl.write(timeValuePair.getTimestamp(), timeValuePair.getValue().getBinary());
+ break;
+ default:
+ LOGGER.error("Not support data type: {}", dataType);
+ break;
+ }
+ }
+
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ private void cleanLastMerge() throws IOException {
+ // if system crushes during last merge, there may be undeleted old files
+ // these files will be deleted in this method
+ File mergeRecord = getMergeFileRecord();
+ if (!mergeRecord.exists()) {
+ return;
+ }
+
+ List<String> filePaths = new ArrayList<>();
+ try (BufferedReader bufferedReader = new BufferedReader(new FileReader(mergeRecord))) {
+ bufferedReader.lines().forEach(filePaths::add);
+ }
+ // the last file should be the merged new file
+ String lastFilePath = filePaths.get(filePaths.size() - 1);
+ if (!lastFilePath.endsWith(MERGE_TEMP_SUFFIX)) {
+ // the record is incomplete, consider last merge failed, do not delete any file
+ LOGGER.warn("The merge record is not complete, the last merge seems failed");
+ mergeRecord.delete();
+ return;
+ }
+ File mergedTempFile = new File(lastFilePath);
+ if (mergedTempFile.exists()) {
+ // the temp file has not been renamed, rename it
+ File newName = new File(mergedTempFile.getParent(),
+ mergedTempFile.getName().replace(MERGE_TEMP_SUFFIX, ""));
+ if (!mergedTempFile.renameTo(newName)) {
+ LOGGER.error("Cannot rename merged temp file {}, abort last merge",
+ mergedTempFile.getAbsolutePath());
+ mergeRecord.delete();
+ return;
+ }
+ }
+ // the merged temp file has been renamed, the old files can be safely deleted
+ for (String filePath : filePaths) {
+ new File(filePath).delete();
+ }
+ }
+}
diff --cc iotdb/src/main/java/org/apache/iotdb/db/engine/tsfiledata/TsFileProcessor.java
index 47f9a16,0000000..df480ea
mode 100644,000000..100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/tsfiledata/TsFileProcessor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/tsfiledata/TsFileProcessor.java
@@@ -1,1002 -1,0 +1,964 @@@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.engine.tsfiledata;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReentrantLock;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.directories.Directories;
+import org.apache.iotdb.db.engine.Processor;
+import org.apache.iotdb.db.engine.EngineConstants;
+import org.apache.iotdb.db.engine.sgmanager.TsFileResource;
+import org.apache.iotdb.db.engine.memcontrol.BasicMemController;
+import org.apache.iotdb.db.engine.memcontrol.BasicMemController.UsageLevel;
+import org.apache.iotdb.db.engine.memtable.IMemTable;
+import org.apache.iotdb.db.engine.memtable.MemSeriesLazyMerger;
+import org.apache.iotdb.db.engine.memtable.MemTableFlushUtil;
+import org.apache.iotdb.db.engine.memtable.PrimitiveMemTable;
+import org.apache.iotdb.db.engine.modification.Deletion;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.pool.FlushManager;
+import org.apache.iotdb.db.engine.datasource.ReadOnlyMemChunk;
+import org.apache.iotdb.db.engine.datasource.SeriesDataSource;
+import org.apache.iotdb.db.engine.datasource.UnsealedTsFile;
+import org.apache.iotdb.db.engine.sgmanager.OperationResult;
+import org.apache.iotdb.db.engine.sgmanager.StorageGroupProcessor;
+import org.apache.iotdb.db.engine.version.VersionController;
+import org.apache.iotdb.db.exception.TsFileProcessorException;
+import org.apache.iotdb.db.qp.constant.DatetimeUtils;
+import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
+import org.apache.iotdb.db.qp.physical.crud.UpdatePlan;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.sync.SyncUtils;
+import org.apache.iotdb.db.utils.ImmediateFuture;
+import org.apache.iotdb.db.utils.MemUtils;
+import org.apache.iotdb.db.utils.QueryUtils;
+import org.apache.iotdb.db.writelog.manager.MultiFileLogNodeManager;
+import org.apache.iotdb.db.writelog.node.WriteLogNode;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.expression.impl.SingleSeriesExpression;
+import org.apache.iotdb.tsfile.write.schema.FileSchema;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
++import org.apache.iotdb.tsfile.write.writer.NativeRestorableIOWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Each storage group has a TsFileProcessor instance. Though there are many clients read/write data
+ * by accessing TsFileProcessor, they need to get the readWriteReentrantLock of this processor
+ * first. Therefore, as for different clients, the class looks like thread safe.
+ * <br/> The class has two backend tasks:
+ * (1) submit a flush job to flush data from memtable to disk async;
+ * (2) close a tsfile and open a new one.
+ * users also need to get the write lock to call these two tasks.
+ */
+public class TsFileProcessor extends Processor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(TsFileProcessor.class);
+
+ //this is just a part of fileSchemaRef: only the measurements that belong to this TsFileProcessor
+ // are in this fileSchemaRef. And, this filed is shared with other classes (i.e., storage group
+ // processor), so be careful if you modify it.
+ private FileSchema fileSchemaRef;
+
+
+ private volatile Future<Boolean> flushFuture = new ImmediateFuture<>(true);
- //when a flush task is in the backend, then it has not conflict with write operatons (because
++ //when a flush task is in the backend, then it has not conflict with write operations (because
+ // write operation modifies wokMemtable, while flush task uses flushMemtable). However, the flush
+ // task has concurrent problem with query operations, because the query needs to read data from
+ // flushMemtable and the table may be clear if the flush operation ends. So, a Lock is needed.
+ private ReentrantLock flushQueryLock = new ReentrantLock();
+ private AtomicLong memSize = new AtomicLong();
+
+ //fileNamePrefix (system time, rather than data time) time unit: nanosecond
+ // this is used for generate the new TsFile name
+ private long fileNamePrefix = System.nanoTime();
+ //the times of calling insertion function (between two flush operations).
+ private long valueCount = 0;
+
+
+ private IMemTable workMemTable;
+ private IMemTable flushMemTable;
+ protected RestorableTsFileIOWriter writer;
+ private File insertFile;
+ private TsFileResource currentResource;
+
+ private List<TsFileResource> tsFileResources;
+ // device -> datafiles
+ private Map<String, List<TsFileResource>> inverseIndexOfResource = new HashMap<>();
+
+ // device -> min timestamp in current data file
+ private Map<String, Long> minWrittenTimeForEachDeviceInCurrentFile;
+ // device -> max timestamp in current data file (maybe in memory)
+ private Map<String, Long> maxWrittenTimeForEachDeviceInCurrentFile;
+ // device -> the max timestamp in current data file on disk
+ private Map<String, Long> lastFlushedTimeForEachDevice;
+ //wal
+ private WriteLogNode logNode;
+ private VersionController versionController;
+
+ private boolean isClosed = true;
++ private boolean isFlush;
++ private long lastFlushTime = -1;
+
+ /**
+ * constructor of BufferWriteProcessor. data will be stored in baseDir/processorName/ folder.
+ *
+ * @param processorName processor name
+ * @param fileSchemaRef file schema
+ * @throws TsFileProcessorException TsFileProcessorException
+ */
+ @SuppressWarnings({"squid:S2259", "squid:S3776"})
+ public TsFileProcessor(String processorName, VersionController versionController,
+ FileSchema fileSchemaRef) throws TsFileProcessorException {
+ super(processorName);
+ this.fileSchemaRef = fileSchemaRef;
+ this.processorName = processorName;
+
+ reopen();
+
+ this.versionController = versionController;
- if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
++
++ // implicit initialization
++ getLogNode();
++ }
++
++ private WriteLogNode getLogNode() throws TsFileProcessorException {
++ if (logNode == null && IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
+ try {
+ logNode = MultiFileLogNodeManager.getInstance().getNode(
+ processorName + getLogSuffix(),
+ writer.getRestoreFilePath());
+ } catch (IOException e) {
+ throw new TsFileProcessorException(e);
+ }
+ }
++ return logNode;
+ }
+
+ public void reopen() throws TsFileProcessorException {
+ if (!isClosed) {
+ return;
+ }
+
+ if (workMemTable == null) {
+ workMemTable = new PrimitiveMemTable();
+ } else {
+ workMemTable.clear();
+ }
+
+ boolean noResources = tsFileResources == null;
+ if (noResources) {
+ initResources();
+ } else {
+ initCurrentTsFile(generateNewTsFilePath());
+ }
+
+ isClosed = false;
+ }
+
+ @SuppressWarnings({"ResultOfMethodCallIgnored"})
+ private void initResources() throws TsFileProcessorException {
+ tsFileResources = new ArrayList<>();
+ lastFlushedTimeForEachDevice = new HashMap<>();
+ minWrittenTimeForEachDeviceInCurrentFile = new HashMap<>();
+ maxWrittenTimeForEachDeviceInCurrentFile = new HashMap<>();
- File unclosedFile = null;
- String unclosedFileName = null;
- int unclosedFileCount = 0;
- for (String folderPath : getAllDataFolders()) {
- File dataFolder = new File(folderPath, processorName);
- if (!dataFolder.exists()) {
- // we do not add the unclosed tsfile into tsFileResources.
- File[] unclosedFiles = dataFolder
- .listFiles(x -> x.getName().contains(RestorableTsFileIOWriter.RESTORE_SUFFIX));
- if (unclosedFiles != null) {
- unclosedFileCount += unclosedFiles.length;
- }
- if (unclosedFileCount > 1) {
- break;
- } else if (unclosedFiles != null) {
- unclosedFileName = unclosedFiles[0].getName()
- .split(RestorableTsFileIOWriter.RESTORE_SUFFIX)[0];
- unclosedFile = new File(unclosedFiles[0].getParentFile(), unclosedFileName);
++ try {
++ for (String folderPath : getAllDataFolders()) {
++ File dataFolder = new File(folderPath, processorName);
++ if (!dataFolder.exists()) {
++ // find unclosed files and recover them
++ File[] unclosedRestoreFiles = dataFolder
++ .listFiles(x -> x.getName().contains(RestorableTsFileIOWriter.RESTORE_SUFFIX));
++ if (unclosedRestoreFiles != null) {
++ for (File unclosedRestoreFile : unclosedRestoreFiles) {
++ String unclosedFileName = unclosedRestoreFile.getName()
++ .split(RestorableTsFileIOWriter.RESTORE_SUFFIX)[0];
++ File unclosedFile = new File(unclosedRestoreFile.getParentFile(), unclosedFileName);
++ // recover unclosed file and close it
++ new NativeRestorableIOWriter(unclosedFile, false);
++ }
++ }
++ addResources(dataFolder);
++ } else {
++ //processor folder does not exist
++ dataFolder.mkdirs();
+ }
- addResources(dataFolder, unclosedFileName);
-
- } else {
- //processor folder does not exist
- dataFolder.mkdirs();
+ }
- }
- if (unclosedFileCount > 1) {
- throw new TsFileProcessorException(String
- .format("TsProcessor %s has more than one unclosed TsFile. please repair it",
- processorName));
- }
- if (unclosedFile == null) {
- unclosedFile = generateNewTsFilePath();
++ } catch (IOException e) {
++ throw new TsFileProcessorException(e);
+ }
+ buildInverseIndex();
- initCurrentTsFile(unclosedFile);
++ initCurrentTsFile(generateNewTsFilePath());
+ }
+
+ // add TsFiles in dataFolder to tsFileResources and update device inserted time map
- private void addResources(File dataFolder, String unclosedFileName)
++ private void addResources(File dataFolder)
+ throws TsFileProcessorException {
+ File[] tsFiles = dataFolder
+ .listFiles(x -> !x.getName().contains(RestorableTsFileIOWriter.RESTORE_SUFFIX)
+ && x.getName().split(EngineConstants.TSFILE_NAME_SEPARATOR).length == 2);
+ if (tsFiles == null || tsFiles.length == 0) {
+ return;
+ }
+ Arrays.sort(tsFiles, Comparator.comparingLong(x -> Long
+ .parseLong(x.getName().split(EngineConstants.TSFILE_NAME_SEPARATOR)[0])));
+
+ for (File tsfile : tsFiles) {
+ if (tsfile.getName().endsWith(StorageGroupProcessor.MERGE_TEMP_SUFFIX)) {
+ // remove temp file of last failed merge
+ //noinspection ResultOfMethodCallIgnored
+ tsfile.delete();
+ continue;
+ }
- if (!tsfile.getName().equals(unclosedFileName)) {
- addResource(tsfile);
- }
++ addResource(tsfile);
+ }
+ }
+
+ // add one TsFiles to tsFileResources and update device inserted time map
+ private void addResource(File tsfile) throws TsFileProcessorException {
+ //TODO we'd better define a file suffix for TsFile, e.g., .ts
+ String[] names = tsfile.getName().split(EngineConstants.TSFILE_NAME_SEPARATOR);
+ long time = Long.parseLong(names[0]);
+ if (fileNamePrefix < time) {
+ fileNamePrefix = time;
+ }
+ TsFileResource resource;
+ try {
+ resource = new TsFileResource(tsfile, true);
+ } catch (IOException e) {
+ throw new TsFileProcessorException(e);
+ }
+ tsFileResources.add(resource);
+ //maintain the inverse index and fileNamePrefix
+ for (String device : resource.getDevices()) {
+ lastFlushedTimeForEachDevice
+ .merge(device, resource.getEndTime(device), (x, y) -> x > y ? x : y);
+ }
+ }
+
- public void buildInverseIndex() {
++ private void buildInverseIndex() {
+ inverseIndexOfResource.clear();
+ for (TsFileResource resource : tsFileResources) {
+ for (String device : resource.getDevices()) {
+ inverseIndexOfResource.computeIfAbsent(device, k -> new ArrayList<>()).add(resource);
+ }
+ }
+ }
+
+
+ private File generateNewTsFilePath() throws TsFileProcessorException {
+ String dataDir = getNextDataFolder();
+ File dataFolder = new File(dataDir, processorName);
+ if (!dataFolder.exists()) {
+ if (!dataFolder.mkdirs()) {
+ throw new TsFileProcessorException(
+ String.format("Can not create TsFileProcess related folder: %s", dataFolder));
+ }
+ LOGGER.debug("The bufferwrite processor data dir doesn't exists, create new directory {}.",
+ dataFolder.getAbsolutePath());
+ }
+ String fileName = (fileNamePrefix + 1) + EngineConstants.TSFILE_NAME_SEPARATOR
+ + System.currentTimeMillis();
+ return new File(dataFolder, fileName);
+ }
+
+
+ private void initCurrentTsFile(File file) throws TsFileProcessorException {
+ this.insertFile = file;
+ try {
+ writer = new RestorableTsFileIOWriter(processorName, insertFile.getAbsolutePath());
+ currentResource = new TsFileResource(insertFile, writer);
+ } catch (IOException e) {
+ throw new TsFileProcessorException(e);
+ }
+
+ minWrittenTimeForEachDeviceInCurrentFile.clear();
+ maxWrittenTimeForEachDeviceInCurrentFile.clear();
+
+ }
+
+ protected boolean canWrite(String device, long timestamp) {
+ return !lastFlushedTimeForEachDevice.containsKey(device)
+ || timestamp > lastFlushedTimeForEachDevice.get(device);
+ }
+ /**
- * wrete a ts record into the memtable. If the memory usage is beyond the memThreshold, an async
++ * write a ts record into the memtable. If the memory usage is beyond the memThreshold, an async
+ * flushing operation will be called.
+ *
+ * @param plan data to be written
+ * @return OperationResult (WRITE_SUCCESS, WRITE_REJECT_BY_TIME, WRITE_IN_WARNING_MEM and
+ * WRITE_REJECT_BY_MEM)
+ * @throws TsFileProcessorException if a flushing operation occurs and failed.
+ */
+ public OperationResult insert(InsertPlan plan) throws TsFileProcessorException {
+ if (isClosed) {
+ reopen();
+ }
+ if (!canWrite(plan.getDeviceId(), plan.getTime())) {
+ return OperationResult.WRITE_REJECT_BY_TIME;
+ }
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
+ try {
- logNode.write(plan);
++ getLogNode().write(plan);
+ } catch (IOException e) {
+ throw new TsFileProcessorException(String.format("Fail to write wal for %s",
+ plan.toString()), e);
+ }
+ }
+ long memUsage = 0;
+ TSDataType type;
+ String measurement;
+ for (int i=0; i < plan.getMeasurements().length; i++){
+ measurement = plan.getMeasurements()[i];
+ type = fileSchemaRef.getMeasurementDataType(measurement);
+ memUsage += MemUtils.getPointSize(type, measurement);
+ }
+ UsageLevel level = BasicMemController.getInstance().acquireUsage(this, memUsage);
+ OperationResult result;
+ switch (level) {
+ case SAFE:
+ doInsert(plan);
+ checkMemThreshold4Flush(memUsage);
+ result = OperationResult.WRITE_SUCCESS;
+ break;
+ case WARNING:
+ if(LOGGER.isWarnEnabled()) {
+ LOGGER.warn("Memory usage will exceed warning threshold, current : {}.",
+ MemUtils.bytesCntToStr(BasicMemController.getInstance().getTotalUsage()));
+ }
+ doInsert(plan);
+ try {
+ flush();
+ } catch (IOException e) {
+ throw new TsFileProcessorException(e);
+ }
+ result = OperationResult.WRITE_IN_WARNING_MEM;
+ break;
+ case DANGEROUS:
+ if (LOGGER.isWarnEnabled()) {
+ LOGGER.warn("Memory usage will exceed dangerous threshold, current : {}.",
+ MemUtils.bytesCntToStr(BasicMemController.getInstance().getTotalUsage()));
+ }
+ result = OperationResult.WRITE_REJECT_BY_MEM;
+ break;
+ default:
+ result = OperationResult.WRITE_REJECT_BY_MEM;
+ }
+ return result;
+ }
+
+ private void doInsert(InsertPlan plan) throws TsFileProcessorException {
+ writeLog(plan);
+ String deviceId = plan.getDeviceId();
+ long time = plan.getTime();
+ TSDataType type;
+ String measurement;
+ for (int i = 0; i < plan.getMeasurements().length; i++) {
+ measurement = plan.getMeasurements()[i];
+ type = fileSchemaRef.getMeasurementDataType(measurement);
+ workMemTable.write(deviceId, measurement, type, time, plan.getValues()[i]);
+ }
+ if (!minWrittenTimeForEachDeviceInCurrentFile.containsKey(deviceId)) {
+ minWrittenTimeForEachDeviceInCurrentFile.put(deviceId, time);
+ }
+ if (!maxWrittenTimeForEachDeviceInCurrentFile.containsKey(deviceId)
+ || maxWrittenTimeForEachDeviceInCurrentFile.get(deviceId) < time) {
+ maxWrittenTimeForEachDeviceInCurrentFile.put(deviceId, time);
+ }
+ valueCount++;
+ }
+
+ public OperationResult update(UpdatePlan plan) {
+// String device = plan.getPath().getDevice();
+// String measurement = plan.getPath().getMeasurement();
+// List<Pair<Long, Long>> intervals = plan.getIntervals();
+ //TODO modify workMemtable, flushMemtable, and existing TsFiles
+ throw new UnsupportedOperationException("Update unimplemented!");
+ }
+
+ /**
+ * Delete data whose timestamp <= 'timestamp' and belonging to timeseries deviceId.measurementId.
+ * Delete data in both working MemTable and flushing MemTable.
+ *
+ * @param deviceId the deviceId of the timeseries to be deleted.
+ * @param measurementId the measurementId of the timeseries to be deleted.
+ * @param timestamp the upper-bound of deletion time.
+ */
+ public void delete(String deviceId, String measurementId, long timestamp) throws IOException {
+ workMemTable.delete(deviceId, measurementId, timestamp);
- if (maxWrittenTimeForEachDeviceInCurrentFile.containsKey(deviceId)
- && maxWrittenTimeForEachDeviceInCurrentFile.get(deviceId) < timestamp) {
- maxWrittenTimeForEachDeviceInCurrentFile
- .put(deviceId, lastFlushedTimeForEachDevice.getOrDefault(deviceId, 0L));
- }
+ boolean deleteFlushTable = false;
+ if (isFlush()) {
+ // flushing MemTable cannot be directly modified since another thread is reading it
+ flushMemTable = flushMemTable.copy();
+ deleteFlushTable = flushMemTable.delete(deviceId, measurementId, timestamp);
+ }
+ String fullPath = deviceId +
+ IoTDBConstant.PATH_SEPARATOR + measurementId;
+ Deletion deletion = new Deletion(fullPath, versionController.nextVersion(), timestamp);
+ if (deleteFlushTable || (currentResource.containsDevice(deviceId)
+ && currentResource.getStartTime(deviceId) <= timestamp)) {
+ currentResource.getModFile().write(deletion);
+ }
+ for (TsFileResource resource : tsFileResources) {
+ if (resource.containsDevice(deviceId) && resource.getStartTime(deviceId) <= timestamp) {
+ resource.getModFile().write(deletion);
+ }
+ }
- if (lastFlushedTimeForEachDevice.containsKey(deviceId)
- && lastFlushedTimeForEachDevice.get(deviceId) <= timestamp) {
- lastFlushedTimeForEachDevice.put(deviceId, 0L);
- }
+ }
+
+
+ private void checkMemThreshold4Flush(long addedMemory) throws TsFileProcessorException {
+ long newMem = memSize.addAndGet(addedMemory);
+ if (newMem > TSFileConfig.groupSizeInByte) {
+ if (LOGGER.isInfoEnabled()) {
+ String usageMem = MemUtils.bytesCntToStr(newMem);
+ String threshold = MemUtils.bytesCntToStr(TSFileConfig.groupSizeInByte);
+ LOGGER.info("The usage of memory {} in bufferwrite processor {} reaches the threshold {}",
+ usageMem, processorName, threshold);
+ }
+ try {
+ flush();
+ } catch (IOException e) {
+ LOGGER.error("Flush bufferwrite error.", e);
+ throw new TsFileProcessorException(e);
+ }
+ }
+ }
+
+
+ /**
+ * this method is for preparing a task task and then submitting it.
+ * @return
+ * @throws IOException
+ */
+ @Override
- public Future<Boolean> flush() throws IOException {
++ public Future<Boolean> flush() throws IOException, TsFileProcessorException {
+ if (isClosed) {
+ return null;
+ }
++
+ // waiting for the end of last flush operation.
+ try {
+ flushFuture.get();
+ } catch (InterruptedException | ExecutionException e) {
+ LOGGER.error(
- "Encounter an interrupt error when waitting for the flushing, the TsFile Processor is {}.",
++ "Encounter an interrupt error when waiting for the flushing, the TsFile Processor is {}.",
+ getProcessorName(), e);
+ Thread.currentThread().interrupt();
+ }
+ // statistic information for flush
+ if (valueCount <= 0) {
+ LOGGER.debug(
- "TsFile Processor {} has zero data to be flushed, will return directly.", processorName);
++ "TsFile Processor {} has no data to be flushed.", processorName);
+ flushFuture = new ImmediateFuture<>(true);
+ return flushFuture;
+ }
-
- if (LOGGER.isInfoEnabled()) {
- long thisFlushTime = System.currentTimeMillis();
- LOGGER.info(
- "The TsFile Processor {}: last flush time is {}, this flush time is {}, "
- + "flush time interval is {} s", getProcessorName(),
- DatetimeUtils.convertMillsecondToZonedDateTime(fileNamePrefix / 1000),
- DatetimeUtils.convertMillsecondToZonedDateTime(thisFlushTime),
- (thisFlushTime - fileNamePrefix / 1000) / 1000);
++ // statistic information for flush
++ if (lastFlushTime > 0) {
++ if (LOGGER.isInfoEnabled()) {
++ long thisFlushTime = System.currentTimeMillis();
++ LOGGER.info(
++ "The bufferwrite processor {}: last flush time is {}, this flush time is {}, "
++ + "flush time interval is {}s", getProcessorName(),
++ DatetimeUtils.convertMillsecondToZonedDateTime(lastFlushTime),
++ DatetimeUtils.convertMillsecondToZonedDateTime(thisFlushTime),
++ (thisFlushTime - lastFlushTime) / 1000.0);
++ }
+ }
- fileNamePrefix = System.nanoTime();
++ lastFlushTime = System.currentTimeMillis();
+
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
- logNode.notifyStartFlush();
++ getLogNode().notifyStartFlush();
+ }
++
+ valueCount = 0;
+ switchWorkToFlush();
+ long version = versionController.nextVersion();
+ BasicMemController.getInstance().releaseUsage(this, memSize.get());
+ memSize.set(0);
- // switch
+ flushFuture = FlushManager.getInstance().submit(() -> flushTask(version));
+ return flushFuture;
+ }
+
+ /**
+ * this method will be concurrent with other methods..
+ *
+ * @param version the operation version that will tagged on the to be flushed memtable (i.e.,
+ * ChunkGroup)
+ * @return true if successfully.
+ */
+ private boolean flushTask(long version) {
+ boolean result;
+ long flushStartTime = System.currentTimeMillis();
+ LOGGER.info("The TsFile Processor {} starts flushing.", processorName);
+ try {
+ if (flushMemTable != null && !flushMemTable.isEmpty()) {
+ // flush data
+ MemTableFlushUtil.flushMemTable(fileSchemaRef, writer, flushMemTable,
+ version);
+ // write restore information
+ writer.flush();
+ } else {
+ //no need to flush.
+ return true;
+ }
+
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
- logNode.notifyEndFlush(null);
++ getLogNode().notifyEndFlush(null);
+ }
+ result = true;
+ } catch (Exception e) {
+ LOGGER.error(
+ "The TsFile Processor {} failed to flush.",
+ processorName, e);
+ result = false;
+ } finally {
+ try {
+ switchFlushToWork();
+ } catch (TsFileProcessorException e) {
+ LOGGER.error(e.getMessage());
+ result = false;
+ }
+ LOGGER.info("The TsFile Processor {} ends flushing.", processorName);
+ }
+ if (LOGGER.isInfoEnabled()) {
+ long flushEndTime = System.currentTimeMillis();
+ LOGGER.info(
+ "The TsFile Processor {} flush, start time is {}, flush end time is {}, "
+ + "flush time consumption is {}ms",
+ processorName,
+ DatetimeUtils.convertMillsecondToZonedDateTime(flushStartTime),
+ DatetimeUtils.convertMillsecondToZonedDateTime(flushEndTime),
+ flushEndTime - flushStartTime);
+ }
+ return result;
+ }
+
+ private void switchWorkToFlush() {
+ flushQueryLock.lock();
+ try {
- if (flushMemTable == null) {
- flushMemTable = workMemTable;
- for (String device : flushMemTable.getMemTableMap().keySet()) {
- lastFlushedTimeForEachDevice.put(device, maxWrittenTimeForEachDeviceInCurrentFile.get(device));
- }
- workMemTable = new PrimitiveMemTable();
- }
++ IMemTable temp = flushMemTable == null ? new PrimitiveMemTable() : flushMemTable;
++ flushMemTable = workMemTable;
++ workMemTable = temp;
++ isFlush = true;
+ } finally {
+ flushQueryLock.unlock();
+ }
+ }
+
+ private void switchFlushToWork() throws TsFileProcessorException {
+ flushQueryLock.lock();
+ try {
+ //we update the index of currentTsResource.
+ for (String device : flushMemTable.getMemTableMap().keySet()) {
+ currentResource.setStartTime(device, minWrittenTimeForEachDeviceInCurrentFile.get(device));
+ // new end time must be larger than the old one.
+ currentResource.setEndTime(device, lastFlushedTimeForEachDevice.get(device));
+ }
+ flushMemTable.clear();
- flushMemTable = null;
- //make chunk groups in this flush task visble
++ //make chunk groups in this flush task visible
+ writer.appendMetadata();
+ if (needCloseCurrentFile()) {
+ closeCurrentTsFileAndOpenNewOne();
+ }
+ } finally {
+ flushQueryLock.unlock();
+ }
+ }
+
+ @Override
+ public boolean canBeClosed() {
- return true;
++ return flushFuture == null || flushFuture.isDone();
+ }
+
+
+ /**
+ * this method do not call flush() to flush data in memory to disk.
+ * @throws TsFileProcessorException
+ */
+ private void closeCurrentTsFileAndOpenNewOne() throws TsFileProcessorException {
+ closeCurrentFile();
+ initCurrentTsFile(generateNewTsFilePath());
+ }
+
+ //very dangerous, how to make sure this function is thread safe (no other functions are running)
+ private void closeCurrentFile() throws TsFileProcessorException {
+ try {
+ flush().get();
+ long closeStartTime = System.currentTimeMillis();
+ // end file
+ if (writer.getChunkGroupMetaDatas().isEmpty()){
+ //this is an empty TsFile, we do not need to save it...
+ writer.endFile(fileSchemaRef);
+ Files.delete(Paths.get(insertFile.getAbsolutePath()));
+ } else {
+ writer.endFile(fileSchemaRef);
+
+ tsFileResources.add(currentResource);
+ //maintain the inverse index
+ for (String device : currentResource.getDevices()) {
+ inverseIndexOfResource.computeIfAbsent(device, k -> new ArrayList<>())
+ .add(currentResource);
+ }
+ }
+ writer = null;
+
+ // delete the restore for this bufferwrite processor
+ if (LOGGER.isInfoEnabled()) {
+ long closeEndTime = System.currentTimeMillis();
+ LOGGER.info(
+ "Close current TsFile {}, start time is {}, end time is {}, time consumption is {}ms",
+ insertFile.getAbsolutePath(),
+ DatetimeUtils.convertMillsecondToZonedDateTime(closeStartTime),
+ DatetimeUtils.convertMillsecondToZonedDateTime(closeEndTime),
+ closeEndTime - closeStartTime);
+ }
+
+ workMemTable.clear();
+ if (flushMemTable != null) {
+ flushMemTable.clear();
+ }
+ if (logNode != null) {
+ logNode.close();
+ }
+
+ } catch (IOException e) {
+ LOGGER.error("Close the bufferwrite processor error, the bufferwrite is {}.",
+ getProcessorName(), e);
+ throw new TsFileProcessorException(e);
+ } catch (Exception e) {
+ LOGGER
+ .error("Failed to close the bufferwrite processor when calling the action function.", e);
+ throw new TsFileProcessorException(e);
+ }
+ }
+
+ @Override
+ public long memoryUsage() {
+ return memSize.get();
+ }
+
+ /**
+ * check if is flushing.
+ *
+ * @return True if flushing
+ */
+ public boolean isFlush() {
- // starting a flush task has two steps: set the flushMemtable, and then set the flushFuture
- // So, the following case exists: flushMemtable != null but flushFuture is done (because the
- // flushFuture refers to the last finished flush.
- // And, the following case exists,too: flushMemtable == null, but flushFuture is not done.
- // (flushTask() is not finished, but switchToWork() has done)
- // So, checking flushMemTable is more meaningful than flushFuture.isDone().
- return flushMemTable != null;
++ return isFlush;
+ }
+
+
+ /**
+ * query data.
+ */
+ public SeriesDataSource query(SingleSeriesExpression expression,
+ QueryContext context) throws IOException {
+ MeasurementSchema mSchema;
+ TSDataType dataType;
+
+ String deviceId = expression.getSeriesPath().getDevice();
+ String measurementId = expression.getSeriesPath().getMeasurement();
+
+ mSchema = fileSchemaRef.getMeasurementSchema(measurementId);
+ dataType = mSchema.getType();
+
- // tsfile dataØØ
- //TODO in the old version, tsfile is deep copied. I do not know why
++ // sealed files
+ List<TsFileResource> dataFiles = new ArrayList<>();
+ tsFileResources.forEach(k -> {
+ if (k.containsDevice(deviceId)) {
+ dataFiles.add(k);
+ }
+ });
- // bufferwrite data
- //TODO unsealedTsFile class is a little redundant.
- UnsealedTsFile unsealedTsFile = null;
+
++ // unsealed file
++ UnsealedTsFile unsealedTsFile = null;
+ if (currentResource.getStartTime(deviceId) >= 0) {
+ unsealedTsFile = new UnsealedTsFile();
+ unsealedTsFile.setFilePath(currentResource.getFile().getAbsolutePath());
+ List<ChunkMetaData> chunks = writer.getMetadatas(deviceId, measurementId, dataType);
+ List<Modification> pathModifications = context.getPathModifications(
+ currentResource.getModFile(), deviceId + IoTDBConstant.PATH_SEPARATOR + measurementId);
+ if (!pathModifications.isEmpty()) {
+ QueryUtils.modifyChunkMetaData(chunks, pathModifications);
+ }
-
+ unsealedTsFile.setTimeSeriesChunkMetaDatas(chunks);
+ }
+ return new SeriesDataSource(
+ new Path(deviceId + IoTDBConstant.PATH_SEPARATOR + measurementId), dataFiles,
+ unsealedTsFile, queryDataInMemtable(deviceId, measurementId, dataType, mSchema.getProps()));
+ }
+
+ /**
+ * get the one (or two) chunk(s) in the memtable ( and the other one in flushing status and then
+ * compact them into one TimeValuePairSorter). Then get its (or their) ChunkMetadata(s).
+ *
+ * @param deviceId device id
+ * @param measurementId sensor id
+ * @param dataType data type
+ * @return corresponding chunk data in memory
+ */
+ private ReadOnlyMemChunk queryDataInMemtable(String deviceId,
+ String measurementId, TSDataType dataType, Map<String, String> props) {
++ // lock to avoid that the mem tables are being switched by the flush thread
+ flushQueryLock.lock();
+ try {
+ MemSeriesLazyMerger memSeriesLazyMerger = new MemSeriesLazyMerger();
+ if (flushMemTable != null) {
+ memSeriesLazyMerger.addMemSeries(flushMemTable.query(deviceId, measurementId, dataType, props));
+ }
+ memSeriesLazyMerger.addMemSeries(workMemTable.query(deviceId, measurementId, dataType, props));
+ // memSeriesLazyMerger has handled the props,
+ // so we do not need to handle it again in the following readOnlyMemChunk
+ return new ReadOnlyMemChunk(dataType, memSeriesLazyMerger, Collections.emptyMap());
+ } finally {
+ flushQueryLock.unlock();
+ }
+ }
+
-
- /**
- * used for test. We can know when the flush() is called.
- *
- * @return the last flush() time. Time unit: nanosecond.
- */
- public long getFileNamePrefix() {
- return fileNamePrefix;
- }
-
- /**
- * used for test. We can block to wait for finishing flushing.
- *
- * @return the future of the flush() task.
- */
- public Future<Boolean> getFlushFuture() {
- return flushFuture;
- }
-
+ @Override
+ public void close() throws TsFileProcessorException {
+ if (isClosed) {
+ return;
+ }
+ closeCurrentFile();
+ try {
+ if (currentResource != null) {
+ currentResource.close();
+ }
+ for (TsFileResource resource : tsFileResources) {
+ resource.close();
+ }
+
+ } catch (IOException e) {
+ throw new TsFileProcessorException(e);
+ }
+ isClosed = true;
+ }
+
+ /**
+ * remove all data of this processor. Used For UT
+ */
+ void removeMe() throws TsFileProcessorException, IOException {
+ try {
+ flushFuture.get(10000, TimeUnit.MILLISECONDS);
+ } catch (ExecutionException | TimeoutException e) {
+ LOGGER.error("can not end running flush task in 10 seconds: {}", e.getMessage());
+ } catch (InterruptedException e) {
+ LOGGER.error("current running flush task is interrupted.", e);
+ Thread.currentThread().interrupt();
+ }
+ close();
+ for (String folder : Directories.getInstance().getAllTsFileFolders()) {
+ File dataFolder = new File(folder, processorName);
+ File[] files;
+ if (dataFolder.exists() && (files = dataFolder.listFiles()) != null) {
+ for (File file: files) {
+ Files.deleteIfExists(Paths.get(file.getAbsolutePath()));
+ }
+ }
+ }
+ }
+
+ /**
- * Check if the currentTsFileResource is toooo large.
++ * Check if the currentTsFileResource is too large.
+ * @return true if the file is too large.
+ */
+ private boolean needCloseCurrentFile() {
+ IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ long fileSize = currentResource.getFile().length();
+ if (fileSize >= config.getBufferwriteFileSizeThreshold()) {
+ if (LOGGER.isInfoEnabled()) {
+ LOGGER.info(
+ "The bufferwrite processor {}, size({}) of the file {} reaches threshold {}.",
+ processorName, MemUtils.bytesCntToStr(fileSize), currentResource.getFilePath(),
+ MemUtils.bytesCntToStr(config.getBufferwriteFileSizeThreshold()));
+ }
+ return true;
+ }
+ return false;
+ }
+
+ protected List<String> getAllDataFolders() {
+ return Directories.getInstance().getAllTsFileFolders();
+ }
+
+ protected String getNextDataFolder() {
+ return Directories.getInstance().getNextFolderForTsfile();
+ }
+
+ protected String getLogSuffix() {
+ return IoTDBConstant.BUFFERWRITE_LOG_NODE_SUFFIX;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return this == o;
+ }
+
+ @Override
+ public int hashCode() {
+ return super.hashCode();
+ }
+
+ public boolean isClosed() {
+ return isClosed;
+ }
+
+ private void writeLog(InsertPlan plan)
+ throws TsFileProcessorException {
+ try {
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
- logNode.write(plan);
++ getLogNode().write(plan);
+ }
+ } catch (IOException e) {
+ throw new TsFileProcessorException(e);
+ }
+ }
+
+ public long getLastInsertTime(String deviceId) {
+ // generally speaking, currentTime > historicalTime, unless currentTime does not exist
+ long historicalTime = lastFlushedTimeForEachDevice.getOrDefault(deviceId, -1L);
+ long currentTime = maxWrittenTimeForEachDeviceInCurrentFile.getOrDefault(deviceId, -1L);
+ return Long.max(historicalTime, currentTime);
+ }
+
+ /**
+ * Append a new TsFile for the sync module. The appended file can only contain new data w.r.t the
+ * existing files.
+ * @param tsFileResource
+ */
+ public void appendFile(TsFileResource tsFileResource) {
+ this.tsFileResources.add(tsFileResource);
+ for (Entry<String, Long> entry : tsFileResource.getEndTimeMap().entrySet()) {
+ lastFlushedTimeForEachDevice.put(entry.getKey(), entry.getValue());
+ maxWrittenTimeForEachDeviceInCurrentFile.put(entry.getKey(), entry.getValue());
+ // the new file may contain new time series
+ if (!minWrittenTimeForEachDeviceInCurrentFile.containsKey(entry.getKey())) {
+ minWrittenTimeForEachDeviceInCurrentFile.put(entry.getKey(), entry.getValue());
+ }
+ inverseIndexOfResource.computeIfAbsent(entry.getKey(),
+ k -> new ArrayList<>()).add(tsFileResource);
+ }
+ }
+
+ /**
+ * get overlap tsfiles which are conflict with the appendFile.
+ *
+ * @param appendFile the appended tsfile information
+ */
+ public List<String> getOverlapFiles(TsFileResource appendFile, String uuid)
+ throws TsFileProcessorException {
+ List<String> overlapFiles = new ArrayList<>();
+ try {
+ for (TsFileResource tsFileResource : tsFileResources) {
+ getOverlapFile(appendFile, tsFileResource, uuid, overlapFiles);
+ }
+ } catch (IOException e) {
+ throw new TsFileProcessorException(String.format("Failed to get tsfiles "
+ + "which overlap with the appendFile: %s.", appendFile.getFilePath()), e);
+ }
+ return overlapFiles;
+ }
+
+ private void getOverlapFile(TsFileResource appendFile, TsFileResource tsFileResource,
+ String uuid, List<String> overlapFiles) throws IOException {
+ if (!tsFileResource.overlaps(appendFile)) {
+ return;
+ }
+ // create a link to the overlapped file to avoid modifying the original file link
+ File newFile = SyncUtils.linkFile(uuid, tsFileResource);
+ overlapFiles.add(newFile.getPath());
+ }
+
+ /**
+ * the total size of all TsFiles.
+ * @return
+ */
+ public long totalFileSize() {
+ long fileSize = 0;
+ for (TsFileResource resource : tsFileResources) {
+ fileSize += resource.getFile().length();
+ }
+ if (currentResource != null) {
+ fileSize += currentResource.getFile().length();
+ }
+ return fileSize;
+ }
+
+ public long currentFileSize() {
+ return currentResource != null ? currentResource.getFile().length() : 0;
+ }
+
+ public List<TsFileResource> getTsFileResources() {
+ return tsFileResources;
+ }
+
+ /**
+ * Merge method, replaces 'oldFiles' in tsfileResources with 'newFile'. If 'newFile' is null, just
+ * remove oldFiles.
+ * @param oldFiles
+ * @param newFile
+ */
+ public void replaceFiles(List<TsFileResource> oldFiles, TsFileResource newFile) {
+ List<TsFileResource> newFiles = new ArrayList<>();
+ int j = 0;
+ for (TsFileResource origin : tsFileResources) {
+ TsFileResource toDelete = j < oldFiles.size() ? oldFiles.get(j) : null;
+ if (origin == toDelete) {
+ if (j == 0 && newFile != null) {
+ // replace the first old file with the new file
+ newFiles.add(newFile);
+ }
+ // this file should be deleted, do not add it to new files
+ j++;
+ } else {
+ // this file is not one of the new files, keep it
+ newFiles.add(origin);
+ }
+ }
+ tsFileResources = newFiles;
+ }
+
+ public long getMetaSize() {
+ return writer.getMetaSize();
+ }
+}
diff --cc iotdb/src/main/java/org/apache/iotdb/db/exception/ConsistencyException.java
index bb7c359,6f4fe91..108d6a7
--- a/iotdb/src/main/java/org/apache/iotdb/db/exception/ConsistencyException.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/exception/ConsistencyException.java
@@@ -16,17 -16,26 +16,22 @@@
* specific language governing permissions and limitations
* under the License.
*/
- package org.apache.iotdb.db.exception.qp;
+ package org.apache.iotdb.db.exception;
--/**
- * This exception is thrown while meeting error in parsing ast tree to generate logical operator.
- * @author liukun
- *
-- */
- public class IllegalASTFormatException extends QueryProcessorException {
-public class ProcessorRuntimException extends RuntimeException {
-
- private static final long serialVersionUID = -5543549255867713835L;
++public class ConsistencyException extends Exception {
+
- public ProcessorRuntimException() {
- super();
++ public ConsistencyException() {
+ }
- private static final long serialVersionUID = -8987915911329315588L;
- public ProcessorRuntimException(String message) {
++ public ConsistencyException(String message) {
+ super(message);
+ }
- public IllegalASTFormatException(String msg) {
- super(msg);
- public ProcessorRuntimException(Throwable cause) {
- super(cause);
++ public ConsistencyException(String message, Throwable cause) {
++ super(message, cause);
}
++ public ConsistencyException(Throwable cause) {
++ super(cause);
++ }
}
diff --cc iotdb/src/main/java/org/apache/iotdb/db/qp/executor/IQueryProcessExecutor.java
index 0000000,6377f44..cdae6b1
mode 000000,100644..100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/IQueryProcessExecutor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/IQueryProcessExecutor.java
@@@ -1,0 -1,146 +1,134 @@@
+ /**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+ package org.apache.iotdb.db.qp.executor;
+
+ import java.io.IOException;
+ import java.util.List;
+ import java.util.Map;
-import org.apache.iotdb.db.exception.FileNodeManagerException;
+ import org.apache.iotdb.db.exception.PathErrorException;
+ import org.apache.iotdb.db.exception.ProcessorException;
++import org.apache.iotdb.db.exception.StorageGroupManagerException;
+ import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+ import org.apache.iotdb.db.query.context.QueryContext;
+ import org.apache.iotdb.db.query.fill.IFill;
+ import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
+ 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;
+ import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+ import org.apache.iotdb.tsfile.utils.Pair;
+
+ public interface IQueryProcessExecutor {
+
+ /**
+ * Process Non-Query Physical plan, including insert/update/delete operation of
+ * data/metadata/Privilege
+ *
+ * @param plan Physical Non-Query Plan
+ */
+ boolean processNonQuery(PhysicalPlan plan) throws ProcessorException;
+
+ /**
+ * process query plan of qp layer, construct queryDataSet.
+ *
+ * @param queryPlan QueryPlan
+ * @return QueryDataSet
+ */
+ QueryDataSet processQuery(PhysicalPlan queryPlan, QueryContext context)
- throws IOException, FileNodeManagerException, PathErrorException,
- QueryFilterOptimizationException, ProcessorException;
++ throws IOException, PathErrorException,
++ QueryFilterOptimizationException, ProcessorException, StorageGroupManagerException;
+
+ /**
+ * process aggregate plan of qp layer, construct queryDataSet.
+ */
+ QueryDataSet aggregate(List<Path> paths, List<String> aggres, IExpression expression,
+ QueryContext context)
- throws ProcessorException, IOException, PathErrorException, FileNodeManagerException, QueryFilterOptimizationException;
++ throws ProcessorException, IOException, PathErrorException, QueryFilterOptimizationException, StorageGroupManagerException;
+
+ /**
+ * process group by plan of qp layer, construct queryDataSet.
+ */
+ QueryDataSet groupBy(List<Path> paths, List<String> aggres, IExpression expression,
+ long unit, long origin, List<Pair<Long, Long>> intervals, QueryContext context)
- throws ProcessorException, IOException, PathErrorException, FileNodeManagerException, QueryFilterOptimizationException;
++ throws ProcessorException, IOException, PathErrorException, QueryFilterOptimizationException, StorageGroupManagerException;
+
+ /**
+ * process fill plan of qp layer, construct queryDataSet.
+ */
+ QueryDataSet fill(List<Path> fillPaths, long queryTime, Map<TSDataType, IFill> fillTypes,
+ QueryContext context)
- throws ProcessorException, IOException, PathErrorException, FileNodeManagerException;
++ throws ProcessorException, IOException, PathErrorException, StorageGroupManagerException;
+
+ /**
+ * execute update command and return whether the operator is successful.
+ *
+ * @param path : update series seriesPath
+ * @param startTime start time in update command
+ * @param endTime end time in update command
+ * @param value - in type of string
+ * @return - whether the operator is successful.
+ */
+ boolean update(Path path, long startTime, long endTime, String value)
+ throws ProcessorException;
+
+ /**
+ * execute delete command and return whether the operator is successful.
+ *
+ * @param paths : delete series paths
+ * @param deleteTime end time in delete command
+ * @return - whether the operator is successful.
+ */
+ boolean delete(List<Path> paths, long deleteTime) throws ProcessorException;
+
+ /**
+ * execute delete command and return whether the operator is successful.
+ *
+ * @param path : delete series seriesPath
+ * @param deleteTime end time in delete command
+ * @return - whether the operator is successful.
+ */
+ boolean delete(Path path, long deleteTime) throws ProcessorException;
+
+ /**
+ * insert a single value. Only used in test
+ *
+ * @param path seriesPath to be inserted
+ * @param insertTime - it's time point but not a range
+ * @param value value to be inserted
+ * @return - Operate Type.
+ */
+ int insert(Path path, long insertTime, String value) throws ProcessorException;
+
- /**
- * execute insert command and return whether the operator is successful.
- *
- * @param deviceId deviceId to be inserted
- * @param insertTime - it's time point but not a range
- * @param measurementList measurements to be inserted
- * @param insertValues values to be inserted
- * @return - Operate Type.
- */
- int multiInsert(String deviceId, long insertTime, String[] measurementList,
- String[] insertValues) throws ProcessorException;
-
+ boolean judgePathExists(Path fullPath);
+
+ /**
+ * Get data type of series
+ */
+ TSDataType getSeriesType(Path path) throws PathErrorException;
+
+ /**
+ * Get all paths of a full path
+ */
+ List<String> getAllPaths(String originPath) throws PathErrorException;
+
+ int getFetchSize();
+
+ void setFetchSize(int fetchSize);
+
+ }
diff --cc iotdb/src/main/java/org/apache/iotdb/db/qp/executor/OverflowQPExecutor.java
index f3ae094,e772bf5..8e2227d
--- a/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/OverflowQPExecutor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/OverflowQPExecutor.java
@@@ -60,10 -66,15 +67,13 @@@ import org.apache.iotdb.tsfile.exceptio
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.Field;
import org.apache.iotdb.tsfile.read.common.Path;
+ import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.apache.iotdb.tsfile.read.expression.IExpression;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+ import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.Pair;
-import org.apache.iotdb.tsfile.write.record.TSRecord;
-import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --cc iotdb/src/main/java/org/apache/iotdb/db/qp/executor/QueryProcessExecutor.java
index 0428801,94939e9..2eebab6
--- a/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/QueryProcessExecutor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/QueryProcessExecutor.java
@@@ -32,36 -31,40 +32,48 @@@ import org.apache.iotdb.db.qp.physical.
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.InsertPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
+ import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.executor.EngineQueryRouter;
+ import org.apache.iotdb.db.query.executor.IEngineQueryRouter;
+import org.apache.iotdb.db.query.fill.IFill;
import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
+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;
import org.apache.iotdb.tsfile.read.expression.QueryExpression;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.utils.Pair;
- public abstract class QueryProcessExecutor {
+ public abstract class QueryProcessExecutor implements IQueryProcessExecutor {
protected ThreadLocal<Integer> fetchSize = new ThreadLocal<>();
- protected EngineQueryRouter queryRouter = new EngineQueryRouter();
- protected IEngineQueryRouter queryRouter = new EngineQueryRouter();
++ IEngineQueryRouter queryRouter = new EngineQueryRouter();
+
- public QueryProcessExecutor() {
++ protected QueryProcessExecutor() {
+ }
- /**
- * process query plan of qp layer, construct queryDataSet.
- * @param queryPlan QueryPlan
- * @return QueryDataSet
- */
- public QueryDataSet processQuery(QueryPlan queryPlan, QueryContext context)
- throws IOException, StorageGroupManagerException, PathErrorException,
- QueryFilterOptimizationException, ProcessorException {
+ @Override
+ public QueryDataSet processQuery(PhysicalPlan queryPlan, QueryContext context)
- throws IOException, FileNodeManagerException, PathErrorException,
- QueryFilterOptimizationException, ProcessorException {
++ throws IOException, PathErrorException,
++ QueryFilterOptimizationException, ProcessorException, StorageGroupManagerException {
+
+ if (queryPlan instanceof QueryPlan) {
+ return processDataQuery((QueryPlan) queryPlan, context);
+ } else if (queryPlan instanceof AuthorPlan) {
+ return processAuthorQuery((AuthorPlan) queryPlan, context);
+ } else {
+ throw new ProcessorException(String.format("Unrecognized query plan %s", queryPlan));
+ }
+ }
+
+ protected abstract QueryDataSet processAuthorQuery(AuthorPlan plan, QueryContext context)
+ throws ProcessorException;
+ private QueryDataSet processDataQuery(QueryPlan queryPlan, QueryContext context)
- throws FileNodeManagerException, QueryFilterOptimizationException, PathErrorException, ProcessorException, IOException {
++ throws StorageGroupManagerException, QueryFilterOptimizationException, PathErrorException, ProcessorException, IOException {
QueryExpression queryExpression = QueryExpression.create().setSelectSeries(queryPlan.getPaths())
.setExpression(queryPlan.getExpression());
if (queryPlan instanceof GroupByPlan) {
@@@ -103,38 -100,7 +109,39 @@@
this.fetchSize.set(fetchSize);
}
+ public abstract QueryDataSet aggregate(List<Path> paths, List<String> aggres,
+ IExpression expression, QueryContext context) throws ProcessorException, IOException,
+ PathErrorException, StorageGroupManagerException, QueryFilterOptimizationException;
+
+ public abstract QueryDataSet groupBy(List<Path> paths, List<String> aggres,
+ IExpression expression, long unit, long origin, List<Pair<Long, Long>> intervals,
+ QueryContext context) throws ProcessorException, IOException, PathErrorException,
+ StorageGroupManagerException, QueryFilterOptimizationException;
+
+ public abstract QueryDataSet fill(List<Path> fillPaths, long queryTime, Map<TSDataType,
- IFill> fillTypes, QueryContext context)
++ IFill> fillTypes, QueryContext context)
+ throws ProcessorException, IOException, PathErrorException, StorageGroupManagerException;
+
+ /**
+ * executeWithGlobalTimeFilter update command and return whether the operator is successful.
+ *
+ * @param path : update series seriesPath
+ * @param startTime start time in update command
+ * @param endTime end time in update command
+ * @param value - in type of string
+ * @return - whether the operator is successful.
+ */
+ public abstract boolean update(Path path, long startTime, long endTime, String value)
+ throws ProcessorException;
+
+ /**
+ * executeWithGlobalTimeFilter delete command and return whether the operator is successful.
+ *
+ * @param paths : delete series paths
+ * @param deleteTime end time in delete command
+ * @return - whether the operator is successful.
+ */
+ @Override
public boolean delete(List<Path> paths, long deleteTime) throws ProcessorException {
try {
boolean result = true;
@@@ -152,8 -118,8 +159,7 @@@
String.format("TimeSeries %s does not exist and cannot be delete its data", onePath));
}
}
-- List<String> fullPath = new ArrayList<>();
-- fullPath.addAll(pathSet);
++ List<String> fullPath = new ArrayList<>(pathSet);
for (String path : fullPath) {
result &= delete(new Path(path), deleteTime);
}
@@@ -163,22 -129,4 +169,22 @@@
}
}
+ /**
+ * executeWithGlobalTimeFilter delete command and return whether the operator is successful.
+ *
+ * @param path : delete series seriesPath
+ * @param deleteTime end time in delete command
+ * @return - whether the operator is successful.
+ */
- protected abstract boolean delete(Path path, long deleteTime) throws ProcessorException;
++ public abstract boolean delete(Path path, long deleteTime) throws ProcessorException;
+
+ /**
+ * execute insert command and return whether the operator is successful.
+ *
+ * @param plan the InsertPlan
+ */
+ public abstract void insert(InsertPlan plan) throws ProcessorException;
+
+ public abstract List<String> getAllPaths(String originPath) throws PathErrorException;
+
}
diff --cc iotdb/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
index dde5a9d,cdb519c..78ce829
--- a/iotdb/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
@@@ -111,6 -108,6 +112,11 @@@ public class InsertPlan extends Physica
}
@Override
++ public int hashCode() {
++ return super.hashCode();
++ }
++
++ @Override
public boolean equals(Object o) {
if (this == o) {
return true;
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java
index ac4cd62,c5f029e..7f0338e
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java
@@@ -138,12 -142,29 +138,29 @@@ public class QueryResourceManager
}
}
+ /**
+ * Begin query and set query tokens of all filter paths in expression. This method is used in
+ * filter calculation.
+ * @param remoteDeviceIdSet device id set which can not handle locally
+ * Note : the method is for cluster
+ */
+ public void beginQueryOfGivenExpression(long jobId, IExpression expression,
- Set<String> remoteDeviceIdSet) throws FileNodeManagerException {
++ Set<String> remoteDeviceIdSet) throws StorageGroupManagerException {
+ Set<String> deviceIdSet = new HashSet<>();
+ getUniquePaths(expression, deviceIdSet);
+ deviceIdSet.removeAll(remoteDeviceIdSet);
+ for (String deviceId : deviceIdSet) {
+ putQueryTokenForCurrentRequestThread(jobId, deviceId,
- FileNodeManager.getInstance().beginQuery(deviceId));
++ DatabaseEngineFactory.getCurrent().beginQuery(deviceId));
+ }
+ }
+
public QueryDataSource getQueryDataSource(Path selectedPath,
QueryContext context)
- throws FileNodeManagerException {
+ throws StorageGroupManagerException {
SingleSeriesExpression singleSeriesExpression = new SingleSeriesExpression(selectedPath, null);
- QueryDataSource queryDataSource = FileNodeManager.getInstance()
+ QueryDataSource queryDataSource = DatabaseEngineFactory.getCurrent()
.query(singleSeriesExpression, context);
// add used files to current thread request cached map
@@@ -161,12 -182,12 +178,14 @@@
// no resource need to be released.
return;
}
- for (Map.Entry<String, List<Integer>> entry : queryTokensMap.get(jobId).entrySet()) {
- for (int token : entry.getValue()) {
- FileNodeManager.getInstance().endQuery(entry.getKey(), token);
++
+ for (Map.Entry<String, List<Integer>> entry : queryTokensMap.get(jobId).entrySet()) {
+ for (int token : entry.getValue()) {
+ DatabaseEngineFactory.getCurrent().endQuery(entry.getKey(), token);
+ }
}
- queryTokensMap.remove(jobId);
- }
++
+ queryTokensMap.remove(jobId);
// remove usage of opened file paths of current thread
filePathsManager.removeUsedFilesForGivenJob(jobId);
}
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/executor/AbstractExecutorWithoutTimeGenerator.java
index 0000000,f5f07c8..eda4678
mode 000000,100644..100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/executor/AbstractExecutorWithoutTimeGenerator.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/executor/AbstractExecutorWithoutTimeGenerator.java
@@@ -1,0 -1,84 +1,86 @@@
+ /**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+ package org.apache.iotdb.db.query.executor;
+
+ import java.io.IOException;
+ import java.util.List;
-import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
-import org.apache.iotdb.db.exception.FileNodeManagerException;
++import org.apache.iotdb.db.engine.datasource.QueryDataSource;
+ import org.apache.iotdb.db.exception.PathErrorException;
++import org.apache.iotdb.db.exception.StorageGroupManagerException;
+ import org.apache.iotdb.db.metadata.MManager;
+ import org.apache.iotdb.db.query.context.QueryContext;
+ import org.apache.iotdb.db.query.control.QueryResourceManager;
+ import org.apache.iotdb.db.query.factory.SeriesReaderFactory;
+ import org.apache.iotdb.db.query.reader.AllDataReader;
+ import org.apache.iotdb.db.query.reader.IPointReader;
+ import org.apache.iotdb.db.query.reader.merge.PriorityMergeReader;
+ import org.apache.iotdb.db.query.reader.sequence.SequenceDataReader;
+ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+ import org.apache.iotdb.tsfile.read.common.Path;
+ import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+
+ /**
+ * Query executor with global time filter.
+ */
-public abstract class AbstractExecutorWithoutTimeGenerator {
++abstract class AbstractExecutorWithoutTimeGenerator {
+
+ /**
+ * Create reader of a series
+ *
+ * @param context query context
+ * @param path series path
+ * @param dataTypes list of data type
+ * @param timeFilter time filter
+ * @return reader of the series
+ */
- public static IPointReader createSeriesReader(QueryContext context, Path path,
++ IPointReader createSeriesReader(QueryContext context, Path path,
+ List<TSDataType> dataTypes, Filter timeFilter)
- throws FileNodeManagerException {
++ throws StorageGroupManagerException {
+ QueryDataSource queryDataSource = QueryResourceManager.getInstance().getQueryDataSource(path,
+ context);
++
+ // add data type
+ try {
+ dataTypes.add(MManager.getInstance().getSeriesType(path.getFullPath()));
+ } catch (PathErrorException e) {
- throw new FileNodeManagerException(e);
++ throw new StorageGroupManagerException(e);
+ }
+
+ // sequence reader for one sealed tsfile
+ SequenceDataReader tsFilesReader;
+ try {
+ tsFilesReader = new SequenceDataReader(queryDataSource.getSeqDataSource(),
+ timeFilter, context);
+ } catch (IOException e) {
- throw new FileNodeManagerException(e);
++ throw new StorageGroupManagerException(e);
+ }
+
+ // unseq reader for all chunk groups in unSeqFile
+ PriorityMergeReader unSeqMergeReader;
+ try {
+ unSeqMergeReader = SeriesReaderFactory.getInstance()
- .createUnSeqMergeReader(queryDataSource.getOverflowSeriesDataSource(), timeFilter);
++ .createUnSeqMergeReader(queryDataSource.getOverflowSeriesDataSource(),
++ context, timeFilter);
+ } catch (IOException e) {
- throw new FileNodeManagerException(e);
++ throw new StorageGroupManagerException(e);
+ }
+ // merge sequence data with unsequence data.
+ return new AllDataReader(tsFilesReader, unSeqMergeReader);
+ }
+ }
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/executor/EngineExecutorWithoutTimeGenerator.java
index cdb286c,24a9e62..dee9d1d
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/executor/EngineExecutorWithoutTimeGenerator.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/executor/EngineExecutorWithoutTimeGenerator.java
@@@ -42,78 -34,26 +41,47 @@@ import org.apache.iotdb.tsfile.read.fil
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
/**
- * IoTDB query executor with global time filter.
+ * IoTDB query executor of Stand-alone version with global time filter.
*/
- public class EngineExecutorWithoutTimeGenerator {
-public class EngineExecutorWithoutTimeGenerator extends AbstractExecutorWithoutTimeGenerator {
++class EngineExecutorWithoutTimeGenerator extends AbstractExecutorWithoutTimeGenerator {
private QueryExpression queryExpression;
-- public EngineExecutorWithoutTimeGenerator(QueryExpression queryExpression) {
++ EngineExecutorWithoutTimeGenerator(QueryExpression queryExpression) {
this.queryExpression = queryExpression;
}
/**
- * with global time filter.
+ * without filter or with global time filter.
*/
- public QueryDataSet executeWithGlobalTimeFilter(QueryContext context)
- public QueryDataSet execute(QueryContext context)
- throws FileNodeManagerException {
++ QueryDataSet executeWithGlobalTimeFilter(QueryContext context)
+ throws StorageGroupManagerException {
- Filter timeFilter = null;
- if (queryExpression.hasQueryFilter()) {
- timeFilter = ((GlobalTimeExpression) queryExpression.getExpression()).getFilter();
+ Filter timeFilter = ((GlobalTimeExpression) queryExpression.getExpression()).getFilter();
+
+ List<IPointReader> readersOfSelectedSeries = new ArrayList<>();
+ List<TSDataType> dataTypes = new ArrayList<>();
+
+ QueryResourceManager.getInstance()
+ .beginQueryOfGivenQueryPaths(context.getJobId(), queryExpression.getSelectedSeries());
+
+ for (Path path : queryExpression.getSelectedSeries()) {
-
- QueryDataSource queryDataSource = QueryResourceManager.getInstance().getQueryDataSource(path,
- context);
-
- // add data type
- try {
- dataTypes.add(MManager.getInstance().getSeriesType(path.getFullPath()));
- } catch (PathErrorException e) {
- throw new StorageGroupManagerException(e);
- }
-
- // sequence reader for one sealed tsfile
- SequenceDataReader tsFilesReader;
- try {
- tsFilesReader = new SequenceDataReader(queryDataSource.getSeqDataSource(),
- timeFilter, context);
- } catch (IOException e) {
- throw new StorageGroupManagerException(e);
- }
-
- // unseq reader for all chunk groups in unSeqFile
- PriorityMergeReader unSeqMergeReader;
- try {
- unSeqMergeReader = SeriesReaderFactory.getInstance()
- .createUnSeqMergeReader(queryDataSource.getOverflowSeriesDataSource(),
- context, timeFilter);
- } catch (IOException e) {
- throw new StorageGroupManagerException(e);
- }
-
- // merge sequence data with unsequence data.
- readersOfSelectedSeries.add(new AllDataReader(tsFilesReader, unSeqMergeReader));
++ readersOfSelectedSeries.add(createSeriesReader(context, path, dataTypes, timeFilter));
+ }
+
+ try {
+ return new EngineDataSetWithoutTimeGenerator(queryExpression.getSelectedSeries(), dataTypes,
+ readersOfSelectedSeries);
+ } catch (IOException e) {
+ throw new StorageGroupManagerException(e);
}
+ }
+
+ /**
+ * without filter.
+ */
- public QueryDataSet executeWithoutFilter(QueryContext context)
++ QueryDataSet executeWithoutFilter(QueryContext context)
+ throws StorageGroupManagerException {
List<IPointReader> readersOfSelectedSeries = new ArrayList<>();
List<TSDataType> dataTypes = new ArrayList<>();
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/executor/EngineQueryRouter.java
index c808b0b,03c600d..52e0b1a
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/executor/EngineQueryRouter.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/executor/EngineQueryRouter.java
@@@ -23,15 -23,13 +23,13 @@@ import java.io.IOException
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
- import org.apache.iotdb.db.engine.tsfiledata.TsFileProcessor;
- import org.apache.iotdb.db.exception.StorageGroupManagerException;
-import org.apache.iotdb.db.exception.FileNodeManagerException;
import org.apache.iotdb.db.exception.PathErrorException;
import org.apache.iotdb.db.exception.ProcessorException;
++import org.apache.iotdb.db.exception.StorageGroupManagerException;
import org.apache.iotdb.db.query.context.QueryContext;
- import org.apache.iotdb.db.query.executor.groupby.GroupByWithOnlyTimeFilterDataSetDataSet;
- import org.apache.iotdb.db.query.executor.groupby.GroupByWithValueFilterDataSetDataSet;
+ import org.apache.iotdb.db.query.dataset.groupby.GroupByWithOnlyTimeFilterDataSet;
+ import org.apache.iotdb.db.query.dataset.groupby.GroupByWithValueFilterDataSet;
import org.apache.iotdb.db.query.fill.IFill;
- import org.apache.iotdb.tsfile.exception.NotImplementedException;
import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Path;
@@@ -49,15 -47,11 +47,11 @@@ import org.apache.iotdb.tsfile.utils.Pa
* Query entrance class of IoTDB query process. All query clause will be transformed to physical
* plan, physical plan will be executed by EngineQueryRouter.
*/
- public class EngineQueryRouter {
-
-
+ public class EngineQueryRouter implements IEngineQueryRouter{
- /**
- * execute physical plan.
- */
+ @Override
public QueryDataSet query(QueryExpression queryExpression, QueryContext context)
- throws FileNodeManagerException {
+ throws StorageGroupManagerException {
if (queryExpression.hasQueryFilter()) {
try {
@@@ -85,12 -79,10 +79,10 @@@
}
}
- /**
- * execute aggregation query.
- */
+ @Override
public QueryDataSet aggregate(List<Path> selectedSeries, List<String> aggres,
IExpression expression, QueryContext context) throws QueryFilterOptimizationException,
- FileNodeManagerException, IOException, PathErrorException, ProcessorException {
+ StorageGroupManagerException, IOException, PathErrorException, ProcessorException {
if (expression != null) {
IExpression optimizedExpression = ExpressionOptimizer.getInstance()
@@@ -179,16 -161,10 +161,10 @@@
}
}
- /**
- * execute fill query.
- *
- * @param fillPaths select path list
- * @param queryTime timestamp
- * @param fillType type IFill map
- */
+ @Override
public QueryDataSet fill(List<Path> fillPaths, long queryTime, Map<TSDataType, IFill> fillType,
QueryContext context)
- throws FileNodeManagerException, PathErrorException, IOException {
+ throws StorageGroupManagerException, PathErrorException, IOException {
long nextJobId = context.getJobId();
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/executor/IEngineQueryRouter.java
index 0000000,01c1aed..ee73047
mode 000000,100644..100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/executor/IEngineQueryRouter.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/executor/IEngineQueryRouter.java
@@@ -1,0 -1,78 +1,78 @@@
+ /**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+ package org.apache.iotdb.db.query.executor;
+
+ import java.io.IOException;
+ import java.util.List;
+ import java.util.Map;
-import org.apache.iotdb.db.exception.FileNodeManagerException;
+ import org.apache.iotdb.db.exception.PathErrorException;
+ import org.apache.iotdb.db.exception.ProcessorException;
++import org.apache.iotdb.db.exception.StorageGroupManagerException;
+ import org.apache.iotdb.db.query.context.QueryContext;
+ import org.apache.iotdb.db.query.fill.IFill;
+ import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
+ 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;
+ import org.apache.iotdb.tsfile.read.expression.QueryExpression;
+ import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+ import org.apache.iotdb.tsfile.utils.Pair;
+
+ public interface IEngineQueryRouter {
+
+ /**
+ * Execute physical plan.
+ */
+ QueryDataSet query(QueryExpression queryExpression, QueryContext context)
- throws FileNodeManagerException, PathErrorException;
++ throws PathErrorException, StorageGroupManagerException;
+
+ /**
+ * Execute aggregation query.
+ */
+ QueryDataSet aggregate(List<Path> selectedSeries, List<String> aggres,
+ IExpression expression, QueryContext context)
- throws QueryFilterOptimizationException, FileNodeManagerException, IOException, PathErrorException, ProcessorException;
++ throws QueryFilterOptimizationException, IOException, PathErrorException, ProcessorException, StorageGroupManagerException;
+
+ /**
+ * Execute groupBy query.
+ *
+ * @param selectedSeries select path list
+ * @param aggres aggregation name list
+ * @param expression filter expression
+ * @param unit time granularity for interval partitioning, unit is ms.
+ * @param origin the datum time point for interval division is divided into a time interval for
+ * each TimeUnit time from this point forward and backward.
+ * @param intervals time intervals, closed interval.
+ */
+ QueryDataSet groupBy(List<Path> selectedSeries, List<String> aggres,
+ IExpression expression, long unit, long origin, List<Pair<Long, Long>> intervals,
+ QueryContext context)
- throws ProcessorException, QueryFilterOptimizationException, FileNodeManagerException,
- PathErrorException, IOException;
++ throws ProcessorException, QueryFilterOptimizationException,
++ PathErrorException, IOException, StorageGroupManagerException;
+ /**
+ * Execute fill query.
+ *
+ * @param fillPaths select path list
+ * @param queryTime timestamp
+ * @param fillType type IFill map
+ */
+ QueryDataSet fill(List<Path> fillPaths, long queryTime, Map<TSDataType, IFill> fillType,
- QueryContext context) throws FileNodeManagerException, PathErrorException, IOException;
++ QueryContext context) throws PathErrorException, IOException, StorageGroupManagerException;
+
+ }
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/timegenerator/AbstractNodeConstructor.java
index 4369c82,5d83314..3d93e26
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/timegenerator/AbstractNodeConstructor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/timegenerator/AbstractNodeConstructor.java
@@@ -21,11 -20,10 +20,10 @@@ package org.apache.iotdb.db.query.timeg
import static org.apache.iotdb.tsfile.read.expression.ExpressionType.AND;
import static org.apache.iotdb.tsfile.read.expression.ExpressionType.OR;
- import static org.apache.iotdb.tsfile.read.expression.ExpressionType.SERIES;
import java.io.IOException;
-import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
-import org.apache.iotdb.db.exception.FileNodeManagerException;
+import org.apache.iotdb.db.engine.datasource.QueryDataSource;
+import org.apache.iotdb.db.exception.StorageGroupManagerException;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.QueryResourceManager;
import org.apache.iotdb.db.query.factory.SeriesReaderFactory;
@@@ -54,39 -51,40 +51,39 @@@ public abstract class AbstractNodeConst
* @param expression expression
* @return Node object
* @throws IOException IOException
- * @throws FileNodeManagerException FileNodeManagerException
+ * @throws StorageGroupManagerException StorageGroupManagerException
*/
- public Node construct(IExpression expression, QueryContext context)
+ public abstract Node construct(IExpression expression, QueryContext context)
- throws FileNodeManagerException;
++ throws StorageGroupManagerException;
+
+ /**
+ * Construct not series type node.
+ *
+ * @param expression expression
+ * @return Node object
- * @throws FileNodeManagerException FileNodeManagerException
++ * @throws StorageGroupManagerException StorageGroupManagerException
+ */
- protected Node constructNotSeriesNode(IExpression expression, QueryContext context)
- throws FileNodeManagerException {
++ Node constructNotSeriesNode(IExpression expression, QueryContext context)
+ throws StorageGroupManagerException {
- if (expression.getType() == SERIES) {
- try {
- return new EngineLeafNode(generateSeriesReader((SingleSeriesExpression) expression,
- context));
- } catch (IOException e) {
- throw new StorageGroupManagerException(e);
- }
+ Node leftChild;
+ Node rightChild;
+ if (expression.getType() == OR) {
+ leftChild = this.construct(((IBinaryExpression) expression).getLeft(), context);
+ rightChild = this.construct(((IBinaryExpression) expression).getRight(), context);
+ return new OrNode(leftChild, rightChild);
+ } else if (expression.getType() == AND) {
+ leftChild = this.construct(((IBinaryExpression) expression).getLeft(), context);
+ rightChild = this.construct(((IBinaryExpression) expression).getRight(), context);
+ return new AndNode(leftChild, rightChild);
} else {
- Node leftChild;
- Node rightChild;
- if (expression.getType() == OR) {
- leftChild = this.construct(((IBinaryExpression) expression).getLeft(), context);
- rightChild = this.construct(((IBinaryExpression) expression).getRight(), context);
- return new OrNode(leftChild, rightChild);
- } else if (expression.getType() == AND) {
- leftChild = this.construct(((IBinaryExpression) expression).getLeft(), context);
- rightChild = this.construct(((IBinaryExpression) expression).getRight(), context);
- return new AndNode(leftChild, rightChild);
- } else {
- throw new UnSupportedDataTypeException(
- "Unsupported QueryFilterType when construct OperatorNode: " + expression.getType());
- }
+ throw new UnSupportedDataTypeException(
+ "Unsupported QueryFilterType when construct OperatorNode: " + expression.getType());
}
}
- private IReader generateSeriesReader(SingleSeriesExpression singleSeriesExpression,
- protected IReader generateSeriesReader(SingleSeriesExpression singleSeriesExpression,
++ IReader generateSeriesReader(SingleSeriesExpression singleSeriesExpression,
QueryContext context)
- throws IOException, FileNodeManagerException {
-
+ throws IOException, StorageGroupManagerException {
-
QueryDataSource queryDataSource = QueryResourceManager.getInstance().getQueryDataSource(
singleSeriesExpression.getSeriesPath(), context);
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/timegenerator/EngineNodeConstructor.java
index 4369c82,45e6ac6..dc75d26
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/timegenerator/EngineNodeConstructor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/timegenerator/EngineNodeConstructor.java
@@@ -22,30 -22,15 +22,16 @@@ package org.apache.iotdb.db.query.timeg
import static org.apache.iotdb.tsfile.read.expression.ExpressionType.SERIES;
import java.io.IOException;
- import org.apache.iotdb.db.engine.datasource.QueryDataSource;
-import org.apache.iotdb.db.exception.FileNodeManagerException;
+import org.apache.iotdb.db.exception.StorageGroupManagerException;
import org.apache.iotdb.db.query.context.QueryContext;
- import org.apache.iotdb.db.query.control.QueryResourceManager;
- import org.apache.iotdb.db.query.factory.SeriesReaderFactory;
- import org.apache.iotdb.db.query.reader.AllDataReader;
- import org.apache.iotdb.db.query.reader.IReader;
- import org.apache.iotdb.db.query.reader.merge.PriorityMergeReader;
- import org.apache.iotdb.db.query.reader.sequence.SequenceDataReader;
- import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
- import org.apache.iotdb.tsfile.read.expression.IBinaryExpression;
import org.apache.iotdb.tsfile.read.expression.IExpression;
import org.apache.iotdb.tsfile.read.expression.impl.SingleSeriesExpression;
- import org.apache.iotdb.tsfile.read.filter.basic.Filter;
- import org.apache.iotdb.tsfile.read.query.timegenerator.node.AndNode;
import org.apache.iotdb.tsfile.read.query.timegenerator.node.Node;
- import org.apache.iotdb.tsfile.read.query.timegenerator.node.OrNode;
- public class EngineNodeConstructor {
+ public class EngineNodeConstructor extends AbstractNodeConstructor {
-
-- public EngineNodeConstructor() {
++ EngineNodeConstructor() {
++ // do nothing
}
/**
@@@ -54,59 -39,20 +40,20 @@@
* @param expression expression
* @return Node object
* @throws IOException IOException
- * @throws FileNodeManagerException FileNodeManagerException
+ * @throws StorageGroupManagerException StorageGroupManagerException
*/
+ @Override
public Node construct(IExpression expression, QueryContext context)
- throws FileNodeManagerException {
+ throws StorageGroupManagerException {
if (expression.getType() == SERIES) {
try {
return new EngineLeafNode(generateSeriesReader((SingleSeriesExpression) expression,
context));
} catch (IOException e) {
- throw new FileNodeManagerException(e);
+ throw new StorageGroupManagerException(e);
}
} else {
- Node leftChild;
- Node rightChild;
- if (expression.getType() == OR) {
- leftChild = this.construct(((IBinaryExpression) expression).getLeft(), context);
- rightChild = this.construct(((IBinaryExpression) expression).getRight(), context);
- return new OrNode(leftChild, rightChild);
- } else if (expression.getType() == AND) {
- leftChild = this.construct(((IBinaryExpression) expression).getLeft(), context);
- rightChild = this.construct(((IBinaryExpression) expression).getRight(), context);
- return new AndNode(leftChild, rightChild);
- } else {
- throw new UnSupportedDataTypeException(
- "Unsupported QueryFilterType when construct OperatorNode: " + expression.getType());
- }
- }
- }
-
- private IReader generateSeriesReader(SingleSeriesExpression singleSeriesExpression,
- QueryContext context)
- throws IOException, StorageGroupManagerException {
-
- QueryDataSource queryDataSource = QueryResourceManager.getInstance().getQueryDataSource(
- singleSeriesExpression.getSeriesPath(), context);
-
- Filter filter = singleSeriesExpression.getFilter();
-
- // reader for all sequence data
- SequenceDataReader tsFilesReader = new SequenceDataReader(queryDataSource.getSeqDataSource(),
- filter, context);
-
- // reader for all unSequence data
- PriorityMergeReader unSeqMergeReader = SeriesReaderFactory.getInstance()
- .createUnSeqMergeReader(queryDataSource.getOverflowSeriesDataSource(), context, filter);
-
- if (!tsFilesReader.hasNext()) {
- //only have unsequence data.
- return unSeqMergeReader;
- } else {
- //merge sequence data with unsequence data.
- return new AllDataReader(tsFilesReader, unSeqMergeReader);
+ return constructNotSeriesNode(expression, context);
}
}
-
}
diff --cc iotdb/src/main/java/org/apache/iotdb/db/service/IoTDB.java
index 8902a55,a7cf296..1d89812
--- a/iotdb/src/main/java/org/apache/iotdb/db/service/IoTDB.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/service/IoTDB.java
@@@ -19,19 -19,19 +19,15 @@@
package org.apache.iotdb.db.service;
import java.io.IOException;
--import java.util.List;
import org.apache.iotdb.db.concurrent.IoTDBDefaultThreadExceptionHandler;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.engine.filenode.FileNodeManager;
+import org.apache.iotdb.db.engine.DatabaseEngineFactory;
import org.apache.iotdb.db.engine.memcontrol.BasicMemController;
- import org.apache.iotdb.db.exception.StorageGroupManagerException;
-import org.apache.iotdb.db.exception.FileNodeManagerException;
--import org.apache.iotdb.db.exception.PathErrorException;
import org.apache.iotdb.db.exception.RecoverException;
import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.exception.builder.ExceptionBuilder;
--import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.monitor.StatMonitor;
import org.apache.iotdb.db.sync.receiver.SyncServiceManager;
import org.apache.iotdb.db.writelog.manager.MultiFileLogNodeManager;
@@@ -46,7 -46,7 +42,7 @@@ public class IoTDB implements IoTDBMBea
IoTDBConstant.JMX_TYPE, "IoTDB");
private RegisterManager registerManager = new RegisterManager();
-- public static final IoTDB getInstance() {
++ public static IoTDB getInstance() {
return IoTDBHolder.INSTANCE;
}
@@@ -113,7 -112,7 +109,7 @@@
LOGGER.info("IoTDB is set up.");
}
-- public void deactivate() {
++ private void deactivate() {
LOGGER.info("Deactivating IoTDB...");
registerManager.deregisterAll();
JMXService.deregisterMBean(mbeanName);
@@@ -121,7 -120,7 +117,7 @@@
}
@Override
- public void stop() throws StorageGroupManagerException {
- public void stop() throws FileNodeManagerException {
++ public void stop() {
deactivate();
}
diff --cc iotdb/src/main/java/org/apache/iotdb/db/service/JDBCServiceEventHandler.java
index b47a166,b47a166..83dc3aa
--- a/iotdb/src/main/java/org/apache/iotdb/db/service/JDBCServiceEventHandler.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/service/JDBCServiceEventHandler.java
@@@ -47,11 -47,11 +47,7 @@@ public class JDBCServiceEventHandler im
@Override
public void deleteContext(ServerContext arg0, TProtocol arg1, TProtocol arg2) {
-- try {
-- serviceImpl.handleClientExit();
-- } catch (TException e) {
-- LOGGER.error("failed to clear client status", e);
-- }
++ serviceImpl.handleClientExit();
}
@Override
diff --cc iotdb/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index 18e1008,c502f46..47d9174
--- a/iotdb/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@@ -34,20 -39,20 +39,18 @@@ import org.apache.iotdb.db.auth.authori
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.engine.filenode.FileNodeManager;
+import org.apache.iotdb.db.engine.DatabaseEngineFactory;
import org.apache.iotdb.db.exception.ArgsErrorException;
- import org.apache.iotdb.db.exception.StorageGroupManagerException;
-import org.apache.iotdb.db.exception.FileNodeManagerException;
++import org.apache.iotdb.db.exception.ConsistencyException;
import org.apache.iotdb.db.exception.PathErrorException;
import org.apache.iotdb.db.exception.ProcessorException;
--import org.apache.iotdb.db.exception.qp.IllegalASTFormatException;
++import org.apache.iotdb.db.exception.StorageGroupManagerException;
import org.apache.iotdb.db.exception.qp.QueryProcessorException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.metadata.Metadata;
import org.apache.iotdb.db.qp.QueryProcessor;
import org.apache.iotdb.db.qp.executor.OverflowQPExecutor;
--import org.apache.iotdb.db.qp.logical.Operator;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
--import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.QueryResourceManager;
@@@ -94,21 -100,21 +98,22 @@@ import org.slf4j.LoggerFactory
public class TSServiceImpl implements TSIService.Iface, ServerContext {
private static final Logger LOGGER = LoggerFactory.getLogger(TSServiceImpl.class);
- protected static final String INFO_NOT_LOGIN = "{}: Not login.";
- protected static final String ERROR_NOT_LOGIN = "Not login";
+ private static final String INFO_NOT_LOGIN = "{}: Not login.";
+ private static final String ERROR_NOT_LOGIN = "Not login";
++ private static final String ERROR_QUERY_IN_BATCH = "statement is query";
- private QueryProcessor processor = new QueryProcessor(new OverflowQPExecutor());
+ protected QueryProcessor processor;
// Record the username for every rpc connection. Username.get() is null if
// login is failed.
- private ThreadLocal<String> username = new ThreadLocal<>();
+ protected ThreadLocal<String> username = new ThreadLocal<>();
- protected ThreadLocal<HashMap<String, PhysicalPlan>> queryStatus = new ThreadLocal<>();
- protected ThreadLocal<HashMap<String, QueryDataSet>> queryRet = new ThreadLocal<>();
- protected ThreadLocal<ZoneId> zoneIds = new ThreadLocal<>();
+ private ThreadLocal<HashMap<String, PhysicalPlan>> queryStatus = new ThreadLocal<>();
+ private ThreadLocal<HashMap<String, QueryDataSet>> queryRet = new ThreadLocal<>();
+ private ThreadLocal<ZoneId> zoneIds = new ThreadLocal<>();
private IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
- protected ThreadLocal<Map<Long, QueryContext>> contextMapLocal = new ThreadLocal<>();
+ private ThreadLocal<Map<Long, QueryContext>> contextMapLocal = new ThreadLocal<>();
-- public TSServiceImpl() throws IOException {
- // do nothing because there is no need
++ public TSServiceImpl() {
+ processor = new QueryProcessor(new OverflowQPExecutor());
}
@Override
@@@ -158,7 -164,7 +163,7 @@@
}
@Override
-- public TSCloseSessionResp closeSession(TSCloseSessionReq req) throws TException {
++ public TSCloseSessionResp closeSession(TSCloseSessionReq req) {
LOGGER.info("{}: receive close session", IoTDBConstant.GLOBAL_DB_NAME);
TS_Status tsStatus;
if (username.get() == null) {
@@@ -178,20 -184,20 +183,20 @@@
}
@Override
-- public TSCancelOperationResp cancelOperation(TSCancelOperationReq req) throws TException {
++ public TSCancelOperationResp cancelOperation(TSCancelOperationReq req) {
return new TSCancelOperationResp(new TS_Status(TS_StatusCode.SUCCESS_STATUS));
}
@Override
-- public TSCloseOperationResp closeOperation(TSCloseOperationReq req) throws TException {
++ public TSCloseOperationResp closeOperation(TSCloseOperationReq req) {
LOGGER.info("{}: receive close operation", IoTDBConstant.GLOBAL_DB_NAME);
try {
releaseQueryResource(req);
clearAllStatusForCurrentRequest();
- } catch (Exception e) {
+ } catch (StorageGroupManagerException e) {
- LOGGER.error("Error in closeOperation : {}", e.getMessage());
+ LOGGER.error("Error in closeOperation : ", e);
}
return new TSCloseOperationResp(new TS_Status(TS_StatusCode.SUCCESS_STATUS));
}
@@@ -228,7 -235,7 +234,7 @@@
}
@Override
-- public TSFetchMetadataResp fetchMetadata(TSFetchMetadataReq req) throws TException {
++ public TSFetchMetadataResp fetchMetadata(TSFetchMetadataReq req) {
TS_Status status;
if (!checkLogin()) {
LOGGER.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME);
@@@ -236,130 -243,163 +242,97 @@@
return new TSFetchMetadataResp(status);
}
TSFetchMetadataResp resp = new TSFetchMetadataResp();
-- switch (req.getType()) {
-- case "SHOW_TIMESERIES":
-- String path = req.getColumnPath();
-- try {
- List<List<String>> showTimeseriesList = MManager.getInstance()
- .getShowTimeseriesPath(path);
++ try {
++ switch (req.getType()) {
++ case "SHOW_TIMESERIES":
++ String path = req.getColumnPath();
+ List<List<String>> showTimeseriesList = getTimeSeriesForPath(path);
resp.setShowTimeseriesList(showTimeseriesList);
- } catch (PathErrorException e) {
- } catch (PathErrorException | InterruptedException | ProcessorException e) {
-- status = getErrorStatus(
-- String.format("Failed to fetch timeseries %s's metadata because: %s",
-- req.getColumnPath(), e));
-- resp.setStatus(status);
-- return resp;
-- } catch (OutOfMemoryError outOfMemoryError) { // TODO OOME
-- LOGGER
-- .error(String.format("Failed to fetch timeseries %s's metadata", req.getColumnPath()),
-- outOfMemoryError);
-- status = getErrorStatus(
-- String.format("Failed to fetch timeseries %s's metadata because: %s",
-- req.getColumnPath(), outOfMemoryError));
-- resp.setStatus(status);
-- return resp;
-- }
-- status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
-- break;
-- case "SHOW_STORAGE_GROUP":
-- try {
- Set<String> storageGroups = MManager.getInstance().getAllStorageGroup();
++ status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
++ break;
++ case "SHOW_STORAGE_GROUP":
+ Set<String> storageGroups = getAllStorageGroups();
resp.setShowStorageGroups(storageGroups);
- } catch (PathErrorException e) {
- } catch (PathErrorException | InterruptedException e) {
-- status = getErrorStatus(
-- String.format("Failed to fetch storage groups' metadata because: %s", e));
-- resp.setStatus(status);
-- return resp;
-- } catch (OutOfMemoryError outOfMemoryError) { // TODO OOME
-- LOGGER.error("Failed to fetch storage groups' metadata", outOfMemoryError);
-- status = getErrorStatus(
-- String.format("Failed to fetch storage groups' metadata because: %s",
-- outOfMemoryError));
-- resp.setStatus(status);
-- return resp;
-- }
-- status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
-- break;
-- case "METADATA_IN_JSON":
- String metadataInJson = null;
- try {
- metadataInJson = MManager.getInstance().getMetadataInString();
- } catch (OutOfMemoryError outOfMemoryError) { // TODO OOME
- LOGGER.error("Failed to fetch all metadata in json", outOfMemoryError);
- status = getErrorStatus(
- String.format("Failed to fetch all metadata in json because: %s", outOfMemoryError));
- resp.setStatus(status);
- return resp;
- }
- resp.setMetadataInJson(metadataInJson);
- status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
- break;
- case "DELTA_OBEJECT":
- Metadata metadata;
- String metadataInJson;
-- try {
++ status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
++ break;
++ case "METADATA_IN_JSON":
++ String metadataInJson;
+ metadataInJson = getMetadataInString();
- } catch (PathErrorException | InterruptedException | ProcessorException e) {
- status = getErrorStatus(
- String.format("Failed to fetch all metadata in json because: %s", e));
- resp.setStatus(status);
- return resp;
- } catch (OutOfMemoryError outOfMemoryError) { // TODO OOME
- LOGGER.error("Failed to fetch all metadata in json", outOfMemoryError);
- status = getErrorStatus(
- String.format("Failed to fetch all metadata in json because: %s", outOfMemoryError));
- resp.setStatus(status);
- return resp;
- }
- resp.setMetadataInJson(metadataInJson);
- status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
- break;
- case "DELTA_OBEJECT":
- Metadata metadata;
- try {
++ resp.setMetadataInJson(metadataInJson);
++ status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
++ break;
++ case "DELTA_OBJECT":
++ Metadata metadata;
String column = req.getColumnPath();
- metadata = MManager.getInstance().getMetadata();
+ metadata = getMetadata();
Map<String, List<String>> deviceMap = metadata.getDeviceMap();
if (deviceMap == null || !deviceMap.containsKey(column)) {
resp.setColumnsList(new ArrayList<>());
} else {
resp.setColumnsList(deviceMap.get(column));
}
- } catch (PathErrorException e) {
- } catch (PathErrorException | InterruptedException | ProcessorException e) {
-- LOGGER.error("cannot get delta object map", e);
-- status = getErrorStatus(String.format("Failed to fetch delta object map because: %s", e));
-- resp.setStatus(status);
-- return resp;
-- } catch (OutOfMemoryError outOfMemoryError) { // TODO OOME
-- LOGGER.error("Failed to get delta object map", outOfMemoryError);
-- status = getErrorStatus(
-- String.format("Failed to get delta object map because: %s", outOfMemoryError));
++ status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
break;
-- }
-- status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
-- break;
-- case "COLUMN":
- try {
- resp.setDataType(MManager.getInstance().getSeriesType(req.getColumnPath()).toString());
- } catch (PathErrorException e) {
- // TODO aggregate seriesPath e.g. last(root.ln.wf01.wt01.status)
- // status = new TS_Status(TS_StatusCode.ERROR_STATUS);
- // status.setErrorMessage(String.format("Failed to fetch %s's data type because: %s",
- // req.getColumnPath(), e));
- // resp.setStatus(status);
- // return resp;
- }
- status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
- break;
- case "ALL_COLUMNS":
-- try {
- resp.setColumnsList(MManager.getInstance().getPaths(req.getColumnPath()));
- } catch (PathErrorException e) {
- status = getErrorStatus(String
- .format("Failed to fetch %s's all columns because: %s", req.getColumnPath(), e));
- resp.setStatus(status);
- return resp;
- } catch (OutOfMemoryError outOfMemoryError) { // TODO OOME
- LOGGER.error("Failed to fetch seriesPath {}'s all columns", req.getColumnPath(),
- outOfMemoryError);
- status = getErrorStatus(String.format("Failed to fetch %s's all columns because: %s",
- req.getColumnPath(), outOfMemoryError));
++ case "COLUMN":
+ resp.setDataType(getSeriesType(req.getColumnPath()).toString());
- } catch (PathErrorException | InterruptedException | ProcessorException e) {
- // TODO aggregate seriesPath e.g. last(root.ln.wf01.wt01.status)
- // status = new TS_Status(TS_StatusCode.ERROR_STATUS);
- // status.setErrorMessage(String.format("Failed to fetch %s's data type because: %s",
- // req.getColumnPath(), e));
- // resp.setStatus(status);
- // return resp;
- }
- status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
- break;
- case "ALL_COLUMNS":
- try {
++ status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
+ break;
- }
- status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
- break;
- default:
- status = new TS_Status(TS_StatusCode.ERROR_STATUS);
- status
- .setErrorMessage(String.format("Unsuport fetch metadata operation %s", req.getType()));
- break;
++ case "ALL_COLUMNS":
+ resp.setColumnsList(getPaths(req.getColumnPath()));
- } catch (PathErrorException | InterruptedException | ProcessorException e) {
- status = getErrorStatus(String
- .format("Failed to fetch %s's all columns because: %s", req.getColumnPath(), e));
- resp.setStatus(status);
- return resp;
- } catch (OutOfMemoryError outOfMemoryError) { // TODO OOME
- LOGGER.error("Failed to fetch seriesPath {}'s all columns", req.getColumnPath(),
- outOfMemoryError);
- status = getErrorStatus(String.format("Failed to fetch %s's all columns because: %s",
- req.getColumnPath(), outOfMemoryError));
++ status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
+ break;
- }
- status = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
- break;
- default:
- status = new TS_Status(TS_StatusCode.ERROR_STATUS);
- status
- .setErrorMessage(String.format("Unsuport fetch metadata operation %s", req.getType()));
- break;
++ default:
++ status = new TS_Status(TS_StatusCode.ERROR_STATUS);
++ status
++ .setErrorMessage(String.format("Unsupported fetch metadata operation %s",
++ req.getType()));
++ break;
++ }
++ } catch (PathErrorException e) {
++ LOGGER.error("cannot get metadata for {} ", req.getType(), e);
++ status = getErrorStatus(String.format("cannot get metadata for %s because: %s",
++ req.getType(), e));
++ resp.setStatus(status);
++ return resp;
++ } catch (OutOfMemoryError outOfMemoryError) { // TODO OOME
++ LOGGER.error("Failed to get delta object map", outOfMemoryError);
++ status = getErrorStatus(
++ String.format("Failed to get delta object map because: %s", outOfMemoryError));
}
resp.setStatus(status);
return resp;
}
- protected Set<String> getAllStorageGroups() throws PathErrorException, InterruptedException {
++ private Set<String> getAllStorageGroups() throws PathErrorException {
+ return MManager.getInstance().getAllStorageGroup();
+ }
+
- protected List<List<String>> getTimeSeriesForPath(String path)
- throws PathErrorException, InterruptedException, ProcessorException {
++ private List<List<String>> getTimeSeriesForPath(String path)
++ throws PathErrorException {
+ return MManager.getInstance().getShowTimeseriesPath(path);
+ }
+
- protected String getMetadataInString()
- throws InterruptedException, PathErrorException, ProcessorException {
++ private String getMetadataInString() {
+ return MManager.getInstance().getMetadataInString();
+ }
+
+ protected Metadata getMetadata()
- throws PathErrorException, InterruptedException, ProcessorException {
++ throws PathErrorException {
+ return MManager.getInstance().getMetadata();
+ }
+
+ protected TSDataType getSeriesType(String path)
- throws PathErrorException, InterruptedException, ProcessorException {
++ throws PathErrorException {
+ return MManager.getInstance().getSeriesType(path);
+ }
+
+ protected List<String> getPaths(String path)
- throws PathErrorException, InterruptedException, ProcessorException {
++ throws PathErrorException {
+ return MManager.getInstance().getPaths(path);
+ }
+
/**
- * Judge whether the statement is ADMIN COMMAND and if true, executeWithGlobalTimeFilter it.
+ * Judge whether the statement is ADMIN COMMAND and if true, execute it.
*
* @param statement command
* @return true if the statement is ADMIN COMMAND
@@@ -396,8 -436,8 +369,7 @@@
}
@Override
-- public TSExecuteBatchStatementResp executeBatchStatement(TSExecuteBatchStatementReq req)
-- throws TException {
++ public TSExecuteBatchStatementResp executeBatchStatement(TSExecuteBatchStatementReq req) {
try {
if (!checkLogin()) {
LOGGER.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME);
@@@ -409,30 -449,31 +381,13 @@@
String batchErrorMessage = "";
for (String statement : statements) {
-- try {
-- PhysicalPlan physicalPlan = processor.parseSQLToPhysicalPlan(statement, zoneIds.get());
-- physicalPlan.setProposer(username.get());
-- if (physicalPlan.isQuery()) {
-- return getTSBathExecuteStatementResp(TS_StatusCode.ERROR_STATUS,
-- "statement is query :" + statement, result);
-- }
-- TSExecuteStatementResp resp = executeUpdateStatement(physicalPlan);
-- if (resp.getStatus().getStatusCode().equals(TS_StatusCode.SUCCESS_STATUS)) {
-- result.add(Statement.SUCCESS_NO_INFO);
-- } else {
-- result.add(Statement.EXECUTE_FAILED);
-- isAllSuccessful = false;
-- batchErrorMessage = resp.getStatus().getErrorMessage();
-- }
-- } catch (Exception e) {
-- String errMessage = String.format(
- "Fail to generate physcial plan and executeWithGlobalTimeFilter for statement "
- "Fail to generate physcial plan and execute for statement "
-- + "%s beacuse %s",
-- statement, e.getMessage());
- LOGGER.warn("Error occurred when executing {}", statement, e);
-- result.add(Statement.EXECUTE_FAILED);
-- isAllSuccessful = false;
-- batchErrorMessage = errMessage;
++ String msg = executeOneStatementInBatch(statement, result);
++ if (ERROR_QUERY_IN_BATCH.equals(msg)) {
++ return getTSBathExecuteStatementResp(TS_StatusCode.ERROR_STATUS,
++ "statement is query :" + statement, result);
}
++ isAllSuccessful = isAllSuccessful && msg != null;
++ batchErrorMessage = msg;
}
if (isAllSuccessful) {
return getTSBathExecuteStatementResp(TS_StatusCode.SUCCESS_STATUS,
@@@ -446,9 -487,9 +401,36 @@@
}
}
-- @Override
-- public TSExecuteStatementResp executeStatement(TSExecuteStatementReq req) throws TException {
++ private String executeOneStatementInBatch(String statement, List<Integer> result) {
++ String batchErrorMessage = null;
try {
++ PhysicalPlan physicalPlan = processor.parseSQLToPhysicalPlan(statement, zoneIds.get());
++ physicalPlan.setProposer(username.get());
++ if (physicalPlan.isQuery()) {
++ return ERROR_QUERY_IN_BATCH;
++ }
++ TSExecuteStatementResp resp = executeUpdateStatement(physicalPlan);
++ if (resp.getStatus().getStatusCode().equals(TS_StatusCode.SUCCESS_STATUS)) {
++ result.add(Statement.SUCCESS_NO_INFO);
++ } else {
++ result.add(Statement.EXECUTE_FAILED);
++ batchErrorMessage = resp.getStatus().getErrorMessage();
++ }
++ } catch (Exception e) {
++ String errMessage = String.format(
++ "Fail to generate physical plan and execute for statement "
++ + "%s because %s",
++ statement, e.getMessage());
++ LOGGER.warn("Error occurred when executing {}", statement, e);
++ result.add(Statement.EXECUTE_FAILED);
++ batchErrorMessage = errMessage;
++ }
++ return batchErrorMessage;
++ }
++
++ @Override
++ @SuppressWarnings("squid:2583")
++ public TSExecuteStatementResp executeStatement(TSExecuteStatementReq req) {
if (!checkLogin()) {
LOGGER.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME);
return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS, ERROR_NOT_LOGIN);
@@@ -464,36 -505,62 +446,57 @@@
return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS, e.getMessage());
}
+ try {
+ if (execSetConsistencyLevel(statement)) {
+ return getTSExecuteStatementResp(TS_StatusCode.SUCCESS_STATUS,
+ "Execute set consistency level successfully");
+ }
- } catch (Exception e) {
++ } catch (ConsistencyException e) {
+ LOGGER.error("Error occurred when executing statement {}", statement, e);
+ return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS, e.getMessage());
+ }
+
PhysicalPlan physicalPlan;
try {
physicalPlan = processor.parseSQLToPhysicalPlan(statement, zoneIds.get());
physicalPlan.setProposer(username.get());
-- } catch (IllegalASTFormatException e) {
- LOGGER.debug("meet error while parsing SQL to physical plan: {}", e.getMessage());
++ } catch (QueryProcessorException | ArgsErrorException | ProcessorException |
++ NullPointerException e) {
+ LOGGER.debug("meet error while parsing SQL to physical plan: ", e);
return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS,
"Statement format is not right:" + e.getMessage());
-- } catch (NullPointerException e) {
- LOGGER.error("meet error while parsing SQL to physical plan.", e);
- LOGGER.error("meet error while parsing SQL to physical plan: ", e);
-- return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS, "Statement is not allowed");
}
++
if (physicalPlan.isQuery()) {
return executeQueryStatement(req);
} else {
return executeUpdateStatement(physicalPlan);
}
-- } catch (Exception e) {
- LOGGER.info("meet error: {} while executing statement: {}", e.getMessage(), req.getStatement());
- LOGGER.info("meet error while executing statement: {}", req.getStatement(), e);
-- return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS, e.getMessage());
- }
+ }
+
+ /**
+ * Set consistency level
+ */
- public boolean execSetConsistencyLevel(String statement) throws Exception {
++ private boolean execSetConsistencyLevel(String statement) throws ConsistencyException {
+ if (statement == null) {
+ return false;
+ }
+ statement = statement.toLowerCase().trim();
+ if (Pattern.matches(IoTDBConstant.SET_READ_CONSISTENCY_LEVEL_PATTERN, statement)) {
- throw new Exception(
++ throw new ConsistencyException(
+ "IoTDB Stand-alone version does not support setting read-write consistency level");
+ } else {
+ return false;
}
}
@Override
-- public TSExecuteStatementResp executeQueryStatement(TSExecuteStatementReq req) throws TException {
++ public TSExecuteStatementResp executeQueryStatement(TSExecuteStatementReq req) {
try {
if (!checkLogin()) {
-- LOGGER.info("{}: Not login.", IoTDBConstant.GLOBAL_DB_NAME);
-- return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS, "Not login");
++ LOGGER.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME);
++ return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS, ERROR_NOT_LOGIN);
}
String statement = req.getStatement();
@@@ -591,8 -591,122 +527,105 @@@
}
}
+ private TSExecuteStatementResp executeAuthQuery(PhysicalPlan plan, List<String> columns) {
+ TSExecuteStatementResp resp = getTSExecuteStatementResp(TS_StatusCode.SUCCESS_STATUS, "");
+ resp.setIgnoreTimeStamp(true);
+ AuthorPlan authorPlan = (AuthorPlan) plan;
+ switch (authorPlan.getAuthorType()) {
+ case LIST_ROLE:
+ columns.add(ROLE);
+ break;
+ case LIST_USER:
+ columns.add(USER);
+ break;
+ case LIST_ROLE_USERS:
+ columns.add(USER);
+ break;
+ case LIST_USER_ROLES:
+ columns.add(ROLE);
+ break;
+ case LIST_ROLE_PRIVILEGE:
+ columns.add(PRIVILEGE);
+ break;
+ case LIST_USER_PRIVILEGE:
+ columns.add(ROLE);
+ columns.add(PRIVILEGE);
+ break;
+ default:
+ return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS, String.format("%s is not an "
+ + "auth query", authorPlan.getAuthorType()));
+ }
+ return resp;
+ }
+
+ private TSExecuteStatementResp executeDataQuery(PhysicalPlan plan, List<String> columns)
- throws AuthException, TException {
++ throws TException {
+ List<Path> paths;
+ paths = plan.getPaths();
+
+ // check seriesPath exists
+ if (paths.isEmpty()) {
+ return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS, "Timeseries does not exist.");
+ }
+
+ // check file level set
-
+ try {
+ checkFileLevelSet(paths);
+ } catch (PathErrorException e) {
+ LOGGER.error("meet error while checking file level.", e);
+ return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS, e.getMessage());
+ }
+
+ // check permissions
- if (!checkAuthorization(paths, plan)) {
++ try {
++ if (!checkAuthorization(paths, plan)) {
++ return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS,
++ "No permissions for this query.");
++ }
++ } catch (AuthException e) {
++ LOGGER.error("meet error in authorization", e);
+ return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS,
- "No permissions for this query.");
++ "Authorization error " + e.getMessage());
+ }
+
+ TSExecuteStatementResp resp = getTSExecuteStatementResp(TS_StatusCode.SUCCESS_STATUS, "");
+ // Restore column header of aggregate to func(column_name), only
+ // support single aggregate function for now
- if (plan instanceof QueryPlan) {
- switch (plan.getOperatorType()) {
- case QUERY:
- case FILL:
- for (Path p : paths) {
- columns.add(p.getFullPath());
- }
- break;
- case AGGREGATION:
- case GROUPBY:
- List<String> aggregations = plan.getAggregations();
- if (aggregations.size() != paths.size()) {
- for (int i = 1; i < paths.size(); i++) {
- aggregations.add(aggregations.get(0));
- }
- }
- for (int i = 0; i < paths.size(); i++) {
- columns.add(aggregations.get(i) + "(" + paths.get(i).getFullPath() + ")");
- }
- break;
- default:
- throw new TException("unsupported query type: " + plan.getOperatorType());
- }
- } else {
- Operator.OperatorType type = plan.getOperatorType();
- switch (type) {
- case QUERY:
- case FILL:
- for (Path p : paths) {
- columns.add(p.getFullPath());
- }
- break;
- case AGGREGATION:
- case GROUPBY:
- List<String> aggregations = plan.getAggregations();
- if (aggregations.size() != paths.size()) {
- for (int i = 1; i < paths.size(); i++) {
- aggregations.add(aggregations.get(0));
- }
- }
- for (int i = 0; i < paths.size(); i++) {
- columns.add(aggregations.get(i) + "(" + paths.get(i).getFullPath() + ")");
++ collectColumns(plan, paths, columns);
++ return resp;
++ }
++
++ private void collectColumns(PhysicalPlan plan, List<Path> paths, List<String> columns) throws TException {
++ switch (plan.getOperatorType()) {
++ case QUERY:
++ case FILL:
++ for (Path p : paths) {
++ columns.add(p.getFullPath());
++ }
++ break;
++ case AGGREGATION:
++ case GROUPBY:
++ List<String> aggregations = plan.getAggregations();
++ if (aggregations.size() != paths.size()) {
++ for (int i = 1; i < paths.size(); i++) {
++ aggregations.add(aggregations.get(0));
+ }
- break;
- default:
- throw new TException("not support " + type + " in new read process");
- }
++ }
++ for (int i = 0; i < paths.size(); i++) {
++ columns.add(aggregations.get(i) + "(" + paths.get(i).getFullPath() + ")");
++ }
++ break;
++ default:
++ throw new TException("unsupported query type: " + plan.getOperatorType());
+ }
- return resp;
+ }
+
- protected void checkFileLevelSet(List<Path> paths) throws PathErrorException {
++ private void checkFileLevelSet(List<Path> paths) throws PathErrorException {
+ MManager.getInstance().checkFileLevel(paths);
+ }
+
@Override
-- public TSFetchResultsResp fetchResults(TSFetchResultsReq req) throws TException {
++ public TSFetchResultsResp fetchResults(TSFetchResultsReq req) {
try {
if (!checkLogin()) {
return getTSFetchResultsResp(TS_StatusCode.ERROR_STATUS, "Not login.");
@@@ -632,7 -746,20 +665,20 @@@
PhysicalPlan physicalPlan = queryStatus.get().get(statement);
processor.getExecutor().setFetchSize(fetchSize);
+ QueryDataSet queryDataSet;
QueryContext context = new QueryContext(QueryResourceManager.getInstance().assignJobId());
+
+ initContextMap();
+ contextMapLocal.get().put(req.queryId, context);
+
+ queryDataSet = processor.getExecutor().processQuery(physicalPlan,
+ context);
+
+ queryRet.get().put(statement, queryDataSet);
+ return queryDataSet;
+ }
+
- protected void initContextMap() {
++ private void initContextMap() {
Map<Long, QueryContext> contextMap = contextMapLocal.get();
if (contextMap == null) {
contextMap = new HashMap<>();
@@@ -641,17 -768,11 +687,10 @@@
}
@Override
-- public TSExecuteStatementResp executeUpdateStatement(TSExecuteStatementReq req)
-- throws TException {
++ public TSExecuteStatementResp executeUpdateStatement(TSExecuteStatementReq req) {
try {
if (!checkLogin()) {
-- return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS, "Not login");
++ return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS, ERROR_NOT_LOGIN);
}
String statement = req.getStatement();
return executeUpdateStatement(statement);
@@@ -699,6 -820,10 +738,10 @@@
return resp;
}
- protected boolean executeNonQuery(PhysicalPlan plan) throws ProcessorException {
++ private boolean executeNonQuery(PhysicalPlan plan) throws ProcessorException {
+ return processor.getExecutor().processNonQuery(plan);
+ }
+
private TSExecuteStatementResp executeUpdateStatement(String statement)
throws ProcessorException {
@@@ -755,17 -847,17 +765,15 @@@
private void recordANewQuery(String statement, PhysicalPlan physicalPlan) {
queryStatus.get().put(statement, physicalPlan);
// refresh current queryRet for statement
-- if (queryRet.get().containsKey(statement)) {
-- queryRet.get().remove(statement);
-- }
++ queryRet.get().remove(statement);
}
/**
-- * Check whether current user has logined.
++ * Check whether current user has logged in.
*
-- * @return true: If logined; false: If not logined
++ * @return true: If logged in; false: If not logged in
*/
- protected boolean checkLogin() {
+ private boolean checkLogin() {
return username.get() != null;
}
@@@ -790,7 -882,8 +798,8 @@@
return resp;
}
- private TSExecuteBatchStatementResp getTSBathExecuteStatementResp(TS_StatusCode code, String msg,
- protected TSExecuteBatchStatementResp getTSBathExecuteStatementResp(TS_StatusCode code,
++ private TSExecuteBatchStatementResp getTSBathExecuteStatementResp(TS_StatusCode code,
+ String msg,
List<Integer> result) {
TSExecuteBatchStatementResp resp = new TSExecuteBatchStatementResp();
TS_Status tsStatus = new TS_Status(code);
@@@ -808,15 -901,15 +817,15 @@@
return resp;
}
- public void handleClientExit() throws TException {
- protected void handleClientExit() throws TException {
++ void handleClientExit() {
closeOperation(null);
closeSession(null);
}
@Override
-- public TSGetTimeZoneResp getTimeZone() throws TException {
- TS_Status tsStatus = null;
- TSGetTimeZoneResp resp = null;
++ public TSGetTimeZoneResp getTimeZone() {
+ TS_Status tsStatus;
+ TSGetTimeZoneResp resp;
try {
tsStatus = new TS_Status(TS_StatusCode.SUCCESS_STATUS);
resp = new TSGetTimeZoneResp(tsStatus, zoneIds.get().toString());
@@@ -830,8 -923,8 +839,8 @@@
}
@Override
-- public TSSetTimeZoneResp setTimeZone(TSSetTimeZoneReq req) throws TException {
- TS_Status tsStatus = null;
++ public TSSetTimeZoneResp setTimeZone(TSSetTimeZoneReq req) {
+ TS_Status tsStatus;
try {
String timeZoneID = req.getTimeZone();
zoneIds.set(ZoneId.of(timeZoneID));
@@@ -845,7 -938,7 +854,7 @@@
}
@Override
-- public ServerProperties getProperties() throws TException {
++ public ServerProperties getProperties() {
ServerProperties properties = new ServerProperties();
properties.setVersion(IoTDBConstant.VERSION);
properties.setSupportedTimeAggregationOperations(new ArrayList<>());
diff --cc iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/SyncServiceImpl.java
index 552c353,ad2722d..944620a
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/SyncServiceImpl.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/SyncServiceImpl.java
@@@ -474,12 -475,12 +474,12 @@@ public class SyncServiceImpl implement
);
// call interface of load external file
try {
- if (!fileNodeManager.appendFileToFileNode(storageGroup, fileNode, path)) {
+ if (!databaseEngine.appendFileToStorageGroup(storageGroup, fileNode, path)) {
// it is a file with overflow data
- if (config.isUpdate_historical_data_possibility()) {
+ if (config.isUpdateHistoricalDataPossibility()) {
loadOldData(path);
} else {
- List<String> overlapFiles = fileNodeManager.getOverlapFilesFromFileNode(
+ List<String> overlapFiles = databaseEngine.getOverlapFilesFromStorageGroup(
storageGroup,
fileNode, uuid.get());
if (overlapFiles.isEmpty()) {
diff --cc iotdb/src/main/java/org/apache/iotdb/db/utils/RecordUtils.java
index 4a443ec,5487ff2..fc618e1
--- a/iotdb/src/main/java/org/apache/iotdb/db/utils/RecordUtils.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/utils/RecordUtils.java
@@@ -18,17 -18,16 +18,9 @@@
*/
package org.apache.iotdb.db.utils;
+import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.tsfile.common.constant.JsonFormatConstant;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
--import org.apache.iotdb.tsfile.utils.Binary;
--import org.apache.iotdb.tsfile.write.record.TSRecord;
--import org.apache.iotdb.tsfile.write.record.datapoint.BooleanDataPoint;
--import org.apache.iotdb.tsfile.write.record.datapoint.DoubleDataPoint;
--import org.apache.iotdb.tsfile.write.record.datapoint.FloatDataPoint;
--import org.apache.iotdb.tsfile.write.record.datapoint.IntDataPoint;
--import org.apache.iotdb.tsfile.write.record.datapoint.LongDataPoint;
--import org.apache.iotdb.tsfile.write.record.datapoint.StringDataPoint;
import org.apache.iotdb.tsfile.write.schema.FileSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@@ -50,9 -49,9 +42,9 @@@ public class RecordUtils
*
* @param str - input string
* @param schema - constructed file schema
- * @return TSRecord constructed from str
+ * @return InsertPlan constructed from str
*/
- public static InsertPlan parseSimpleTuplePlan(String str, FileSchema schema) {
- public static TSRecord parseSimpleTupleRecord(String str, FileSchema schema) {
++ static InsertPlan parseSimpleTuplePlan(String str, FileSchema schema) {
// split items
String[] items = str.split(JsonFormatConstant.TSRECORD_SEPARATOR);
// get deviceId and timestamp, then create a new TSRecord
@@@ -63,12 -62,10 +55,12 @@@
} catch (NumberFormatException e) {
LOG.warn("given timestamp is illegal:{}", str);
// return a TSRecord without any data points
- return new InsertPlan(deviceId, -1, null, null);
- return new TSRecord(-1, deviceId);
++ return new InsertPlan(deviceId, -1L, (String) null, null);
}
- TSRecord ret = new TSRecord(timestamp, deviceId);
-
+ int valuesNum = (items.length - 2) / 2;
+ String[] measurements = new String[valuesNum];
+ String[] values = new String[valuesNum];
+ int idx = 0;
// loop all rest items except the last one
String measurementId;
TSDataType type;
diff --cc iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/ExclusiveLogRecoverPerformer.java
index 5972425,0b58283..423cc66
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/ExclusiveLogRecoverPerformer.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/ExclusiveLogRecoverPerformer.java
@@@ -48,7 -48,7 +48,7 @@@ public class ExclusiveLogRecoverPerform
public static final String RECOVER_FLAG_NAME = "recover-flag";
public static final String RECOVER_SUFFIX = "-recover";
-- public static final String FLAG_SEPERATOR = "-";
++ private static final String FLAG_SEPARATOR = "-";
private static final Logger logger = LoggerFactory.getLogger(ExclusiveLogRecoverPerformer.class);
// The two fields can be made static only because the recovery is a serial process.
private static RAFLogReader rafLogReader = new RAFLogReader();
@@@ -107,7 -116,7 +107,7 @@@
String flagName = flagFile.getName();
recoveryFlagPath = flagFile.getPath();
// the flag name is like "recover-flag-{flagType}"
-- String[] parts = flagName.split(FLAG_SEPERATOR);
++ String[] parts = flagName.split(FLAG_SEPARATOR);
if (parts.length != 3) {
logger.error("Log node {} invalid recover flag name {}", writeLogNode.getIdentifier(),
flagName);
@@@ -149,7 -158,7 +149,7 @@@
private void setFlag(RecoverStage stage) {
if (recoveryFlagPath == null) {
recoveryFlagPath =
-- writeLogNode.getLogDirectory() + File.separator + RECOVER_FLAG_NAME + FLAG_SEPERATOR
++ writeLogNode.getLogDirectory() + File.separator + RECOVER_FLAG_NAME + FLAG_SEPARATOR
+ stage.name();
try {
File flagFile = new File(recoveryFlagPath);
@@@ -164,8 -173,8 +164,8 @@@
}
} else {
File flagFile = new File(recoveryFlagPath);
-- recoveryFlagPath = recoveryFlagPath.replace(FLAG_SEPERATOR + currStage.name(),
-- FLAG_SEPERATOR + stage.name());
++ recoveryFlagPath = recoveryFlagPath.replace(FLAG_SEPARATOR + currStage.name(),
++ FLAG_SEPARATOR + stage.name());
if (!flagFile.renameTo(new File(recoveryFlagPath))) {
logger
.error("Log node {} cannot update flag at stage {}", writeLogNode.getLogDirectory(),
@@@ -277,11 -312,10 +277,10 @@@
+ " logs failed to recover, see logs above for details");
}
try {
- FileNodeManager.getInstance().closeOneFileNode(writeLogNode.getFileNodeName());
- } catch (FileNodeManagerException e) {
+ DatabaseEngineFactory.getCurrent().closeStorageGroup(writeLogNode.getFileNodeName());
+ } catch (StorageGroupManagerException e) {
- logger.error("Log node {} cannot perform flush after replaying logs! Because {}",
- writeLogNode.getIdentifier(), e.getMessage());
- throw new RecoverException(e);
+ throw new RecoverException(String.format("Log node %s cannot perform flush"
- + " after replaying logs!", writeLogNode.getIdentifier()), e);
++ + " after replaying logs!", writeLogNode.getIdentifier()), e);
}
currStage = CLEAN_UP;
setFlag(REPLAY_LOG);
diff --cc iotdb/src/test/java/org/apache/iotdb/db/engine/tsfiledata/TsFileProcessorTest.java
index 866fb89,0000000..64b06e1
mode 100644,000000..100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/tsfiledata/TsFileProcessorTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/tsfiledata/TsFileProcessorTest.java
@@@ -1,249 -1,0 +1,249 @@@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.engine.tsfiledata;
+
+import static org.apache.iotdb.db.utils.EnvironmentUtils.TEST_QUERY_CONTEXT;
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.datasource.SeriesDataSource;
+import org.apache.iotdb.db.engine.sgmanager.OperationResult;
+import org.apache.iotdb.db.engine.version.SysTimeVersionController;
+import org.apache.iotdb.db.exception.StorageGroupManagerException;
+import org.apache.iotdb.db.exception.TsFileProcessorException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.reader.sequence.SequenceDataReader;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.db.utils.ImmediateFuture;
+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.BatchData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.expression.impl.SingleSeriesExpression;
+import org.apache.iotdb.tsfile.write.schema.FileSchema;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TsFileProcessorTest {
+ private static Logger LOGGER = LoggerFactory.getLogger(TsFileProcessorTest.class);
+ protected TsFileProcessor processor;
+ private MManager mManager;
+ private Map<String, MeasurementSchema> measurementSchemaMap = new HashMap<>();
+
+ protected FileSchema schema;
+
+ private long oldTsFileSizeThreshold = IoTDBDescriptor.getInstance().getConfig().getBufferwriteFileSizeThreshold();
+ @Before
+ public void setUp() throws Exception {
+ EnvironmentUtils.envSetUp();
+// now we do not support wal because it need to modify the wal module.
+ IoTDBDescriptor.getInstance().getConfig().setEnableWal(true);
+ IoTDBDescriptor.getInstance().getConfig().setBufferwriteFileSizeThreshold(2*1024*1024);
+ mManager = MManager.getInstance();
+ measurementSchemaMap.put("s1", new MeasurementSchema("s1", TSDataType.FLOAT,
+ TSEncoding.RLE));
+ measurementSchemaMap.put("s2", new MeasurementSchema("s2", TSDataType.FLOAT,
+ TSEncoding.RLE));
+ measurementSchemaMap.put("s3", new MeasurementSchema("s3", TSDataType.FLOAT,
+ TSEncoding.RLE));
+ schema = new FileSchema(measurementSchemaMap);
+ processor = new TsFileProcessor("root.test", SysTimeVersionController.INSTANCE,
+ schema);
+ mManager.setStorageLevelToMTree("root.test");
+ mManager.addPathToMTree("root.test.d1.s1", TSDataType.FLOAT, TSEncoding.RLE,
+ CompressionType.SNAPPY, Collections.emptyMap());
+ mManager.addPathToMTree("root.test.d2.s1", TSDataType.FLOAT, TSEncoding.RLE,
+ CompressionType.SNAPPY, Collections.emptyMap());
+ mManager.addPathToMTree("root.test.d1.s2", TSDataType.FLOAT, TSEncoding.RLE,
+ CompressionType.SNAPPY, Collections.emptyMap());
+ mManager.addPathToMTree("root.test.d2.s2", TSDataType.FLOAT, TSEncoding.RLE,
+ CompressionType.SNAPPY, Collections.emptyMap());
+ }
+
+
+ @After
+ public void tearDown() throws Exception {
+ //processor.close();
+ // processor.writeLock();
+ processor.removeMe();
+ EnvironmentUtils.cleanEnv();
+ IoTDBDescriptor.getInstance().getConfig().setEnableWal(false);
+ IoTDBDescriptor.getInstance().getConfig().
+ setBufferwriteFileSizeThreshold(oldTsFileSizeThreshold);
+ }
+
+ @Test
+ public void insert()
+ throws IOException, ExecutionException, InterruptedException, TsFileProcessorException {
+ String[] s1 = new String[]{"s1"};
+ String[] s2 = new String[]{"s2"};
+ String[] value = new String[]{"5.0"};
+
+ assertEquals(
+ OperationResult.WRITE_SUCCESS,
+ processor.insert(new InsertPlan("root.test.d1", 10, s1, value)));
+ assertEquals(OperationResult.WRITE_SUCCESS,
+ processor.insert(new InsertPlan("root.test.d1", 10, s2, value)));
+ assertEquals(OperationResult.WRITE_SUCCESS,
+ processor.insert(new InsertPlan("root.test.d1", 12, s1, value)));
+ Future<Boolean> ok = processor.flush();
+ ok.get();
+ ok = processor.flush();
+ Assert.assertTrue(ok instanceof ImmediateFuture);
+ ok.get();
+ ok = processor.flush();
+ Assert.assertTrue(ok instanceof ImmediateFuture);
+ ok.get();
+
+ //let's rewrite timestamp =12 again..
+ assertEquals(OperationResult.WRITE_REJECT_BY_TIME,
+ processor.insert(new InsertPlan("root.test.d1", 12, s1, value)));
+ processor.delete("root.test.d1", "s1",12);
+ assertEquals(OperationResult.WRITE_SUCCESS,
+ processor.insert(new InsertPlan("root.test.d1", 12, s1, value)));
+ assertEquals(OperationResult.WRITE_SUCCESS,
+ processor.insert(new InsertPlan("root.test.d1", 13, s1, value)));
+ assertEquals(OperationResult.WRITE_SUCCESS,
+ processor.insert(new InsertPlan("root.test.d2", 10, s1, value)));
+ assertEquals(OperationResult.WRITE_SUCCESS,
+ processor.insert(new InsertPlan("root.test.d1", 14, s1, value)));
+ processor.delete("root.test.d1", "s1",12);
+ processor.delete("root.test.d3", "s1",12);
+
+ executeQuery("root.test.d1", "s1", TEST_QUERY_CONTEXT, 2);
+ }
+
+ private void executeQuery(String deviceId, String measurementId, QueryContext context,
+ int expectedCnt)
+ throws IOException {
+ SingleSeriesExpression singleSeriesExpression = new SingleSeriesExpression(
+ new Path(deviceId, measurementId), null);
+ SeriesDataSource dataSource = processor.query(singleSeriesExpression, context);
+ SequenceDataReader dataReader = new SequenceDataReader(dataSource, null, context);
+ int i = 0;
+ while (dataReader.hasNext()) {
+ BatchData batch = dataReader.nextBatch();
+ while (batch.hasNext()) {
+ //System.out.println(batch.currentTime() +"," + batch.getFloat());
+ batch.next();
+ i ++;
+ }
+ }
+ if (expectedCnt >= 0) {
+ assertEquals(expectedCnt, i);
+ }
+ }
+
+
+
+ @Test
+ public void bruteForceTest() throws InterruptedException, IOException {
+
+ final boolean[] exception = {false, false, false};
+ final boolean[] goon = {true};
+ int totalSize = 50000;
+
+ final int[] count = {0};
+ Thread insertThread = new Thread(() -> {
+ long time = 100L;
+ long start = System.currentTimeMillis();
+ String[] sensors = new String[]{"s1"};
+ String[] values = new String[1];
+ try {
+ for (int j = 0; j < totalSize && goon[0]; j++) {
+ processor.lock(true);
+// processor.insert("root.test.d1","s1", time++, String.valueOf(j));
+// processor.insert("root.test.d2","s1", time++, String.valueOf(j));
+ values[0] = String.valueOf(j);
+ processor.insert(new InsertPlan("root.test.d1", time++, sensors, values));
+ processor.insert(new InsertPlan("root.test.d2", time++, sensors, values));
+ processor.writeUnlock();
+ count[0]++;
+ }
+ System.out.println((System.currentTimeMillis() - start));
+ } catch (TsFileProcessorException e) {
+ LOGGER.error("", e);
+ exception[0] = true;
+ }
+ });
+ Thread flushThread = new Thread(() -> {
+ try {
+ for (int j = 0; j < totalSize * 2 && goon[0]; j++) {
+ processor.lock(true);
+ processor.flush();
+ processor.writeUnlock();
+ }
- } catch (IOException e) {
++ } catch (IOException | TsFileProcessorException e) {
+ // we will break out.
+ LOGGER.error(e.getMessage());
+ exception[1] = true;
+ }
+ });
+ //we temporary disable the query because there are bugs..
+ Thread queryThread = new Thread(() -> {
+ try {
+ for (int j = 0; j < totalSize * 2 && goon[0]; j++) {
+ processor.lock(false);
+ QueryContext context = new QueryContext(QueryResourceManager.getInstance().assignJobId());
+ executeQuery("root.test.d1", "s1", context, -1);
+ QueryResourceManager.getInstance().endQueryForGivenJob(context.getJobId());
+ processor.readUnlock();
+ }
+ } catch (IOException | StorageGroupManagerException e) {
+ // we will break out.
+ LOGGER.error(e.getMessage());
+ exception[2] = true;
+ }
+ });
+ flushThread.start();
+ insertThread.start();
+ queryThread.start();
+ //wait at most 20 seconds.
+ insertThread.join(20000);
+ goon[0] = false;
+ //queryThread.join(5000);
+ Assert.assertFalse(exception[0]);
+ Assert.assertFalse(exception[1]);
+ Assert.assertFalse(exception[2]);
+ synchronized (this) {
+ while (queryThread.isAlive()) {
+ this.wait(50);
+ }
+ }
+ executeQuery("root.test.d1", "s1", TEST_QUERY_CONTEXT, 50000);
+ }
+}
+
+
diff --cc iotdb/src/test/java/org/apache/iotdb/db/qp/utils/MemIntQpExecutor.java
index c254657,7afced3..10c9832
--- a/iotdb/src/test/java/org/apache/iotdb/db/qp/utils/MemIntQpExecutor.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/qp/utils/MemIntQpExecutor.java
@@@ -32,8 -36,11 +32,9 @@@ import org.apache.iotdb.db.qp.physical.
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.crud.UpdatePlan;
+ import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.query.context.QueryContext;
-import org.apache.iotdb.db.query.executor.EngineQueryRouter;
import org.apache.iotdb.db.query.fill.IFill;
-import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
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;
@@@ -170,8 -181,8 +171,13 @@@ public class MemIntQpExecutor extends Q
}
@Override
+ public int insert(Path path, long insertTime, String value) {
- String strPath = path.toString();
++ return 0;
++ }
++
++ @Override
+ public void insert(InsertPlan plan) {
+ String strPath = plan.getDeviceId() + "." + plan.getMeasurements()[0];
if (!demoMemDataBase.containsKey(strPath)) {
demoMemDataBase.put(strPath, new TestSeries());
}
@@@ -189,6 -201,17 +195,11 @@@
};
}
+ @Override
- public int multiInsert(String deviceId, long insertTime, String[] measurementList,
- String[] insertValues) {
- return 0;
- }
-
- @Override
+ protected QueryDataSet processAuthorQuery(AuthorPlan plan, QueryContext context) {
+ return null;
+ }
+
private class TestSeries {
public TreeMap<Long, Integer> data = new TreeMap<>();
diff --cc iotdb/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java
index 435a2c8,31f12fd..48d108f
--- a/iotdb/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java
@@@ -104,13 -110,13 +104,11 @@@ public class EnvironmentUtils
cleanDir(config.getMetadataDir());
// delete wal
cleanDir(config.getWalFolder());
-- // delete derby
-- cleanDir(config.getDerbyHome());
// delete index
cleanDir(config.getIndexFileDir());
- // delte data
+ // delete data
cleanDir("data");
- // delte derby log
+ // delete derby log
// cleanDir("derby.log");
}
diff --cc iotdb/src/test/java/org/apache/iotdb/db/writelog/PerformanceTest.java
index 9e71f63,17777e7..abcccda
--- a/iotdb/src/test/java/org/apache/iotdb/db/writelog/PerformanceTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/writelog/PerformanceTest.java
@@@ -31,7 -34,9 +31,7 @@@ import org.apache.iotdb.db.qp.physical.
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.db.writelog.node.ExclusiveWriteLogNode;
import org.apache.iotdb.db.writelog.node.WriteLogNode;
- import org.apache.iotdb.db.writelog.transfer.PhysicalPlanLogTransfer;
+ import org.apache.iotdb.db.qp.physical.transfer.PhysicalPlanLogTransfer;
-import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
-import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
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;