This is an automated email from the ASF dual-hosted git repository. hxd pushed a commit to branch cluster- in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5f980240986c4c5100f46396d20f15d136560ea0 Merge: 12ad291 cbbdc6c Author: xiangdong huang <[email protected]> AuthorDate: Tue Oct 5 14:55:09 2021 +0800 merge with master .asf.yaml | 5 +- .github/workflows/client.yml | 1 + README.md | 2 +- README_ZH.md | 2 +- .../antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4 | 13 +- client-cpp/src/main/CMakeLists.txt | 5 +- client-cpp/src/main/Session.cpp | 413 ++-- client-cpp/src/main/Session.h | 313 ++-- .../java/org/apache/iotdb/cluster/ClientMain.java | 6 +- .../org/apache/iotdb/cluster/ClusterIoTDB.java | 2 +- .../iotdb/cluster/coordinator/Coordinator.java | 2 +- .../cluster/log/applier/AsyncDataLogApplier.java | 8 +- .../iotdb/cluster/log/applier/DataLogApplier.java | 6 +- .../log/manage/PartitionedSnapshotLogManager.java | 11 +- .../apache/iotdb/cluster/metadata/CMManager.java | 118 +- .../apache/iotdb/cluster/metadata/MetaPuller.java | 32 +- .../iotdb/cluster/partition/PartitionTable.java | 4 +- .../cluster/query/ClusterPhysicalGenerator.java | 7 +- .../iotdb/cluster/query/ClusterPlanExecutor.java | 19 +- .../iotdb/cluster/query/ClusterPlanRouter.java | 15 +- .../iotdb/cluster/query/ClusterQueryRouter.java | 5 +- .../cluster/query/ClusterUDTFQueryExecutor.java | 31 +- .../iotdb/cluster/query/LocalQueryExecutor.java | 4 +- .../query/aggregate/ClusterAggregateExecutor.java | 30 +- .../caller/PullMeasurementSchemaHandler.java | 4 +- .../cluster/server/member/MetaGroupMember.java | 2 +- .../apache/iotdb/cluster/utils/ClusterUtils.java | 4 +- .../cluster/client/sync/SyncClientAdaptorTest.java | 4 +- .../org/apache/iotdb/cluster/common/TestUtils.java | 8 +- .../query/ClusterAggregateExecutorTest.java | 8 +- .../cluster/server/member/DataGroupMemberTest.java | 2 +- .../cluster/server/member/MetaGroupMemberTest.java | 2 +- compile-tools/thrift/pom.xml | 38 +- .../tests/tools/importCsv/ImportCsvTestIT.java | 8 +- docs/Development/ContributeGuide.md | 12 +- docs/Development/HowToCommit.md | 18 + docs/SystemDesign/SchemaManager/SchemaManager.md | 2 +- docs/UserGuide/API/Programming-JDBC.md | 2 + docs/UserGuide/Advanced-Features/Select-Into.md | 8 + .../Advanced-Features/UDF-User-Defined-Function.md | 12 +- docs/UserGuide/Appendix/SQL-Reference.md | 121 +- .../Collaboration-of-Edge-and-Cloud/Sync-Tool.md | 2 +- .../Data-Concept/Data-Model-and-Terminology.md | 23 +- .../DDL-Data-Definition-Language.md | 69 +- .../DML-Data-Manipulation-Language.md | 793 ++++---- docs/UserGuide/UserGuideReadme.md | 2 +- docs/zh/Development/ContributeGuide.md | 10 + docs/zh/Development/HowToCommit.md | 18 + .../zh/SystemDesign/SchemaManager/SchemaManager.md | 2 +- docs/zh/UserGuide/API/Programming-JDBC.md | 2 + docs/zh/UserGuide/Advanced-Features/Select-Into.md | 8 + .../Advanced-Features/UDF-User-Defined-Function.md | 10 +- docs/zh/UserGuide/Appendix/Config-Manual.md | 1494 +++++++++++---- docs/zh/UserGuide/Appendix/SQL-Reference.md | 127 +- .../Data-Concept/Data-Model-and-Terminology.md | 28 +- .../DDL-Data-Definition-Language.md | 72 +- .../DML-Data-Manipulation-Language.md | 1617 ++++++++-------- example/client-cpp-example/src/SessionExample.cpp | 3 +- .../apache/iotdb/flink/FlinkTsFileBatchSink.java | 11 +- .../apache/iotdb/flink/FlinkTsFileStreamSink.java | 11 +- .../java/org/apache/iotdb/flink/TsFileUtils.java | 9 +- .../iotdb/hadoop/tsfile/TSMRWriteExample.java | 6 +- .../apache/iotdb/hadoop/tsfile/TsFileHelper.java | 8 +- .../iotdb/hadoop/tsfile/TsFileWriteToHDFS.java | 8 +- .../server/CustomizedJsonPayloadFormatter.java | 2 +- .../iotdb/AlignedTimeseriesSessionExample.java | 245 ++- .../org/apache/iotdb/DataMigrationExample.java | 4 +- .../org/apache/iotdb/SessionConcurrentExample.java | 8 +- .../main/java/org/apache/iotdb/SessionExample.java | 97 +- .../main/java/org/apache/iotdb/TabletExample.java | 4 +- .../iotdb/tsfile/TsFileForceAppendWrite.java | 14 +- .../iotdb/tsfile/TsFileWriteWithTSRecord.java | 8 +- .../apache/iotdb/tsfile/TsFileWriteWithTablet.java | 8 +- .../tsfile/RowTsFileOutputFormatTestBase.java | 10 +- .../apache/iotdb/flink/util/TsFileWriteUtil.java | 18 +- .../iotdb/hadoop/tsfile/TsFileTestHelper.java | 6 +- .../org/apache/iotdb/hive/TsFileTestHelper.java | 6 +- .../java/org/apache/iotdb/jdbc/IoTDBStatement.java | 5 +- .../src/main/java/org/apache/iotdb/jdbc/Utils.java | 47 +- .../test/java/org/apache/iotdb/jdbc/UtilsTest.java | 12 + .../resources/conf/iotdb-engine.properties | 5 + .../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 11 + .../org/apache/iotdb/db/conf/IoTDBConfigCheck.java | 10 + .../org/apache/iotdb/db/conf/IoTDBConstant.java | 7 +- .../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 7 + .../org/apache/iotdb/db/engine/StorageEngine.java | 18 +- .../db/engine/cache/TimeSeriesMetadataCache.java | 6 +- .../level/LevelCompactionTsFileManagement.java | 2 +- .../engine/compaction/utils/CompactionUtils.java | 7 +- .../iotdb/db/engine/memtable/AbstractMemTable.java | 4 +- .../apache/iotdb/db/engine/memtable/IMemTable.java | 8 +- .../db/engine/memtable/IWritableMemChunk.java | 21 +- .../iotdb/db/engine/memtable/WritableMemChunk.java | 56 +- .../iotdb/db/engine/merge/task/MergeTask.java | 20 +- .../engine/storagegroup/StorageGroupProcessor.java | 164 +- .../db/engine/storagegroup/TsFileProcessor.java | 7 +- .../db/engine/storagegroup/TsFileResource.java | 5 +- .../service/TriggerRegistrationService.java | 4 +- .../metadata/MNodeTypeMismatchException.java} | 18 +- .../metadata/StorageGroupAlreadySetException.java | 11 +- .../org/apache/iotdb/db/metadata/MManager.java | 1207 ++++++------ .../java/org/apache/iotdb/db/metadata/MTree.java | 1963 -------------------- .../apache/iotdb/db/metadata/MetadataConstant.java | 27 +- .../org/apache/iotdb/db/metadata/PartialPath.java | 47 +- .../iotdb/db/metadata/VectorPartialPath.java | 6 + .../db/metadata/lastCache/LastCacheManager.java | 82 +- .../iotdb/db/metadata/logfile/MLogWriter.java | 4 +- .../iotdb/db/metadata/mnode/IEntityMNode.java | 2 +- .../org/apache/iotdb/db/metadata/mnode/IMNode.java | 9 +- .../iotdb/db/metadata/mnode/IMeasurementMNode.java | 10 +- .../iotdb/db/metadata/mnode/InternalMNode.java | 30 +- .../org/apache/iotdb/db/metadata/mnode/MNode.java | 35 + .../iotdb/db/metadata/mnode/MeasurementMNode.java | 161 +- .../db/metadata/mnode/MultiMeasurementMNode.java | 65 + .../db/metadata/mnode/UnaryMeasurementMNode.java | 63 + .../org/apache/iotdb/db/metadata/mtree/MTree.java | 1371 ++++++++++++++ .../mtree/traverser/PathGrouperByStorageGroup.java | 103 + .../db/metadata/mtree/traverser/Traverser.java | 247 +++ .../collector/BelongedEntityPathCollector.java | 83 + .../traverser/collector/CollectorTraverser.java | 93 + .../traverser/collector/EntityPathCollector.java | 66 + .../collector/FlatMeasurementCollector.java | 138 ++ .../collector/FlatMeasurementPathCollector.java | 60 + .../collector/FlatMeasurementSchemaCollector.java | 117 ++ .../mtree/traverser/collector/MNodeCollector.java | 76 + .../traverser/collector/MeasurementCollector.java | 58 + .../collector/StorageGroupPathCollector.java | 65 + .../mtree/traverser/counter/CounterTraverser.java} | 37 +- .../mtree/traverser/counter/EntityCounter.java} | 31 +- .../traverser/counter/FlatMeasurementCounter.java | 78 + .../traverser/counter/MNodeLevelCounter.java} | 38 +- .../traverser/counter/MeasurementCounter.java | 43 +- .../traverser/counter/StorageGroupCounter.java} | 37 +- .../apache/iotdb/db/metadata/tag/TagLogFile.java | 85 +- .../iotdb/db/metadata/template/Template.java | 10 +- .../apache/iotdb/db/metadata/utils/MetaUtils.java | 9 +- .../apache/iotdb/db/mqtt/JSONPayloadFormatter.java | 51 +- .../org/apache/iotdb/db/mqtt/PayloadFormatter.java | 8 +- .../apache/iotdb/db/qp/constant/SQLConstant.java | 2 +- .../apache/iotdb/db/qp/executor/PlanExecutor.java | 79 +- .../iotdb/db/qp/logical/crud/QueryOperator.java | 35 +- .../iotdb/db/qp/physical/crud/AggregationPlan.java | 5 - .../db/qp/physical/crud/AlignByDevicePlan.java | 5 + .../iotdb/db/qp/physical/crud/FillQueryPlan.java | 5 - .../db/qp/physical/crud/InsertMultiTabletPlan.java | 17 +- .../iotdb/db/qp/physical/crud/InsertRowPlan.java | 108 +- .../physical/crud/InsertRowsOfOneDevicePlan.java | 41 +- .../iotdb/db/qp/physical/crud/InsertRowsPlan.java | 32 +- .../db/qp/physical/crud/InsertTabletPlan.java | 60 +- .../iotdb/db/qp/physical/crud/LastQueryPlan.java | 5 - .../iotdb/db/qp/physical/crud/QueryIndexPlan.java | 5 - .../iotdb/db/qp/physical/crud/QueryPlan.java | 10 - .../db/qp/physical/crud/RawDataQueryPlan.java | 32 +- .../apache/iotdb/db/qp/physical/crud/UDFPlan.java | 5 - .../apache/iotdb/db/qp/physical/crud/UDTFPlan.java | 98 +- .../qp/physical/sys/CreateMultiTimeSeriesPlan.java | 15 +- .../db/qp/physical/sys/MeasurementMNodePlan.java | 4 +- .../apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java | 75 +- .../iotdb/db/qp/strategy/PhysicalGenerator.java | 7 +- .../db/query/aggregation/AggregateResult.java | 10 +- .../db/query/aggregation/impl/AvgAggrResult.java | 17 +- .../db/query/aggregation/impl/CountAggrResult.java | 15 +- .../query/aggregation/impl/ExtremeAggrResult.java | 19 +- .../aggregation/impl/FirstValueAggrResult.java | 25 +- .../aggregation/impl/FirstValueDescAggrResult.java | 17 +- .../aggregation/impl/LastValueAggrResult.java | 21 +- .../aggregation/impl/LastValueDescAggrResult.java | 17 +- .../query/aggregation/impl/MaxTimeAggrResult.java | 19 +- .../aggregation/impl/MaxTimeDescAggrResult.java | 13 +- .../query/aggregation/impl/MaxValueAggrResult.java | 21 +- .../query/aggregation/impl/MinTimeAggrResult.java | 17 +- .../aggregation/impl/MinTimeDescAggrResult.java | 11 +- .../query/aggregation/impl/MinValueAggrResult.java | 19 +- .../db/query/aggregation/impl/SumAggrResult.java | 17 +- .../db/query/dataset/AlignByDeviceDataSet.java | 19 +- .../iotdb/db/query/dataset/ShowDevicesDataSet.java | 3 +- .../db/query/dataset/UDTFAlignByTimeDataSet.java | 24 +- .../apache/iotdb/db/query/dataset/UDTFDataSet.java | 180 +- .../db/query/dataset/UDTFNonAlignDataSet.java | 22 +- .../dataset/groupby/LocalGroupByExecutor.java | 14 +- .../db/query/executor/AggregationExecutor.java | 377 +++- .../iotdb/db/query/executor/LastQueryExecutor.java | 27 +- .../iotdb/db/query/executor/QueryRouter.java | 11 +- .../iotdb/db/query/executor/UDTFQueryExecutor.java | 30 +- .../iotdb/db/query/expression/Expression.java | 64 +- .../iotdb/db/query/expression/ResultColumn.java | 13 +- .../expression/binary/AdditionExpression.java | 10 + .../query/expression/binary/BinaryExpression.java | 82 +- .../expression/binary/DivisionExpression.java | 10 + .../query/expression/binary/ModuloExpression.java | 9 + .../binary/MultiplicationExpression.java | 10 + .../expression/binary/SubtractionExpression.java | 10 + .../query/expression/unary/FunctionExpression.java | 143 +- .../query/expression/unary/NegationExpression.java | 64 +- .../query/expression/unary/TimeSeriesOperand.java | 53 +- .../db/query/reader/series/IAggregateReader.java | 2 +- .../iotdb/db/query/reader/series/SeriesReader.java | 38 + .../reader/series/VectorSeriesAggregateReader.java | 177 ++ .../reader/universal/DescPriorityMergeReader.java | 9 +- .../apache/iotdb/db/query/udf/api/access/Row.java | 18 +- .../api/customizer/parameter/UDFParameters.java | 49 +- .../db/query/udf/builtin/BuiltinFunction.java | 3 + .../iotdb/db/query/udf/builtin/UDTFConst.java | 123 ++ .../iotdb/db/query/udf/builtin/UDTFConstE.java} | 45 +- .../iotdb/db/query/udf/builtin/UDTFConstPi.java} | 45 +- .../iotdb/db/query/udf/builtin/UDTFSelectK.java | 2 +- .../iotdb/db/query/udf/builtin/UDTFValueTrend.java | 3 +- ...alizableRowRecordListBackedMultiColumnRow.java} | 38 +- ...zableRowRecordListBackedMultiColumnWindow.java} | 58 +- ...RecordListBackedMultiColumnWindowIterator.java} | 28 +- ...ticSerializableTVListBackedSingleColumnRow.java | 99 + ...SerializableTVListBackedSingleColumnWindow.java | 83 + ...bleTVListBackedSingleColumnWindowIterator.java} | 32 +- .../LayerPointReaderBackedSingleColumnRow.java | 91 + .../db/query/udf/core/executor/UDTFExecutor.java | 12 +- .../iotdb/db/query/udf/core/input/InputLayer.java | 571 ------ .../iotdb/db/query/udf/core/layer/DAGBuilder.java | 106 ++ .../db/query/udf/core/layer/IntermediateLayer.java | 80 + .../db/query/udf/core/layer/LayerCacheUtils.java | 104 ++ .../query/udf/core/layer/LayerMemoryAssigner.java | 69 + .../layer/MultiInputColumnIntermediateLayer.java | 342 ++++ .../query/udf/core/layer/RawQueryInputLayer.java | 197 ++ .../udf/core/{input => layer}/SafetyLine.java | 2 +- ...InputColumnMultiReferenceIntermediateLayer.java | 336 ++++ ...nputColumnSingleReferenceIntermediateLayer.java | 250 +++ .../db/query/udf/core/reader/LayerRowReader.java | 2 +- .../core/transformer/RawQueryPointTransformer.java | 73 - .../udf/core/transformer/UDFQueryTransformer.java | 2 +- .../primitive/ElasticSerializableIntList.java | 99 - .../primitive/SerializableIntList.java | 133 -- .../row/ElasticSerializableRowRecordList.java | 9 +- .../tv/ElasticSerializableTVList.java | 4 +- .../query/udf/service/UDFRegistrationService.java | 4 +- .../org/apache/iotdb/db/service/TSServiceImpl.java | 66 +- .../db/sync/receiver/transfer/SyncServiceImpl.java | 4 +- .../iotdb/db/sync/sender/transfer/SyncClient.java | 17 +- .../apache/iotdb/db/tools/TsFileRewriteTool.java | 14 +- .../db/tools/upgrade/TsFileOnlineUpgradeTool.java | 8 +- .../db/tools/virtualsg/DeviceMappingViewer.java | 2 +- .../org/apache/iotdb/db/utils/FilePathUtils.java | 4 +- .../org/apache/iotdb/db/utils/SchemaUtils.java | 6 +- .../iotdb/db/utils/datastructure/BinaryTVList.java | 59 +- .../db/utils/datastructure/BooleanTVList.java | 59 +- .../iotdb/db/utils/datastructure/DoubleTVList.java | 59 +- .../iotdb/db/utils/datastructure/FloatTVList.java | 59 +- .../iotdb/db/utils/datastructure/IntTVList.java | 59 +- .../iotdb/db/utils/datastructure/LongTVList.java | 69 +- .../iotdb/db/utils/datastructure/TVList.java | 16 +- .../iotdb/db/utils/datastructure/VectorTVList.java | 82 +- .../iotdb/db/writelog/recover/LogReplayer.java | 73 +- .../writelog/recover/TsFileRecoverPerformer.java | 58 +- .../compaction/LevelCompactionMoreDataTest.java | 10 +- .../db/engine/compaction/LevelCompactionTest.java | 12 +- .../db/engine/memtable/MemTableTestUtils.java | 13 +- .../db/engine/memtable/MemtableBenchmark.java | 4 +- .../db/engine/memtable/PrimitiveMemTableTest.java | 20 +- .../iotdb/db/engine/merge/MergeOverLapTest.java | 4 +- .../iotdb/db/engine/merge/MergeTaskTest.java | 4 +- .../apache/iotdb/db/engine/merge/MergeTest.java | 12 +- .../iotdb/db/engine/merge/MergeUpgradeTest.java | 10 +- .../storagegroup/StorageGroupProcessorTest.java | 36 +- .../iotdb/db/engine/storagegroup/TTLTest.java | 16 +- .../engine/storagegroup/TsFileProcessorTest.java | 22 +- .../iotdb/db/integration/IoTDBAddSubDeviceIT.java | 249 --- .../iotdb/db/integration/IoTDBAlignByDeviceIT.java | 27 +- .../db/integration/IoTDBAutoCreateSchemaIT.java | 50 - .../apache/iotdb/db/integration/IoTDBCloseIT.java | 2 +- .../iotdb/db/integration/IoTDBCompleteIT.java | 26 +- .../apache/iotdb/db/integration/IoTDBDaemonIT.java | 2 +- .../db/integration/IoTDBDeleteStorageGroupIT.java | 2 +- .../db/integration/IoTDBDeleteTimeseriesIT.java | 4 +- .../iotdb/db/integration/IoTDBDeletionIT.java | 6 +- .../iotdb/db/integration/IoTDBDisableAlignIT.java | 6 +- .../iotdb/db/integration/IoTDBExecuteBatchIT.java | 4 +- .../db/integration/IoTDBFloatPrecisionIT.java | 4 +- .../db/integration/IoTDBFlushQueryMergeIT.java | 2 +- .../db/integration/IoTDBInsertWithQueryIT.java | 4 +- .../iotdb/db/integration/IoTDBLargeDataIT.java | 4 +- .../apache/iotdb/db/integration/IoTDBLastIT.java | 2 +- ...IoTDBLoadExternalTsFileWithTimePartitionIT.java | 141 +- .../db/integration/IoTDBLoadExternalTsfileIT.java | 31 +- .../iotdb/db/integration/IoTDBMetadataFetchIT.java | 10 +- .../iotdb/db/integration/IoTDBMultiDeviceIT.java | 14 +- .../iotdb/db/integration/IoTDBMultiSeriesIT.java | 7 +- .../db/integration/IoTDBMultiStatementsIT.java | 3 +- .../iotdb/db/integration/IoTDBNestedQueryIT.java | 388 ++++ .../iotdb/db/integration/IoTDBQueryDemoIT.java | 13 +- .../db/integration/IoTDBQueryMemoryControlIT.java | 6 +- .../iotdb/db/integration/IoTDBRestartIT.java | 4 +- .../db/integration/IoTDBRpcCompressionIT.java | 6 +- .../IoTDBSameMeasurementsDifferentTypesIT.java | 3 +- .../iotdb/db/integration/IoTDBSelectIntoIT.java | 33 +- .../iotdb/db/integration/IoTDBSensorUpdateIT.java | 4 +- .../db/integration/IoTDBSequenceDataQueryIT.java | 2 +- .../IoTDBSetSystemReadOnlyWritableIT.java | 2 +- .../iotdb/db/integration/IoTDBSimpleQueryIT.java | 26 +- .../apache/iotdb/db/integration/IoTDBTagIT.java | 9 +- .../iotdb/db/integration/IoTDBTimeZoneIT.java | 4 +- .../db/integration/IoTDBTriggerManagementIT.java | 2 +- .../apache/iotdb/db/integration/IoTDBTtlIT.java | 2 +- .../iotdb/db/integration/IoTDBUDFManagementIT.java | 21 + .../db/integration/IoTDBUDFWindowQueryIT.java | 8 +- .../integration/IoTDBUDTFAlignByTimeQueryIT.java | 6 +- .../db/integration/IoTDBUDTFBuiltinFunctionIT.java | 79 +- .../db/integration/IoTDBUDTFHybridQueryIT.java | 2 - .../db/integration/IoTDBUDTFNonAlignQueryIT.java | 6 +- .../aggregation/IoTDBAggregationSmallDataIT.java | 6 +- .../db/integration/auth/IoTDBAuthorizationIT.java | 6 +- .../iotdb/db/metadata/MManagerAdvancedTest.java | 32 +- .../iotdb/db/metadata/MManagerBasicTest.java | 162 +- .../iotdb/db/metadata/MManagerImproveTest.java | 2 +- .../org/apache/iotdb/db/metadata/MTreeTest.java | 276 ++- .../apache/iotdb/db/metadata/MetaUtilsTest.java | 11 + .../apache/iotdb/db/metadata/PartialPathTest.java | 8 + .../apache/iotdb/db/metadata/mnode/MNodeTest.java | 2 +- ...rmatTest.java => JSONPayloadFormatterTest.java} | 2 +- .../java/org/apache/iotdb/db/qp/PlannerTest.java | 3 +- .../iotdb/db/qp/logical/LogicalPlanSmallTest.java | 6 +- .../iotdb/db/qp/physical/InsertRowPlanTest.java | 2 + .../db/qp/physical/InsertTabletMultiPlanTest.java | 3 +- .../iotdb/db/qp/physical/InsertTabletPlanTest.java | 112 +- .../iotdb/db/qp/physical/PhysicalPlanTest.java | 2 +- .../iotdb/db/query/dataset/SingleDataSetTest.java | 2 +- .../query/dataset/UDTFAlignByTimeDataSetTest.java | 18 +- .../reader/series/SeriesAggregateReaderTest.java | 14 +- .../reader/series/SeriesReaderByTimestampTest.java | 4 +- .../db/query/reader/series/SeriesReaderTest.java | 4 +- .../query/reader/series/SeriesReaderTestUtil.java | 19 +- .../ElasticSerializableTVListTest.java | 2 +- .../iotdb/db/query/udf/example/Accumulator.java | 12 +- .../apache/iotdb/db/query/udf/example/Adder.java | 4 +- .../apache/iotdb/db/query/udf/example/Counter.java | 24 +- .../org/apache/iotdb/db/query/udf/example/Max.java | 2 +- .../apache/iotdb/db/sink/LocalIoTDBSinkTest.java | 4 +- .../org/apache/iotdb/db/sink/MQTTSinkTest.java | 4 +- .../org/apache/iotdb/db/tools/MLogParserTest.java | 18 +- .../iotdb/db/tools/TsFileSketchToolTest.java | 8 +- .../iotdb/db/utils/TsFileRewriteToolTest.java | 8 +- .../db/utils/datastructure/BinaryTVListTest.java | 59 +- .../db/utils/datastructure/BooleanTVListTest.java | 63 +- .../db/utils/datastructure/DoubleTVListTest.java | 63 +- .../db/utils/datastructure/FloatTVListTest.java | 63 +- .../db/utils/datastructure/IntTVListTest.java | 63 +- .../db/utils/datastructure/LongTVListTest.java | 62 +- .../db/utils/datastructure/PrecisionTest.java | 8 +- .../db/utils/datastructure/VectorTVListTest.java | 40 +- .../db/writelog/recover/DeviceStringTest.java | 8 +- .../iotdb/db/writelog/recover/LogReplayerTest.java | 12 +- .../recover/RecoverResourceFromReaderTest.java | 12 +- .../db/writelog/recover/SeqTsFileRecoverTest.java | 257 ++- .../writelog/recover/UnseqTsFileRecoverTest.java | 12 +- .../java/org/apache/iotdb/rpc/TSStatusCode.java | 20 + .../org/apache/iotdb/rpc/TSStatusCodeTest.java | 18 +- .../java/org/apache/iotdb/session/Session.java | 334 +++- .../apache/iotdb/session/SessionConnection.java | 6 +- .../org/apache/iotdb/session/SessionUtils.java | 4 +- .../org/apache/iotdb/session/pool/SessionPool.java | 189 ++ .../iotdb/session/IoTDBSessionComplexIT.java | 26 +- .../apache/iotdb/session/IoTDBSessionSimpleIT.java | 81 +- .../session/IoTDBSessionVectorABDeviceIT.java | 231 +++ .../session/IoTDBSessionVectorAggregationIT.java | 266 +++ .../IoTDBSessionVectorAggregationWithUnSeqIT.java | 192 ++ .../apache/iotdb/session/IoTDBSessionVectorIT.java | 212 --- .../iotdb/session/IoTDBSessionVectorInsertIT.java | 449 +++++ .../apache/iotdb/session/SessionCacheLeaderUT.java | 30 +- .../java/org/apache/iotdb/session/SessionTest.java | 12 +- site/src/main/.vuepress/config.js | 16 +- .../org/apache/iotdb/spark/db/IoTDBTest.scala | 14 +- .../iotdb/spark/tsfile/NarrowConverter.scala | 4 +- .../apache/iotdb/spark/tsfile/WideConverter.scala | 4 +- .../org/apache/iotdb/spark/tool/TsFileExample.java | 10 +- .../apache/iotdb/spark/tool/TsFileWriteTool.java | 22 +- .../test/java/org/apache/iotdb/db/sql/Cases.java | 14 +- thrift/src/main/thrift/rpc.thrift | 15 +- .../iotdb/tsfile/file/header/PageHeader.java | 15 +- .../iotdb/tsfile/file/metadata/ChunkMetadata.java | 13 +- .../iotdb/tsfile/file/metadata/IChunkMetadata.java | 3 +- .../tsfile/file/metadata/TimeseriesMetadata.java | 9 +- .../tsfile/file/metadata/VectorChunkMetadata.java | 4 + .../file/metadata/VectorTimeSeriesMetadata.java | 4 + .../file/metadata/statistics/BinaryStatistics.java | 2 +- .../metadata/statistics/BooleanStatistics.java | 2 +- .../file/metadata/statistics/DoubleStatistics.java | 2 +- .../file/metadata/statistics/FloatStatistics.java | 2 +- .../metadata/statistics/IntegerStatistics.java | 2 +- .../file/metadata/statistics/LongStatistics.java | 2 +- .../file/metadata/statistics/Statistics.java | 29 +- .../file/metadata/statistics/TimeStatistics.java | 4 +- .../iotdb/tsfile/read/TsFileSequenceReader.java | 8 +- .../apache/iotdb/tsfile/read/common/BatchData.java | 84 +- .../tsfile/read/common/DescReadWriteBatchData.java | 42 + .../tsfile/read/common/IBatchDataIterator.java | 16 +- .../tsfile/read/reader/page/VectorPageReader.java | 6 +- .../java/org/apache/iotdb/tsfile/utils/BitMap.java | 22 + .../iotdb/tsfile/v2/file/header/PageHeaderV2.java | 5 +- .../tsfile/v2/file/metadata/ChunkMetadataV2.java | 3 +- .../v2/file/metadata/statistics/StatisticsV2.java | 14 +- .../tsfile/write/chunk/ChunkGroupWriterImpl.java | 4 +- .../iotdb/tsfile/write/chunk/ChunkWriterImpl.java | 7 +- .../iotdb/tsfile/write/chunk/ValueChunkWriter.java | 3 +- .../apache/iotdb/tsfile/write/page/PageWriter.java | 5 +- .../iotdb/tsfile/write/page/ValuePageWriter.java | 5 +- .../apache/iotdb/tsfile/write/record/Tablet.java | 13 +- ...mentSchema.java => UnaryMeasurementSchema.java} | 32 +- .../iotdb/tsfile/write/writer/TsFileIOWriter.java | 3 +- .../tsfile/file/metadata/utils/TestHelper.java | 14 +- .../iotdb/tsfile/read/ReadOnlyTsFileTest.java | 5 +- .../read/query/timegenerator/ReadWriteTest.java | 8 +- .../timegenerator/TimeGeneratorReadEmptyTest.java | 6 +- .../timegenerator/TimeGeneratorReadWriteTest.java | 6 +- .../TsFileGeneratorForSeriesReaderByTimestamp.java | 27 +- .../apache/iotdb/tsfile/utils/FileGenerator.java | 27 +- .../apache/iotdb/tsfile/utils/RecordUtilsTest.java | 19 +- .../iotdb/tsfile/utils/TsFileGeneratorForTest.java | 26 +- .../tsfile/write/DefaultSchemaTemplateTest.java | 8 +- .../tsfile/write/MetadataIndexConstructorTest.java | 4 +- .../org/apache/iotdb/tsfile/write/PerfTest.java | 24 +- .../iotdb/tsfile/write/ReadPageInMemTest.java | 26 +- ...SameMeasurementsWithDifferentDataTypesTest.java | 12 +- .../iotdb/tsfile/write/TsFileIOWriterTest.java | 4 +- .../iotdb/tsfile/write/TsFileReadWriteTest.java | 8 +- .../iotdb/tsfile/write/TsFileWriterTest.java | 22 +- .../org/apache/iotdb/tsfile/write/WriteTest.java | 16 +- .../write/schema/converter/SchemaBuilderTest.java | 16 +- .../write/writer/ForceAppendTsFileWriterTest.java | 10 +- .../writer/MeasurementSchemaSerializeTest.java | 14 +- .../iotdb/tsfile/write/writer/PageWriterTest.java | 14 +- .../write/writer/RestorableTsFileIOWriterTest.java | 52 +- 428 files changed, 16188 insertions(+), 8879 deletions(-) diff --cc cluster/src/main/java/org/apache/iotdb/cluster/ClusterIoTDB.java index 57d9304,0000000..bde0f05 mode 100644,000000..100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/ClusterIoTDB.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/ClusterIoTDB.java @@@ -1,667 -1,0 +1,667 @@@ +/* + * 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.cluster; + +import org.apache.iotdb.cluster.client.ClientCategory; +import org.apache.iotdb.cluster.client.ClientManager; +import org.apache.iotdb.cluster.client.IClientManager; +import org.apache.iotdb.cluster.client.async.AsyncDataClient; +import org.apache.iotdb.cluster.client.async.AsyncMetaClient; +import org.apache.iotdb.cluster.client.sync.SyncClientAdaptor; +import org.apache.iotdb.cluster.client.sync.SyncDataClient; +import org.apache.iotdb.cluster.config.ClusterConfig; +import org.apache.iotdb.cluster.config.ClusterConstant; +import org.apache.iotdb.cluster.config.ClusterDescriptor; +import org.apache.iotdb.cluster.coordinator.Coordinator; +import org.apache.iotdb.cluster.exception.ConfigInconsistentException; +import org.apache.iotdb.cluster.exception.StartUpCheckFailureException; +import org.apache.iotdb.cluster.metadata.CMManager; +import org.apache.iotdb.cluster.metadata.MetaPuller; +import org.apache.iotdb.cluster.partition.slot.SlotPartitionTable; +import org.apache.iotdb.cluster.partition.slot.SlotStrategy; +import org.apache.iotdb.cluster.rpc.thrift.Node; +import org.apache.iotdb.cluster.server.ClusterRPCService; +import org.apache.iotdb.cluster.server.ClusterTSServiceImpl; +import org.apache.iotdb.cluster.server.HardLinkCleaner; +import org.apache.iotdb.cluster.server.Response; +import org.apache.iotdb.cluster.server.clusterinfo.ClusterInfoServer; +import org.apache.iotdb.cluster.server.member.MetaGroupMember; +import org.apache.iotdb.cluster.server.monitor.NodeReport; +import org.apache.iotdb.cluster.server.raft.DataRaftHeartBeatService; +import org.apache.iotdb.cluster.server.raft.DataRaftService; +import org.apache.iotdb.cluster.server.raft.MetaRaftHeartBeatService; +import org.apache.iotdb.cluster.server.raft.MetaRaftService; +import org.apache.iotdb.cluster.server.service.DataGroupEngine; +import org.apache.iotdb.cluster.server.service.DataGroupServiceImpls; +import org.apache.iotdb.cluster.server.service.MetaAsyncService; +import org.apache.iotdb.cluster.server.service.MetaSyncService; +import org.apache.iotdb.cluster.utils.ClusterUtils; +import org.apache.iotdb.cluster.utils.nodetool.ClusterMonitor; +import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory; +import org.apache.iotdb.db.conf.IoTDBConfigCheck; +import org.apache.iotdb.db.conf.IoTDBConstant; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.exception.ConfigurationException; +import org.apache.iotdb.db.exception.StartupException; +import org.apache.iotdb.db.exception.query.QueryProcessException; +import org.apache.iotdb.db.service.IoTDB; +import org.apache.iotdb.db.service.JMXService; +import org.apache.iotdb.db.service.RegisterManager; +import org.apache.iotdb.db.service.thrift.ThriftServiceThread; +import org.apache.iotdb.db.utils.TestOnly; + +import org.apache.thrift.TException; +import org.apache.thrift.async.TAsyncClientManager; +import org.apache.thrift.protocol.TBinaryProtocol.Factory; +import org.apache.thrift.protocol.TCompactProtocol; +import org.apache.thrift.protocol.TProtocolFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +import static org.apache.iotdb.cluster.config.ClusterConstant.THREAD_POLL_WAIT_TERMINATION_TIME_S; +import static org.apache.iotdb.cluster.utils.ClusterUtils.UNKNOWN_CLIENT_IP; + +// we do not inherent IoTDB instance, as it may break the singleton mode of IoTDB. +public class ClusterIoTDB implements ClusterIoTDBMBean { + + private static final Logger logger = LoggerFactory.getLogger(ClusterIoTDB.class); + private final String mbeanName = + String.format( + "%s:%s=%s", "org.apache.iotdb.cluster.service", IoTDBConstant.JMX_TYPE, "ClusterIoTDB"); + + // TODO fix me: better to throw exception if the client can not be get. Then we can remove this + // field. + public static boolean printClientConnectionErrorStack = false; + + // establish the cluster as a seed + private static final String MODE_START = "-s"; + // join an established cluster + private static final String MODE_ADD = "-a"; + // send a request to remove a node, more arguments: ip-of-removed-node + // metaport-of-removed-node + private static final String MODE_REMOVE = "-r"; + + private MetaGroupMember metaGroupEngine; + + // split DataGroupServiceImpls into engine and impls + private DataGroupEngine dataGroupEngine; + + private Node thisNode; + private Coordinator coordinator; + + private final IoTDB iotdb = IoTDB.getInstance(); + + // Cluster IoTDB uses a individual registerManager with its parent. + private RegisterManager registerManager = new RegisterManager(); + + /** + * a single thread pool, every "REPORT_INTERVAL_SEC" seconds, "reportThread" will print the status + * of all raft members in this node + */ + private ScheduledExecutorService reportThread; + + private boolean allowReport = true; + + /** hardLinkCleaner will periodically clean expired hardlinks created during snapshots */ + private ScheduledExecutorService hardLinkCleanerThread; + + // currently, clientManager is only used for those instances who do not belong to any + // DataGroup.. + private IClientManager clientManager; + + private ClusterIoTDB() { + // we do not init anything here, so that we can re-initialize the instance in IT. + } + + public void initLocalEngines() { + ClusterConfig config = ClusterDescriptor.getInstance().getConfig(); + thisNode = new Node(); + // set internal rpc ip and ports + thisNode.setInternalIp(config.getInternalIp()); + thisNode.setMetaPort(config.getInternalMetaPort()); + thisNode.setDataPort(config.getInternalDataPort()); + // set client rpc ip and ports + thisNode.setClientPort(config.getClusterRpcPort()); + thisNode.setClientIp(IoTDBDescriptor.getInstance().getConfig().getRpcAddress()); + coordinator = new Coordinator(); + // local engine + TProtocolFactory protocolFactory = + ThriftServiceThread.getProtocolFactory( + IoTDBDescriptor.getInstance().getConfig().isRpcThriftCompressionEnable()); + metaGroupEngine = new MetaGroupMember(protocolFactory, thisNode, coordinator); + IoTDB.setClusterMode(); + IoTDB.setMetaManager(CMManager.getInstance()); + ((CMManager) IoTDB.metaManager).setMetaGroupMember(metaGroupEngine); + ((CMManager) IoTDB.metaManager).setCoordinator(coordinator); + MetaPuller.getInstance().init(metaGroupEngine); + + // from the scope of the DataGroupEngine,it should be singleton pattern + // the way of setting MetaGroupMember in DataGroupEngine may need a better modification in + // future commit. + DataGroupEngine.setProtocolFactory(protocolFactory); + DataGroupEngine.setMetaGroupMember(metaGroupEngine); + dataGroupEngine = DataGroupEngine.getInstance(); + clientManager = + new ClientManager( + ClusterDescriptor.getInstance().getConfig().isUseAsyncServer(), + ClientManager.Type.RequestForwardClient); + initTasks(); + try { + // we need to check config after initLocalEngines. + startServerCheck(); + } catch (StartupException e) { + logger.error("Failed to check cluster config.", e); + stop(); + } + JMXService.registerMBean(metaGroupEngine, metaGroupEngine.getMBeanName()); + } + + private void initTasks() { + reportThread = IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor("NodeReportThread"); + reportThread.scheduleAtFixedRate( + this::generateNodeReport, + ClusterConstant.REPORT_INTERVAL_SEC, + ClusterConstant.REPORT_INTERVAL_SEC, + TimeUnit.SECONDS); + hardLinkCleanerThread = + IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor("HardLinkCleaner"); + hardLinkCleanerThread.scheduleAtFixedRate( + new HardLinkCleaner(), + ClusterConstant.CLEAN_HARDLINK_INTERVAL_SEC, + ClusterConstant.CLEAN_HARDLINK_INTERVAL_SEC, + TimeUnit.SECONDS); + } + + /** + * Generate a report containing the status of both MetaGroupMember and DataGroupMembers of this + * node. This will help to see if the node is in a consistent and right state during debugging. + */ + private void generateNodeReport() { + if (logger.isDebugEnabled() && allowReport) { + try { + NodeReport report = new NodeReport(thisNode); + report.setMetaMemberReport(metaGroupEngine.genMemberReport()); + report.setDataMemberReportList(dataGroupEngine.genMemberReports()); + logger.debug(report.toString()); + } catch (Exception e) { + logger.error("exception occurred when generating node report", e); + } + } + } + + public static void main(String[] args) { + if (args.length < 1) { + logger.error( + "Usage: <-s|-a|-r> " + + "[-D{} <configure folder>] \n" + + "-s: start the node as a seed\n" + + "-a: start the node as a new node\n" + + "-r: remove the node out of the cluster\n", + IoTDBConstant.IOTDB_CONF); + return; + } + + ClusterIoTDB cluster = ClusterIoTDBHolder.INSTANCE; + // check config of iotdb,and set some configs in cluster mode + try { + if (!cluster.serverCheckAndInit()) { + return; + } + } catch (ConfigurationException | IOException e) { + logger.error("meet error when doing start checking", e); + return; + } + String mode = args[0]; + logger.info("Running mode {}", mode); + + // initialize the current node and its services + cluster.initLocalEngines(); + + // we start IoTDB kernel first. then we start the cluster module. + if (MODE_START.equals(mode)) { + cluster.activeStartNodeMode(); + } else if (MODE_ADD.equals(mode)) { + cluster.activeAddNodeMode(); + } else if (MODE_REMOVE.equals(mode)) { + try { + cluster.doRemoveNode(args); + } catch (IOException e) { + logger.error("Fail to remove node in cluster", e); + } + } else { + logger.error("Unrecognized mode {}", mode); + } + } + + private boolean serverCheckAndInit() throws ConfigurationException, IOException { + IoTDBConfigCheck.getInstance().checkConfig(); + // init server's configuration first, because the cluster configuration may read settings from + // the server's configuration. + IoTDBDescriptor.getInstance().getConfig().setSyncEnable(false); + // auto create schema is took over by cluster module, so we disable it in the server module. + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(false); + // check cluster config + String checkResult = clusterConfigCheck(); + if (checkResult != null) { + logger.error(checkResult); + return false; + } + return true; + } + + private String clusterConfigCheck() { + try { + ClusterDescriptor.getInstance().replaceHostnameWithIp(); + } catch (Exception e) { + return String.format("replace hostname with ip failed, %s", e.getMessage()); + } + ClusterConfig config = ClusterDescriptor.getInstance().getConfig(); + // check the initial replicateNum and refuse to start when the replicateNum <= 0 + if (config.getReplicationNum() <= 0) { + return String.format( + "ReplicateNum should be greater than 0 instead of %d.", config.getReplicationNum()); + } + // check the initial cluster size and refuse to start when the size < quorum + int quorum = config.getReplicationNum() / 2 + 1; + if (config.getSeedNodeUrls().size() < quorum) { + return String.format( + "Seed number less than quorum, seed number: %s, quorum: " + "%s.", + config.getSeedNodeUrls().size(), quorum); + } + // TODO duplicate code,consider to solve it later + Set<Node> seedNodes = new HashSet<>(); + for (String url : config.getSeedNodeUrls()) { + Node node = ClusterUtils.parseNode(url); + if (seedNodes.contains(node)) { + return String.format( + "SeedNodes must not repeat each other. SeedNodes: %s", config.getSeedNodeUrls()); + } + seedNodes.add(node); + } + return null; + } + + public void activeStartNodeMode() { + try { + // start iotdb server first + IoTDB.getInstance().active(); + // some work about cluster + preInitCluster(); + // try to build cluster + metaGroupEngine.buildCluster(); + // register service after cluster build + postInitCluster(); + // init ServiceImpl to handle request of client + startClientRPC(); + } catch (StartupException + | StartUpCheckFailureException + | ConfigInconsistentException + | QueryProcessException e) { + logger.error("Fail to start server", e); + stop(); + } + } + + private void preInitCluster() throws StartupException { + stopRaftInfoReport(); + preStartCustomize(); + JMXService.registerMBean(this, mbeanName); + // register MetaGroupMember. MetaGroupMember has the same position with "StorageEngine" in the + // cluster moduel. + // TODO fixme it is better to remove coordinator out of metaGroupEngine + + registerManager.register(metaGroupEngine); + registerManager.register(dataGroupEngine); + + // rpc service initialize + DataGroupServiceImpls dataGroupServiceImpls = new DataGroupServiceImpls(); + if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) { + MetaAsyncService metaAsyncService = new MetaAsyncService(metaGroupEngine); + MetaRaftHeartBeatService.getInstance().initAsyncedServiceImpl(metaAsyncService); + MetaRaftService.getInstance().initAsyncedServiceImpl(metaAsyncService); + DataRaftService.getInstance().initAsyncedServiceImpl(dataGroupServiceImpls); + DataRaftHeartBeatService.getInstance().initAsyncedServiceImpl(dataGroupServiceImpls); + } else { + MetaSyncService syncService = new MetaSyncService(metaGroupEngine); + MetaRaftHeartBeatService.getInstance().initSyncedServiceImpl(syncService); + MetaRaftService.getInstance().initSyncedServiceImpl(syncService); + DataRaftService.getInstance().initSyncedServiceImpl(dataGroupServiceImpls); + DataRaftHeartBeatService.getInstance().initSyncedServiceImpl(dataGroupServiceImpls); + } + // start RPC service + logger.info("start Meta Heartbeat RPC service... "); + registerManager.register(MetaRaftHeartBeatService.getInstance()); + // TODO: better to start the Meta RPC service untill the heartbeatservice has elected the + // leader. + // and quorum of followers have caught up. + logger.info("start Meta RPC service... "); + registerManager.register(MetaRaftService.getInstance()); + } + + private void postInitCluster() throws StartupException, QueryProcessException { + logger.info("start Data Heartbeat RPC service... "); + registerManager.register(DataRaftHeartBeatService.getInstance()); + logger.info("start Data RPC service... "); + registerManager.register(DataRaftService.getInstance()); + // RPC based DBA API + registerManager.register(ClusterInfoServer.getInstance()); + // JMX based DBA API + registerManager.register(ClusterMonitor.INSTANCE); + } + + private void startClientRPC() throws QueryProcessException, StartupException { + // we must wait until the metaGroup established. + // So that the ClusterRPCService can work. + ClusterTSServiceImpl clusterRPCServiceImpl = new ClusterTSServiceImpl(); + clusterRPCServiceImpl.setCoordinator(coordinator); + clusterRPCServiceImpl.setExecutor(metaGroupEngine); + ClusterRPCService.getInstance().initSyncedServiceImpl(clusterRPCServiceImpl); + registerManager.register(ClusterRPCService.getInstance()); + } + + public void activeAddNodeMode() { + try { + long startTime = System.currentTimeMillis(); + + preInitCluster(); + metaGroupEngine.joinCluster(); + postInitCluster(); + dataGroupEngine.pullSnapshots(); + startClientRPC(); + logger.info( + "Adding this node {} to cluster costs {} ms", + thisNode, + (System.currentTimeMillis() - startTime)); + } catch (StartupException + | QueryProcessException + | StartUpCheckFailureException + | ConfigInconsistentException e) { + stop(); + logger.error("Fail to join cluster", e); + } + } + + private void startServerCheck() throws StartupException { + ClusterConfig config = ClusterDescriptor.getInstance().getConfig(); + // assert not duplicated nodes + Set<Node> seedNodes = new HashSet<>(); + for (String url : config.getSeedNodeUrls()) { + Node node = ClusterUtils.parseNode(url); + if (seedNodes.contains(node)) { + String message = + String.format( + "SeedNodes must not repeat each other. SeedNodes: %s", config.getSeedNodeUrls()); + throw new StartupException(metaGroupEngine.getName(), message); + } + seedNodes.add(node); + } + + // assert this node is in all nodes when restart + if (!metaGroupEngine.getAllNodes().isEmpty()) { + if (!metaGroupEngine.getAllNodes().contains(metaGroupEngine.getThisNode())) { + String message = + String.format( + "All nodes in partitionTables must contains local node in start-server mode. " + + "LocalNode: %s, AllNodes: %s", + metaGroupEngine.getThisNode(), metaGroupEngine.getAllNodes()); + throw new StartupException(metaGroupEngine.getName(), message); + } else { + return; + } + } + + // assert this node is in seed nodes list + if (!seedNodes.contains(thisNode)) { + String message = + String.format( + "SeedNodes must contains local node in start-server mode. LocalNode: %s ,SeedNodes: %s", + thisNode.toString(), config.getSeedNodeUrls()); + throw new StartupException(metaGroupEngine.getName(), message); + } + } + + private void doRemoveNode(String[] args) throws IOException { + if (args.length != 3) { - logger.error("Usage: -r <ip> <metaPort>"); ++ logger.error("Usage: <ip> <metaPort>"); + return; + } + String ip = args[1]; + int metaPort = Integer.parseInt(args[2]); + ClusterConfig config = ClusterDescriptor.getInstance().getConfig(); + TProtocolFactory factory = + config.isRpcThriftCompressionEnabled() ? new TCompactProtocol.Factory() : new Factory(); + Node nodeToRemove = new Node(); + nodeToRemove.setInternalIp(ip).setMetaPort(metaPort).setClientIp(UNKNOWN_CLIENT_IP); + // try sending the request to each seed node + for (String url : config.getSeedNodeUrls()) { + Node node = ClusterUtils.parseNode(url); + if (node == null) { + continue; + } + AsyncMetaClient client = new AsyncMetaClient(factory, new TAsyncClientManager(), node, null); + Long response = null; + long startTime = System.currentTimeMillis(); + try { + logger.info("Start removing node {} with the help of node {}", nodeToRemove, node); + response = SyncClientAdaptor.removeNode(client, nodeToRemove); + } catch (TException e) { + logger.warn("Cannot send remove node request through {}, try next node", node); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + logger.warn("Cannot send remove node request through {}, try next node", node); + } + if (response != null) { + handleNodeRemovalResp(response, nodeToRemove, startTime); + return; + } + } + } + + private void handleNodeRemovalResp(Long response, Node nodeToRemove, long startTime) { + if (response == Response.RESPONSE_AGREE) { + logger.info( + "Node {} is successfully removed, cost {}ms", + nodeToRemove, + (System.currentTimeMillis() - startTime)); + } else if (response == Response.RESPONSE_CLUSTER_TOO_SMALL) { + logger.error("Cluster size is too small, cannot remove any node"); + } else if (response == Response.RESPONSE_REJECT) { + logger.error("Node {} is not found in the cluster, please check", nodeToRemove); + } else if (response == Response.RESPONSE_DATA_MIGRATION_NOT_FINISH) { + logger.warn( + "The data migration of the previous membership change operation is not finished. Please try again later"); + } else { + logger.error("Unexpected response {}", response); + } + } + + /** Developers may perform pre-start customizations here for debugging or experiments. */ + @SuppressWarnings("java:S125") // leaving examples + private void preStartCustomize() { + // customize data distribution + // The given example tries to divide storage groups like "root.sg_1", "root.sg_2"... into k + // nodes evenly, and use default strategy for other groups + SlotPartitionTable.setSlotStrategy( + new SlotStrategy() { + SlotStrategy defaultStrategy = new SlotStrategy.DefaultStrategy(); + int k = 3; + + @Override + public int calculateSlotByTime(String storageGroupName, long timestamp, int maxSlotNum) { + int sgSerialNum = extractSerialNumInSGName(storageGroupName) % k; + if (sgSerialNum >= 0) { + return maxSlotNum / k * sgSerialNum; + } else { + return defaultStrategy.calculateSlotByTime(storageGroupName, timestamp, maxSlotNum); + } + } + + @Override + public int calculateSlotByPartitionNum( + String storageGroupName, long partitionId, int maxSlotNum) { + int sgSerialNum = extractSerialNumInSGName(storageGroupName) % k; + if (sgSerialNum >= 0) { + return maxSlotNum / k * sgSerialNum; + } else { + return defaultStrategy.calculateSlotByPartitionNum( + storageGroupName, partitionId, maxSlotNum); + } + } + + private int extractSerialNumInSGName(String storageGroupName) { + String[] s = storageGroupName.split("_"); + if (s.length != 2) { + return -1; + } + try { + return Integer.parseInt(s[1]); + } catch (NumberFormatException e) { + return -1; + } + } + }); + } + + public void stop() { + deactivate(); + } + + private void deactivate() { + logger.info("Deactivating Cluster IoTDB..."); + stopThreadPools(); + registerManager.deregisterAll(); + JMXService.deregisterMBean(mbeanName); + logger.info("ClusterIoTDB is deactivated."); + // stop the iotdb kernel + iotdb.stop(); + } + + private void stopThreadPools() { + stopThreadPool(reportThread, "reportThread"); + stopThreadPool(hardLinkCleanerThread, "hardLinkCleanerThread"); + } + + private void stopThreadPool(ExecutorService pool, String name) { + if (pool != null) { + pool.shutdownNow(); + try { + pool.awaitTermination(THREAD_POLL_WAIT_TERMINATION_TIME_S, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + logger.error("Unexpected interruption when waiting for {} to end", name, e); + } + } + } + + @TestOnly + public void setClientManager(IClientManager clientManager) { + this.clientManager = clientManager; + } + + public MetaGroupMember getMetaGroupEngine() { + return metaGroupEngine; + } + + public Node getThisNode() { + return thisNode; + } + + public Coordinator getCoordinator() { + return coordinator; + } + + public IoTDB getIotdb() { + return iotdb; + } + + public RegisterManager getRegisterManager() { + return registerManager; + } + + public DataGroupEngine getDataGroupEngine() { + return dataGroupEngine; + } + + public void setMetaGroupEngine(MetaGroupMember metaGroupEngine) { + this.metaGroupEngine = metaGroupEngine; + } + + public static ClusterIoTDB getInstance() { + return ClusterIoTDBHolder.INSTANCE; + } + + @Override + public boolean startRaftInfoReport() { + logger.info("Raft status report is enabled."); + allowReport = true; + if (logger.isDebugEnabled()) { + return true; + } + return false; + } + + @Override + public void stopRaftInfoReport() { + logger.info("Raft status report is disabled."); + allowReport = false; + } + + @Override + public void enablePrintClientConnectionErrorStack() { + printClientConnectionErrorStack = true; + } + + @Override + public void disablePrintClientConnectionErrorStack() { + printClientConnectionErrorStack = false; + } + + public SyncDataClient getSyncDataClient(Node node, int readOperationTimeoutMS) throws Exception { + SyncDataClient dataClient = + (SyncDataClient) clientManager.borrowSyncClient(node, ClientCategory.DATA); + if (dataClient != null) { + dataClient.setTimeout(readOperationTimeoutMS); + } + return dataClient; + } + + public AsyncDataClient getAsyncDataClient(Node node, int readOperationTimeoutMS) + throws Exception { + AsyncDataClient dataClient = + (AsyncDataClient) clientManager.borrowAsyncClient(node, ClientCategory.DATA); + if (dataClient != null) { + dataClient.setTimeout(readOperationTimeoutMS); + } + return dataClient; + } + + private static class ClusterIoTDBHolder { + + private static final ClusterIoTDB INSTANCE = new ClusterIoTDB(); + + private ClusterIoTDBHolder() {} + } +} diff --cc cluster/src/main/java/org/apache/iotdb/cluster/metadata/MetaPuller.java index 78d516e,6715e23..405bcc9 --- a/cluster/src/main/java/org/apache/iotdb/cluster/metadata/MetaPuller.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/metadata/MetaPuller.java @@@ -227,32 -229,32 +230,32 @@@ public class MetaPuller List<IMeasurementSchema> schemas; if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) { AsyncDataClient client = - metaGroupMember - .getClientProvider() - .getAsyncDataClient(node, RaftServer.getReadOperationTimeoutMS()); + ClusterIoTDB.getInstance() + .getAsyncDataClient(node, ClusterConstant.getReadOperationTimeoutMS()); schemas = SyncClientAdaptor.pullMeasurementSchema(client, request); } else { - try (SyncDataClient syncDataClient = - metaGroupMember - .getClientProvider() - .getSyncDataClient(node, RaftServer.getReadOperationTimeoutMS())) { - try { - // only need measurement name - PullSchemaResp pullSchemaResp = syncDataClient.pullMeasurementSchema(request); - ByteBuffer buffer = pullSchemaResp.schemaBytes; - int size = buffer.getInt(); - schemas = new ArrayList<>(size); - for (int i = 0; i < size; i++) { - schemas.add( - buffer.get() == 0 - ? UnaryMeasurementSchema.partialDeserializeFrom(buffer) - : VectorMeasurementSchema.partialDeserializeFrom(buffer)); - } - } catch (TException e) { - // the connection may be broken, close it to avoid it being reused - syncDataClient.getInputProtocol().getTransport().close(); - throw e; + SyncDataClient syncDataClient = null; + try { + syncDataClient = + ClusterIoTDB.getInstance() + .getSyncDataClient(node, ClusterConstant.getReadOperationTimeoutMS()); + // only need measurement name + PullSchemaResp pullSchemaResp = syncDataClient.pullMeasurementSchema(request); + ByteBuffer buffer = pullSchemaResp.schemaBytes; + int size = buffer.getInt(); + schemas = new ArrayList<>(size); + for (int i = 0; i < size; i++) { + schemas.add( + buffer.get() == 0 - ? MeasurementSchema.partialDeserializeFrom(buffer) ++ ? UnaryMeasurementSchema.partialDeserializeFrom(buffer) + : VectorMeasurementSchema.partialDeserializeFrom(buffer)); } + } catch (TException e) { + // the connection may be broken, close it to avoid it being reused + syncDataClient.close(); + throw e; + } finally { + if (syncDataClient != null) syncDataClient.returnSelf(); } } diff --cc server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfigCheck.java index d9947a5,ce8d55e..baf8af3 --- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfigCheck.java +++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfigCheck.java @@@ -322,11 -327,15 +328,15 @@@ public class IoTDBConfigCheck } if (!(properties.getProperty(TAG_ATTRIBUTE_SIZE_STRING).equals(tagAttributeTotalSize))) { - printErrorLogAndExit(TAG_ATTRIBUTE_SIZE_STRING); + throwException(TAG_ATTRIBUTE_SIZE_STRING, tagAttributeTotalSize); } + if (!(properties.getProperty(TAG_ATTRIBUTE_FLUSH_INTERVAL).equals(tagAttributeFlushInterval))) { + printErrorLogAndExit(TAG_ATTRIBUTE_FLUSH_INTERVAL); + } + if (!(properties.getProperty(MAX_DEGREE_OF_INDEX_STRING).equals(maxDegreeOfIndexNode))) { - printErrorLogAndExit(MAX_DEGREE_OF_INDEX_STRING); + throwException(MAX_DEGREE_OF_INDEX_STRING, maxDegreeOfIndexNode); } if (!(properties.getProperty(VIRTUAL_STORAGE_GROUP_NUM).equals(virtualStorageGroupNum))) {
