This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch rel/0.13
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/0.13 by this push:
new 28b1830f74 make operation sync cover all protocol (#6321)
28b1830f74 is described below
commit 28b1830f7403ba7f818b10206f56eed3f7d2d259
Author: Jialin Qiao <[email protected]>
AuthorDate: Fri Jun 17 14:24:23 2022 +0800
make operation sync cover all protocol (#6321)
---
.../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")
