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 a1ec9f323ea40a253b12b5c4655d32c331b9c39b Merge: 91417c6 9931149 Author: xiangdong huang <[email protected]> AuthorDate: Sat Sep 4 13:03:04 2021 +0800 merge with master README.md | 86 ++++++++++------ README_ZH.md | 86 ++++++++++------ .../iotdb/cluster/server/ClusterTSServiceImpl.java | 10 +- .../iotdb/cluster/server/member/RaftMember.java | 17 ++++ .../apache/iotdb/cluster/utils/PartitionUtils.java | 4 +- docs/Download/README.md | 8 +- docs/SystemDesign/DataQuery/RawDataQuery.md | 26 +++-- docs/SystemDesign/DataQuery/SeriesReader.md | 36 ++++--- docs/UserGuide/API/Time-zone.md | 6 ++ docs/UserGuide/Appendix/Config-Manual.md | 81 +++++++++++++++ .../UserGuide/Data-Concept/Auto-Create-MetaData.md | 107 ++++++++++++++++++++ docs/UserGuide/Data-Concept/Time-Partition.md | 64 ++++++++++++ .../DML-Data-Manipulation-Language.md | 74 ++++++++++++++ docs/zh/Download/README.md | 6 +- docs/zh/SystemDesign/DataQuery/OrderByTimeQuery.md | 2 +- docs/zh/SystemDesign/DataQuery/RawDataQuery.md | 26 +++-- docs/zh/SystemDesign/DataQuery/SeriesReader.md | 36 ++++--- docs/zh/UserGuide/API/Time-zone.md | 8 +- docs/zh/UserGuide/Appendix/Config-Manual.md | 81 +++++++++++++++ .../UserGuide/Data-Concept/Auto-Create-MetaData.md | 107 ++++++++++++++++++++ docs/zh/UserGuide/Data-Concept/Time-Partition.md | 64 ++++++++++++ .../DML-Data-Manipulation-Language.md | 77 +++++++++++++++ .../resources/conf/iotdb-engine.properties | 24 ++--- .../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 12 +-- .../apache/iotdb/db/qp/physical/PhysicalPlan.java | 7 +- .../db/qp/physical/sys/SetSystemModePlan.java | 33 +++++++ .../apache/iotdb/db/tools/TsFileRewriteTool.java | 11 ++- .../db/tools/upgrade/TsFileOnlineUpgradeTool.java | 2 +- .../db/writelog/node/ExclusiveWriteLogNode.java | 3 + .../iotdb/db/utils/TsFileRewriteToolTest.java | 109 +++++++++++++++++++++ .../apache/iotdb/db/writelog/WriteLogNodeTest.java | 26 +++++ site/pom.xml | 10 ++ site/src/main/.vuepress/config.js | 38 ++++--- .../.vuepress/theme/global-components/IoTDB.vue | 2 +- .../.vuepress/theme/global-components/IoTDBZH.vue | 2 +- .../test/java/org/apache/iotdb/db/sql/Cases.java | 21 ++++ .../iotdb/tsfile/read/TsFileSequenceReader.java | 4 + .../tsfile/v2/read/TsFileSequenceReaderForV2.java | 4 - 38 files changed, 1156 insertions(+), 164 deletions(-) diff --cc cluster/src/main/java/org/apache/iotdb/cluster/server/ClusterTSServiceImpl.java index bd26602,0000000..c2a75fc mode 100644,000000..100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/server/ClusterTSServiceImpl.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/server/ClusterTSServiceImpl.java @@@ -1,162 -1,0 +1,170 @@@ +/* + * 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.server; + +import org.apache.iotdb.cluster.client.async.AsyncDataClient; +import org.apache.iotdb.cluster.client.sync.SyncDataClient; +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.query.ClusterPlanExecutor; +import org.apache.iotdb.cluster.query.RemoteQueryContext; +import org.apache.iotdb.cluster.rpc.thrift.Node; +import org.apache.iotdb.cluster.rpc.thrift.RaftNode; +import org.apache.iotdb.cluster.server.handlers.caller.GenericHandler; +import org.apache.iotdb.cluster.server.member.MetaGroupMember; ++import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.exception.StorageEngineException; +import org.apache.iotdb.db.exception.query.QueryProcessException; +import org.apache.iotdb.db.qp.physical.PhysicalPlan; ++import org.apache.iotdb.db.qp.physical.sys.FlushPlan; ++import org.apache.iotdb.db.qp.physical.sys.SetSystemModePlan; +import org.apache.iotdb.db.query.context.QueryContext; +import org.apache.iotdb.db.service.TSServiceImpl; +import org.apache.iotdb.rpc.RpcUtils; +import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.service.rpc.thrift.TSStatus; - +import org.apache.thrift.TException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicReference; + +/** + * ClusterTSServiceImpl is the cluster version of TSServiceImpl, which is responsible for the + * processing of the user requests (sqls and session api). It inherits the basic procedures from + * TSServiceImpl, but redirect the queries of data and metadata to a MetaGroupMember of the local + * node. + */ +public class ClusterTSServiceImpl extends TSServiceImpl { + + private static final Logger logger = LoggerFactory.getLogger(ClusterTSServiceImpl.class); + /** + * The Coordinator of the local node. Through this node queries data and meta from the cluster and + * performs data manipulations to the cluster. + */ + private Coordinator coordinator; + + /** + * queryId -> queryContext map. When a query ends either normally or accidentally, the resources + * used by the query can be found in the context and then released. + */ + private Map<Long, RemoteQueryContext> queryContextMap = new ConcurrentHashMap<>(); + + public ClusterTSServiceImpl() throws QueryProcessException {} + + public void setExecutor(MetaGroupMember metaGroupMember) throws QueryProcessException { + executor = new ClusterPlanExecutor(metaGroupMember); + } + + public void setCoordinator(Coordinator coordinator) { + this.coordinator = coordinator; + } + + /** + * Redirect the plan to the local Coordinator so that it will be processed cluster-wide. + * + * @param plan + * @return + */ + @Override + protected TSStatus executeNonQueryPlan(PhysicalPlan plan) { + try { + plan.checkIntegrity(); ++ if (!(plan instanceof SetSystemModePlan) ++ && !(plan instanceof FlushPlan) ++ && IoTDBDescriptor.getInstance().getConfig().isReadOnly()) { ++ throw new QueryProcessException( ++ "Current system mode is read-only, does not support non-query operation"); ++ } + } catch (QueryProcessException e) { + logger.warn("Illegal plan detected： {}", plan); + return RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, e.getMessage()); + } + + return coordinator.executeNonQueryPlan(plan); + } + + /** + * Generate and cache a QueryContext using "queryId". In the distributed version, the QueryContext + * is a RemoteQueryContext. + * + * @param queryId + * @return a RemoteQueryContext using queryId + */ + @Override + protected QueryContext genQueryContext(long queryId, boolean debug) { + RemoteQueryContext context = new RemoteQueryContext(queryId, debug); + queryContextMap.put(queryId, context); + return context; + } + + /** + * Release the local and remote resources used by a query. + * + * @param queryId + * @throws StorageEngineException + */ + @Override + protected void releaseQueryResource(long queryId) throws StorageEngineException { + // release resources locally + super.releaseQueryResource(queryId); + // release resources remotely + RemoteQueryContext context = queryContextMap.remove(queryId); + if (context != null) { + // release the resources in every queried node + for (Entry<RaftNode, Set<Node>> headerEntry : context.getQueriedNodesMap().entrySet()) { + RaftNode header = headerEntry.getKey(); + Set<Node> queriedNodes = headerEntry.getValue(); + + for (Node queriedNode : queriedNodes) { + GenericHandler<Void> handler = new GenericHandler<>(queriedNode, new AtomicReference<>()); + try { + if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) { + AsyncDataClient client = + coordinator.getAsyncDataClient( + queriedNode, ClusterConstant.getReadOperationTimeoutMS()); + client.endQuery(header, coordinator.getThisNode(), queryId, handler); + } else { + try (SyncDataClient syncDataClient = + coordinator.getSyncDataClient( + queriedNode, ClusterConstant.getReadOperationTimeoutMS())) { + try { + syncDataClient.endQuery(header, coordinator.getThisNode(), queryId); + } catch (TException e) { + // the connection may be broken, close it to avoid it being reused + syncDataClient.getInputProtocol().getTransport().close(); + throw e; + } + } + } + } catch (IOException | TException e) { + logger.error("Cannot end query {} in {}", queryId, queriedNode); + } + } + } + } + } +}
