This is an automated email from the ASF dual-hosted git repository. jackietien pushed a commit to branch ty/TableModelGrammar in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 4e6364f03ef0686a30e0ebd572869da56d30b0a0 Merge: 9aac8228947 44b263e2a93 Author: JackieTien97 <[email protected]> AuthorDate: Sun Apr 14 14:19:50 2024 +0800 merge master README.md | 62 +- example/pipe-opc-ua-sink/pom.xml | 34 ++ .../apache/iotdb/SubscriptionSessionExample.java | 18 +- .../it/env/cluster/config/MppConfigNodeConfig.java | 6 + .../iotdb/it/env/cluster/env/AbstractEnv.java | 29 +- .../it/env/cluster/node/AbstractNodeWrapper.java | 5 + .../env/remote/config/RemoteConfigNodeConfig.java | 6 + .../apache/iotdb/itbase/env/BaseNodeWrapper.java | 2 + .../apache/iotdb/itbase/env/ConfigNodeConfig.java | 2 + ...oTDBRegionMigrateDataNodeCrashITFramework.java} | 14 +- .../IoTDBRegionMigrateReliabilityITFramework.java | 412 +++++++++---- .../it/regionmigration/KillPointContext.java | 25 +- .../pass/IoTDBRegionMigrateClusterCrashIT.java | 68 +++ .../pass/IoTDBRegionMigrateConfigNodeCrashIT.java | 92 ++- .../pass/IoTDBRegionMigrateNormalIT.java | 8 +- .../pass/IoTDBRegionMigrateOtherIT.java | 9 +- ...ateCoordinatorCrashWhenRemoveRemotePeerIT.java} | 24 +- .../IoTDBRegionMigrateDataNodeCrashIT.java | 59 +- ...MigrateOriginalCrashWhenDeleteLocalPeerIT.java} | 19 +- ...igrateOriginalCrashWhenRemoveRemotePeerIT.java} | 19 +- .../pipe/it/autocreate/AbstractPipeDualAutoIT.java | 4 + .../pipe/it/autocreate/IoTDBPipeClusterIT.java | 75 ++- .../pipe/it/autocreate/IoTDBPipeIdempotentIT.java | 4 + .../pipe/it/autocreate/IoTDBPipeProcessorIT.java | 4 + .../pipe/it/autocreate/IoTDBPipeProtocolIT.java | 12 + .../pipe/it/manual/AbstractPipeDualManualIT.java | 4 + .../pipe/it/manual/IoTDBPipeMetaHistoricalIT.java | 4 + .../it/manual/IoTDBPipeMetaLeaderChangeIT.java | 4 + .../iotdb/pipe/it/single/AbstractPipeSingleIT.java | 2 + .../iotdb/session/it/IoTDBSessionSimpleIT.java | 1 + .../it/dual/AbstractSubscriptionDualIT.java | 4 + .../it/dual/IoTDBSubscriptionTopicIT.java | 157 ++++- .../it/local/IoTDBSubscriptionBasicIT.java | 12 +- .../it/local/IoTDBSubscriptionIdempotentIT.java | 22 +- .../it/local/IoTDBSubscriptionRestartIT.java | 298 +++++++++- .../org/apache/iotdb/isession/SessionDataSet.java | 6 +- .../apache/iotdb/rpc/TElasticFramedTransport.java | 3 +- .../rpc/subscription/config/ConsumerConstant.java | 17 +- .../exception}/SubscriptionException.java | 2 +- .../SubscriptionParameterNotValidException.java | 17 +- .../SubscriptionPollTimeOutException.java | 16 +- .../subscription/ConsumerHeartbeatWorker.java | 26 +- .../subscription/PullConsumerAutoCommitWorker.java | 10 +- .../session/subscription/SubscriptionConsumer.java | 383 ++++++++++-- .../subscription/SubscriptionEndpointsSyncer.java | 131 +++++ .../session/subscription/SubscriptionProvider.java | 48 +- .../subscription/SubscriptionPullConsumer.java | 99 +++- .../subscription/SubscriptionPushConsumer.java | 2 +- .../session/subscription/SubscriptionSession.java | 20 +- .../SubscriptionSessionConnection.java | 16 - .../session/subscription/model/Subscription.java | 6 +- .../iotdb/session/subscription/model/Topic.java | 4 +- iotdb-connector/grafana-plugin/yarn.lock | 16 +- .../client/async/AsyncDataNodeClientPool.java | 7 + .../client/async/handlers/AsyncClientHandler.java | 1 + .../statemachine/ConfigRegionStateMachine.java | 3 + .../iotdb/confignode/manager/ConfigManager.java | 11 + .../apache/iotdb/confignode/manager/IManager.java | 7 + .../iotdb/confignode/manager/ProcedureManager.java | 9 +- .../receiver/IoTDBConfigNodeReceiverAgent.java | 21 + .../pipe/execution/PipeConfigNodeSubtask.java | 9 +- .../receiver/protocol/IoTDBConfigNodeReceiver.java | 36 +- .../persistence/subscription/SubscriptionInfo.java | 2 +- .../iotdb/confignode/procedure/Procedure.java | 15 - .../procedure/env/RegionMaintainHandler.java | 85 +-- .../procedure/impl/StateMachineProcedure.java | 18 + .../impl/region/AddRegionPeerProcedure.java | 91 ++- .../impl/region/RemoveRegionPeerProcedure.java | 23 +- .../AbstractOperateSubscriptionProcedure.java | 2 +- .../consumer/AlterConsumerGroupProcedure.java | 2 +- .../runtime/ConsumerGroupMetaSyncProcedure.java | 2 +- .../subscription/CreateSubscriptionProcedure.java | 2 +- .../subscription/DropSubscriptionProcedure.java | 2 +- .../subscription/topic/AlterTopicProcedure.java | 2 +- .../subscription/topic/CreateTopicProcedure.java | 2 +- .../subscription/topic/DropTopicProcedure.java | 2 +- .../topic/runtime/TopicMetaSyncProcedure.java | 2 +- .../testonly/CreateManyDatabasesProcedure.java | 10 +- .../thrift/ConfigNodeRPCServiceProcessor.java | 8 +- iotdb-core/consensus/pom.xml | 4 + .../org/apache/iotdb/consensus/IConsensus.java | 19 + .../consensus/exception/ConsensusException.java | 4 + .../apache/iotdb/consensus/iot/IoTConsensus.java | 91 ++- .../consensus/iot/IoTConsensusServerImpl.java | 130 +++-- .../service/IoTConsensusRPCServiceProcessor.java | 8 +- .../iotdb/consensus/ratis/RatisConsensus.java | 42 +- .../iotdb/consensus/simple/SimpleConsensus.java | 11 + .../apache/iotdb/consensus/iot/ReplicateTest.java | 21 + .../apache/iotdb/consensus/iot/StabilityTest.java | 12 +- .../iotdb/consensus/ratis/RatisConsensusTest.java | 26 +- .../dataregion/DataRegionStateMachine.java | 15 +- .../db/exception/PartitionViolationException.java | 6 +- ...IoTDBThriftSyncPipeTransferBatchReqBuilder.java | 2 +- .../builder/PipeTransferBatchReqBuilder.java | 49 +- .../async/IoTDBDataRegionAsyncConnector.java | 26 + .../PipeTransferTabletBatchEventHandler.java | 12 +- .../PipeTransferTabletInsertionEventHandler.java | 5 - .../PipeTransferTsFileInsertionEventHandler.java | 2 - .../common/tablet/PipeRawTabletInsertionEvent.java | 4 +- .../iotdb/db/pipe/metric/PipeConnectorMetrics.java | 46 +- .../iotdb/db/pipe/metric/PipeExtractorMetrics.java | 49 +- .../iotdb/db/pipe/metric/PipeProcessorMetrics.java | 46 +- .../processor/aggregate/AggregateProcessor.java | 60 +- .../protocol/thrift/IoTDBDataNodeReceiver.java | 60 +- .../thrift/IoTDBDataNodeReceiverAgent.java | 18 + .../db/pipe/resource/memory/PipeMemoryBlock.java | 7 +- .../resource/tsfile/PipeTsFileResourceManager.java | 1 - .../pipe/task/connection/PipeEventCollector.java | 8 +- .../db/pipe/task/stage/PipeTaskConnectorStage.java | 3 +- .../db/pipe/task/stage/PipeTaskProcessorStage.java | 2 +- .../subtask/connector/PipeConnectorSubtask.java | 6 +- .../connector/PipeConnectorSubtaskManager.java | 10 +- .../subtask/processor/PipeProcessorSubtask.java | 5 +- .../iotdb/db/protocol/client/ConfigNodeClient.java | 7 + .../impl/DataNodeInternalRPCServiceImpl.java | 17 +- .../db/queryengine/common/MPPQueryContext.java | 17 +- .../execution/exchange/sink/ShuffleSinkHandle.java | 4 +- .../exchange/source/LocalSourceHandle.java | 9 - .../queryengine/execution/memory/MemoryPool.java | 14 +- .../iotdb/db/queryengine/plan/Coordinator.java | 13 +- .../queryengine/plan/analyze/AnalyzeVisitor.java | 35 +- .../analyze/cache/schema/DataNodeSchemaCache.java | 47 +- .../analyze/cache/schema/SchemaCacheEntry.java | 8 + .../cache/schema/TimeSeriesSchemaCache.java | 32 + .../cache/schema/dualkeycache/IDualKeyCache.java | 9 + .../schema/dualkeycache/impl/DualKeyCacheImpl.java | 59 ++ .../schema/lastcache/DataNodeLastCacheManager.java | 7 + .../schema/lastcache/ILastCacheContainer.java | 5 +- .../cache/schema/lastcache/LastCacheContainer.java | 7 + .../analyze/lock/DataNodeSchemaLockManager.java | 59 ++ .../plan/analyze/lock/SchemaLockType.java | 51 ++ .../plan/analyze/schema/ClusterSchemaFetcher.java | 14 +- .../config/executor/ClusterConfigTaskExecutor.java | 20 +- .../config/executor/IConfigTaskExecutor.java | 4 +- .../db/queryengine/plan/parser/ASTVisitor.java | 2 +- .../java/org/apache/iotdb/db/service/DataNode.java | 69 +-- .../iotdb/db/service/RegionMigrateService.java | 7 +- .../db/storageengine/dataregion/DataRegion.java | 44 +- .../exception/CompactionExceptionHandler.java | 15 +- .../execute/recover/CompactionRecoverTask.java | 18 +- .../execute/task/AbstractCompactionTask.java | 10 +- .../execute/task/InnerSpaceCompactionTask.java | 2 +- .../compaction/schedule/CompactionScheduler.java | 6 - .../compaction/schedule/CompactionTaskQueue.java | 52 +- .../impl/RewriteCrossSpaceCompactionSelector.java | 2 +- .../memtable/AlignedWritableMemChunk.java | 5 +- .../dataregion/snapshot/SnapshotLoader.java | 6 - .../storageengine/dataregion/tsfile/TsFileID.java | 18 +- .../tsfile/timeindex/DeviceTimeIndex.java | 64 +- .../dataregion/tsfile/timeindex/FileTimeIndex.java | 22 +- .../dataregion/tsfile/timeindex/ITimeIndex.java | 2 - .../agent/SubscriptionAgentLauncher.java | 2 +- .../agent/SubscriptionConsumerAgent.java | 2 +- .../receiver/SubscriptionReceiverV1.java | 7 +- .../task/stage/SubscriptionTaskConnectorStage.java | 2 +- .../SubscriptionConnectorSubtaskManager.java | 12 +- .../cache/dualkeycache/DualKeyCacheTest.java | 114 ++++ .../inner/InnerSpaceCompactionExceptionTest.java | 5 +- .../src/assembly/resources/sbin/health_check.bat | 641 +++++++++++++++++++++ .../src/assembly/resources/sbin/health_check.sh | 388 +++++++++++++ .../iotdb/commons/pipe/event/EnrichedEvent.java | 24 +- .../pipe/progress/PipeEventCommitManager.java | 26 +- .../commons/pipe/progress/PipeEventCommitter.java | 15 +- .../commons/pipe/receiver/IoTDBFileReceiver.java | 263 ++++++--- .../commons/pipe/receiver/IoTDBReceiverAgent.java | 46 +- .../task/subtask/PipeAbstractConnectorSubtask.java | 2 +- .../pipe/task/subtask/PipeReportableSubtask.java | 2 +- .../commons/pipe/task/subtask/PipeSubtask.java | 18 +- .../meta/consumer/ConsumerGroupMeta.java | 2 +- ... => IoTConsensusDeleteLocalPeerKillPoints.java} | 8 +- ...a => IoTConsensusInactivatePeerKillPoints.java} | 8 +- ...TConsensusRemovePeerCoordinatorKillPoints.java} | 2 +- ...nsusRemovePeerKillPoints.java => KillNode.java} | 11 +- .../consumer/ConsumerGroupDeSerTest.java | 2 +- .../src/main/thrift/confignode.thrift | 8 +- .../src/main/thrift/iotconsensus.thrift | 1 + pom.xml | 2 +- 177 files changed, 4744 insertions(+), 1177 deletions(-) diff --cc iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/SubscriptionSession.java index 15611a1c5ce,c286c801489..22b59df9651 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/SubscriptionSession.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/SubscriptionSession.java @@@ -61,12 -65,12 +65,12 @@@ public class SubscriptionSession extend @Override public SessionConnection constructSessionConnection( Session session, TEndPoint endpoint, ZoneId zoneId) throws IoTDBConnectionException { - if (endpoint == null) { - return new SubscriptionSessionConnection( - session, zoneId, availableNodes, maxRetryCount, retryIntervalInMs, sqlDialect); + if (Objects.isNull(endpoint)) { + throw new SubscriptionParameterNotValidException( + "Subscription session must be configured with an endpoint."); } return new SubscriptionSessionConnection( - session, endpoint, zoneId, availableNodes, maxRetryCount, retryIntervalInMs); + session, endpoint, zoneId, availableNodes, maxRetryCount, retryIntervalInMs, sqlDialect); } /////////////////////////////// topic /////////////////////////////// diff --cc iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/SubscriptionSessionConnection.java index 6021caae8b5,891cd0fc4ff..4f60c4ef582 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/SubscriptionSessionConnection.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/SubscriptionSessionConnection.java @@@ -65,23 -64,11 +64,12 @@@ public class SubscriptionSessionConnect ZoneId zoneId, Supplier<List<TEndPoint>> availableNodes, int maxRetryCount, - long retryIntervalInMs) + long retryIntervalInMs, + String sqlDialect) throws IoTDBConnectionException { - super(session, endPoint, zoneId, availableNodes, maxRetryCount, retryIntervalInMs); + super(session, endPoint, zoneId, availableNodes, maxRetryCount, retryIntervalInMs, sqlDialect); } - public SubscriptionSessionConnection( - Session session, - ZoneId zoneId, - Supplier<List<TEndPoint>> availableNodes, - int maxRetryCount, - long retryIntervalInMs, - String sqlDialect) - throws IoTDBConnectionException { - super(session, zoneId, availableNodes, maxRetryCount, retryIntervalInMs, sqlDialect); - } - // from org.apache.iotdb.session.NodesSupplier.updateDataNodeList public Map<Integer, TEndPoint> fetchAllEndPoints() throws IoTDBConnectionException, StatementExecutionException { diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java index aa2da4fce12,5bd53a23403..3286a256695 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java @@@ -2681,111 -2674,15 +2685,123 @@@ public class ClusterConfigTaskExecutor } @Override + public SettableFuture<ConfigTaskResult> showDatabases(ShowDB showDB) { + SettableFuture<ConfigTaskResult> future = SettableFuture.create(); + // Construct request using statement + List<String> databasePathPattern = Arrays.asList(ALL_RESULT_NODES); + try (ConfigNodeClient client = + CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) { + // Send request to some API server + TGetDatabaseReq req = new TGetDatabaseReq(databasePathPattern, ALL_MATCH_SCOPE.serialize()); + TShowDatabaseResp resp = client.showDatabase(req); + // build TSBlock + ShowDBTask.buildTSBlock(resp.getDatabaseInfoMap(), future); + } catch (IOException | ClientManagerException | TException e) { + future.setException(e); + } + return future; + } + + @Override + public SettableFuture<ConfigTaskResult> useDatabase(Use useDB, IClientSession clientSession) { + SettableFuture<ConfigTaskResult> future = SettableFuture.create(); + // Construct request using statement + List<String> databasePathPattern = Arrays.asList(ROOT, useDB.getDatabase().getValue()); + try (ConfigNodeClient client = + CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) { + // Send request to some API server + TGetDatabaseReq req = new TGetDatabaseReq(databasePathPattern, ALL_MATCH_SCOPE.serialize()); + TShowDatabaseResp resp = client.showDatabase(req); + if (!resp.getDatabaseInfoMap().isEmpty()) { + clientSession.setDatabaseName(useDB.getDatabase().getValue()); + future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS)); + } else { + future.setException( + new IoTDBException( + String.format("Database %s doesn't exists.", useDB.getDatabase().getValue()), + TSStatusCode.DATABASE_NOT_EXIST.getStatusCode())); + } + } catch (IOException | ClientManagerException | TException e) { + future.setException(e); + } + return future; + } + + private String transformDBName(String dbName) { + return ROOT + PATH_SEPARATOR_CHAR + dbName; + } + + @Override + public SettableFuture<ConfigTaskResult> dropDatabase(DropDB dropDB) { + SettableFuture<ConfigTaskResult> future = SettableFuture.create(); + TDeleteDatabasesReq req = + new TDeleteDatabasesReq( + Collections.singletonList(transformDBName(dropDB.getDbName().getValue()))); + try (ConfigNodeClient client = + CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) { + TSStatus tsStatus = client.deleteDatabases(req); + if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) { + LOGGER.warn( + "Failed to execute delete database {} in config node, status is {}.", + dropDB.getDbName().getValue(), + tsStatus); + future.setException(new IoTDBException(tsStatus.message, tsStatus.getCode())); + } else { + future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS)); + } + } catch (ClientManagerException | TException e) { + future.setException(e); + } + return future; + } + + @Override + public SettableFuture<ConfigTaskResult> createDatabase(CreateDB createDB) { + SettableFuture<ConfigTaskResult> future = SettableFuture.create(); + // Construct request using statement + TDatabaseSchema databaseSchema = new TDatabaseSchema(); + databaseSchema.setName(ROOT + PATH_SEPARATOR_CHAR + createDB.getDbName()); + try (ConfigNodeClient configNodeClient = + CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) { + // Send request to some API server + TSStatus tsStatus = configNodeClient.setDatabase(databaseSchema); + // Get response or throw exception + if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) { + // If database already exists when loading, we do not throw exceptions to avoid printing too + // many logs + if (TSStatusCode.DATABASE_ALREADY_EXISTS.getStatusCode() == tsStatus.getCode() + && createDB.isSetIfNotExists()) { + future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS)); + } else { + LOGGER.warn( + "Failed to execute create database {} in config node, status is {}.", + createDB.getDbName(), + tsStatus); + future.setException(new IoTDBException(tsStatus.message, tsStatus.code)); + } + } else { + future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS)); + } + } catch (ClientManagerException | TException e) { + future.setException(e); + } + return future; + } + + @Override + public SettableFuture<ConfigTaskResult> createTable(CreateTable createTable) { + return null; + } ++ + public void handlePipeConfigClientExit(String clientId) { + try (final ConfigNodeClient configNodeClient = + CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) { + final TSStatus status = configNodeClient.handlePipeConfigClientExit(clientId); + if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != status.getCode()) { + LOGGER.warn("Failed to handlePipeConfigClientExit, status is {}.", status); + } + } catch (Exception e) { + LOGGER.warn("Failed to handlePipeConfigClientExit.", e); + } + } } diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java index a13ddaef735,df24fec47da..91c5ea88884 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java @@@ -242,17 -236,7 +242,19 @@@ public interface IConfigTaskExecutor TThrottleQuotaResp getThrottleQuota(); - TPipeTransferResp handleTransferConfigPlan(TPipeTransferReq req); + TPipeTransferResp handleTransferConfigPlan(String clientId, TPipeTransferReq req); + + void handlePipeConfigClientExit(String clientId); + + // =============================== table syntax ========================================= + + SettableFuture<ConfigTaskResult> showDatabases(ShowDB showDB); + + SettableFuture<ConfigTaskResult> useDatabase(Use useDB, IClientSession clientSession); + + SettableFuture<ConfigTaskResult> dropDatabase(DropDB dropDB); + + SettableFuture<ConfigTaskResult> createDatabase(CreateDB createDB); + + SettableFuture<ConfigTaskResult> createTable(CreateTable createTable); }
