This is an automated email from the ASF dual-hosted git repository. qiaojialin pushed a commit to branch transmit_mqtt_plan in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0b285a26c18fd238aabfb9299cca7cafb41bcf1c Author: qiaojialin <[email protected]> AuthorDate: Fri Jun 17 12:42:27 2022 +0800 make operation sync cover all protocol --- .../iotdb/db/service/basic/ServiceProvider.java | 2 +- .../service/basic/StandaloneServiceProvider.java | 115 +++++++++++++++++++++ .../db/service/thrift/impl/TSServiceImpl.java | 110 -------------------- 3 files changed, 116 insertions(+), 111 deletions(-) diff --git a/server/src/main/java/org/apache/iotdb/db/service/basic/ServiceProvider.java b/server/src/main/java/org/apache/iotdb/db/service/basic/ServiceProvider.java index 78a969959d..d71e54e128 100644 --- a/server/src/main/java/org/apache/iotdb/db/service/basic/ServiceProvider.java +++ b/server/src/main/java/org/apache/iotdb/db/service/basic/ServiceProvider.java @@ -59,7 +59,7 @@ import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onNPEOrUnexpectedExce public abstract class ServiceProvider { - private static final Logger LOGGER = LoggerFactory.getLogger(ServiceProvider.class); + protected static final Logger LOGGER = LoggerFactory.getLogger(ServiceProvider.class); public static final Logger AUDIT_LOGGER = LoggerFactory.getLogger(IoTDBConstant.AUDIT_LOGGER_NAME); public static final Logger SLOW_SQL_LOGGER = diff --git a/server/src/main/java/org/apache/iotdb/db/service/basic/StandaloneServiceProvider.java b/server/src/main/java/org/apache/iotdb/db/service/basic/StandaloneServiceProvider.java index d9cc7cfdc5..bddb0cac9d 100644 --- a/server/src/main/java/org/apache/iotdb/db/service/basic/StandaloneServiceProvider.java +++ b/server/src/main/java/org/apache/iotdb/db/service/basic/StandaloneServiceProvider.java @@ -18,7 +18,15 @@ */ package org.apache.iotdb.db.service.basic; +import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.doublelive.OperationSyncConsumer; +import org.apache.iotdb.db.doublelive.OperationSyncDDLProtector; +import org.apache.iotdb.db.doublelive.OperationSyncDMLProtector; +import org.apache.iotdb.db.doublelive.OperationSyncLogService; +import org.apache.iotdb.db.doublelive.OperationSyncPlanTypeUtils; +import org.apache.iotdb.db.doublelive.OperationSyncProducer; +import org.apache.iotdb.db.doublelive.OperationSyncWriteTask; import org.apache.iotdb.db.exception.StorageEngineException; import org.apache.iotdb.db.exception.StorageEngineReadonlyException; import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException; @@ -28,11 +36,73 @@ 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.session.pool.SessionPool; +import org.apache.iotdb.tsfile.utils.Pair; + +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; public class StandaloneServiceProvider extends ServiceProvider { + /* OperationSync module */ + private static final boolean isEnableOperationSync = + IoTDBDescriptor.getInstance().getConfig().isEnableOperationSync(); + private final SessionPool operationSyncsessionPool; + private final OperationSyncProducer operationSyncProducer; + private final OperationSyncDDLProtector operationSyncDDLProtector; + private final OperationSyncLogService operationSyncDDLLogService; + public StandaloneServiceProvider() throws QueryProcessException { super(new PlanExecutor()); + if (isEnableOperationSync) { + /* Open OperationSync */ + IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + // create SessionPool for OperationSync + operationSyncsessionPool = + new SessionPool( + config.getSecondaryAddress(), + config.getSecondaryPort(), + config.getSecondaryUser(), + config.getSecondaryPassword(), + 5); + + // create operationSyncDDLProtector and operationSyncDDLLogService + operationSyncDDLProtector = new OperationSyncDDLProtector(operationSyncsessionPool); + new Thread(operationSyncDDLProtector).start(); + operationSyncDDLLogService = + new OperationSyncLogService("OperationSyncDDLLog", operationSyncDDLProtector); + new Thread(operationSyncDDLLogService).start(); + + // create OperationSyncProducer + BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>> + blockingQueue = new ArrayBlockingQueue<>(config.getOperationSyncProducerCacheSize()); + operationSyncProducer = new OperationSyncProducer(blockingQueue); + + // create OperationSyncDMLProtector and OperationSyncDMLLogService + OperationSyncDMLProtector operationSyncDMLProtector = + new OperationSyncDMLProtector(operationSyncDDLProtector, operationSyncProducer); + new Thread(operationSyncDMLProtector).start(); + OperationSyncLogService operationSyncDMLLogService = + new OperationSyncLogService("OperationSyncDMLLog", operationSyncDMLProtector); + new Thread(operationSyncDMLLogService).start(); + + // create OperationSyncConsumer + for (int i = 0; i < config.getOperationSyncConsumerConcurrencySize(); i++) { + OperationSyncConsumer consumer = + new OperationSyncConsumer( + blockingQueue, operationSyncsessionPool, operationSyncDMLLogService); + new Thread(consumer).start(); + } + } else { + operationSyncsessionPool = null; + operationSyncProducer = null; + operationSyncDDLProtector = null; + operationSyncDDLLogService = null; + } } @Override @@ -50,6 +120,51 @@ public class StandaloneServiceProvider extends ServiceProvider { && IoTDBDescriptor.getInstance().getConfig().isReadOnly()) { throw new StorageEngineReadonlyException(); } + + if (isEnableOperationSync) { + // OperationSync should transmit before execute + transmitOperationSync(plan); + } + return executor.processNonQuery(plan); } + + private void transmitOperationSync(PhysicalPlan physicalPlan) { + + OperationSyncPlanTypeUtils.OperationSyncPlanType planType = + OperationSyncPlanTypeUtils.getOperationSyncPlanType(physicalPlan); + if (planType == null) { + // Don't need OperationSync + return; + } + + // serialize physical plan + ByteBuffer buffer; + try { + int size = physicalPlan.getSerializedSize(); + ByteArrayOutputStream operationSyncByteStream = new ByteArrayOutputStream(size); + DataOutputStream operationSyncSerializeStream = new DataOutputStream(operationSyncByteStream); + physicalPlan.serialize(operationSyncSerializeStream); + buffer = ByteBuffer.wrap(operationSyncByteStream.toByteArray()); + } catch (IOException e) { + LOGGER.error("OperationSync can't serialize PhysicalPlan", e); + return; + } + + switch (planType) { + case DDLPlan: + // Create OperationSyncWriteTask and wait + OperationSyncWriteTask ddlTask = + new OperationSyncWriteTask( + buffer, + operationSyncsessionPool, + operationSyncDDLProtector, + operationSyncDDLLogService); + ddlTask.run(); + break; + case DMLPlan: + // Put into OperationSyncProducer + operationSyncProducer.put(new Pair<>(buffer, planType)); + } + } } diff --git a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/TSServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/TSServiceImpl.java index 949d1c1853..bc40c5010c 100644 --- a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/TSServiceImpl.java +++ b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/TSServiceImpl.java @@ -25,13 +25,7 @@ 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.OperationType; -import org.apache.iotdb.db.doublelive.OperationSyncConsumer; -import org.apache.iotdb.db.doublelive.OperationSyncDDLProtector; -import org.apache.iotdb.db.doublelive.OperationSyncDMLProtector; -import org.apache.iotdb.db.doublelive.OperationSyncLogService; import org.apache.iotdb.db.doublelive.OperationSyncPlanTypeUtils; -import org.apache.iotdb.db.doublelive.OperationSyncProducer; -import org.apache.iotdb.db.doublelive.OperationSyncWriteTask; import org.apache.iotdb.db.engine.selectinto.InsertTabletPlansIterator; import org.apache.iotdb.db.exception.IoTDBException; import org.apache.iotdb.db.exception.QueryInBatchStatementException; @@ -131,7 +125,6 @@ import org.apache.iotdb.service.rpc.thrift.TSSetUsingTemplateReq; import org.apache.iotdb.service.rpc.thrift.TSStatus; import org.apache.iotdb.service.rpc.thrift.TSTracingInfo; import org.apache.iotdb.service.rpc.thrift.TSUnsetSchemaTemplateReq; -import org.apache.iotdb.session.pool.SessionPool; import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException; import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException; import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; @@ -139,14 +132,11 @@ 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.query.dataset.QueryDataSet; -import org.apache.iotdb.tsfile.utils.Pair; import org.apache.thrift.TException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.ByteArrayOutputStream; -import java.io.DataOutputStream; import java.io.IOException; import java.nio.ByteBuffer; import java.sql.SQLException; @@ -157,8 +147,6 @@ import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Map.Entry; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.BlockingQueue; import java.util.concurrent.Callable; import java.util.concurrent.Future; import java.util.stream.Collectors; @@ -319,63 +307,9 @@ public class TSServiceImpl implements TSIService.Iface { protected final ServiceProvider serviceProvider; - /* OperationSync module */ - private static final boolean isEnableOperationSync = - IoTDBDescriptor.getInstance().getConfig().isEnableOperationSync(); - private final SessionPool operationSyncsessionPool; - private final OperationSyncProducer operationSyncProducer; - private final OperationSyncDDLProtector operationSyncDDLProtector; - private final OperationSyncLogService operationSyncDDLLogService; - public TSServiceImpl() { super(); serviceProvider = IoTDB.serviceProvider; - - if (isEnableOperationSync) { - /* Open OperationSync */ - IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); - // create SessionPool for OperationSync - operationSyncsessionPool = - new SessionPool( - config.getSecondaryAddress(), - config.getSecondaryPort(), - config.getSecondaryUser(), - config.getSecondaryPassword(), - 5); - - // create operationSyncDDLProtector and operationSyncDDLLogService - operationSyncDDLProtector = new OperationSyncDDLProtector(operationSyncsessionPool); - new Thread(operationSyncDDLProtector).start(); - operationSyncDDLLogService = - new OperationSyncLogService("OperationSyncDDLLog", operationSyncDDLProtector); - new Thread(operationSyncDDLLogService).start(); - - // create OperationSyncProducer - BlockingQueue<Pair<ByteBuffer, OperationSyncPlanTypeUtils.OperationSyncPlanType>> - blockingQueue = new ArrayBlockingQueue<>(config.getOperationSyncProducerCacheSize()); - operationSyncProducer = new OperationSyncProducer(blockingQueue); - - // create OperationSyncDMLProtector and OperationSyncDMLLogService - OperationSyncDMLProtector operationSyncDMLProtector = - new OperationSyncDMLProtector(operationSyncDDLProtector, operationSyncProducer); - new Thread(operationSyncDMLProtector).start(); - OperationSyncLogService operationSyncDMLLogService = - new OperationSyncLogService("OperationSyncDMLLog", operationSyncDMLProtector); - new Thread(operationSyncDMLLogService).start(); - - // create OperationSyncConsumer - for (int i = 0; i < config.getOperationSyncConsumerConcurrencySize(); i++) { - OperationSyncConsumer consumer = - new OperationSyncConsumer( - blockingQueue, operationSyncsessionPool, operationSyncDMLLogService); - new Thread(consumer).start(); - } - } else { - operationSyncsessionPool = null; - operationSyncProducer = null; - operationSyncDDLProtector = null; - operationSyncDDLLogService = null; - } } @Override @@ -2195,51 +2129,7 @@ public class TSServiceImpl implements TSIService.Iface { } } - private void transmitOperationSync(PhysicalPlan physicalPlan) { - - OperationSyncPlanTypeUtils.OperationSyncPlanType planType = - OperationSyncPlanTypeUtils.getOperationSyncPlanType(physicalPlan); - if (planType == null) { - // Don't need OperationSync - return; - } - - // serialize physical plan - ByteBuffer buffer; - try { - int size = physicalPlan.getSerializedSize(); - ByteArrayOutputStream operationSyncByteStream = new ByteArrayOutputStream(size); - DataOutputStream operationSyncSerializeStream = new DataOutputStream(operationSyncByteStream); - physicalPlan.serialize(operationSyncSerializeStream); - buffer = ByteBuffer.wrap(operationSyncByteStream.toByteArray()); - } catch (IOException e) { - LOGGER.error("OperationSync can't serialize PhysicalPlan", e); - return; - } - - switch (planType) { - case DDLPlan: - // Create OperationSyncWriteTask and wait - OperationSyncWriteTask ddlTask = - new OperationSyncWriteTask( - buffer, - operationSyncsessionPool, - operationSyncDDLProtector, - operationSyncDDLLogService); - ddlTask.run(); - break; - case DMLPlan: - // Put into OperationSyncProducer - operationSyncProducer.put(new Pair<>(buffer, planType)); - } - } - protected TSStatus executeNonQueryPlan(PhysicalPlan plan) { - if (isEnableOperationSync) { - // OperationSync should transmit before execute - transmitOperationSync(plan); - } - try { return serviceProvider.executeNonQuery(plan) ? RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, "Execute successfully")
