This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch load_v2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/load_v2 by this push:
new 8805ca8f0a8 Integrate pipe with TsFileSplitSender
8805ca8f0a8 is described below
commit 8805ca8f0a877b120c68c801bdebbbc48a516ca7
Author: Tian Jiang <[email protected]>
AuthorDate: Mon Oct 16 16:53:50 2023 +0800
Integrate pipe with TsFileSplitSender
---
.../org/apache/iotdb/pipe/api/PipeConnector.java | 12 ++
.../apache/iotdb/session/util/SessionUtils.java | 2 +-
.../thrift/ConfigNodeRPCServiceProcessor.java | 10 ++
.../org/apache/iotdb/db/audit/AuditLogger.java | 2 +-
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 6 +-
.../config/constant/PipeConnectorConstant.java | 11 ++
.../thrift/async/IoTDBThriftAsyncConnector.java | 147 +++++++++++++++++++++
.../tsfile/PipeBatchTsFileInsertionEvent.java | 49 +++++--
.../tsfile/TsFileListInsertionDataContainer.java | 43 +++---
.../historical/BatchedTsFileExtractor.java | 30 +++--
.../pipe/receiver/airgap/IoTDBAirGapReceiver.java | 2 +-
.../db/pipe/receiver/legacy/loader/ILoader.java | 2 +-
.../iotdb/db/protocol/client/ConfigNodeClient.java | 17 ++-
.../db/protocol/client/DataNodeInternalClient.java | 2 +-
.../iotdb/db/protocol/mqtt/MPPPublishHandler.java | 2 +-
.../rest/v1/impl/GrafanaApiServiceImpl.java | 2 +-
.../protocol/rest/v1/impl/RestApiServiceImpl.java | 2 +-
.../rest/v2/impl/GrafanaApiServiceImpl.java | 2 +-
.../protocol/rest/v2/impl/RestApiServiceImpl.java | 2 +-
.../protocol/thrift/impl/ClientRPCServiceImpl.java | 2 +-
.../impl/DataNodeInternalRPCServiceImpl.java | 2 +-
.../protocol/thrift/impl/MLNodeRPCServiceImpl.java | 2 +-
.../db/queryengine/plan/analyze/Analyzer.java | 1 +
.../BasicPartitionFetcher.java} | 60 +++------
.../analyze/partition/ClusterPartitionFetcher.java | 52 ++++++++
.../partition/ExternalPartitionFetcher.java | 52 ++++++++
.../analyze/schema/AutoCreateSchemaExecutor.java | 2 +-
.../analyze/schema/ClusterSchemaFetchExecutor.java | 2 +-
.../config/executor/ClusterConfigTaskExecutor.java | 2 +-
.../metrics/IoTDBInternalLocalReporter.java | 2 +-
.../src/main/thrift/confignode.thrift | 7 +
31 files changed, 417 insertions(+), 114 deletions(-)
diff --git
a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeConnector.java
b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeConnector.java
index 3ceb6f73f56..96ac6f436d3 100644
---
a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeConnector.java
+++
b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeConnector.java
@@ -24,6 +24,7 @@ import
org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileBatchInsertionEvent;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
import org.apache.iotdb.pipe.api.exception.PipeConnectionException;
@@ -138,6 +139,17 @@ public interface PipeConnector extends PipePlugin {
}
}
+ default void transfer(TsFileBatchInsertionEvent tsFileInsertionEvent) throws
Exception {
+ try {
+ for (final TabletInsertionEvent tabletInsertionEvent :
+ tsFileInsertionEvent.toTabletInsertionEvents()) {
+ transfer(tabletInsertionEvent);
+ }
+ } finally {
+ tsFileInsertionEvent.close();
+ }
+ }
+
/**
* This method is used to transfer the generic events, including
HeartbeatEvent.
*
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
index 07baee5ef93..4d8efb17fc0 100644
---
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
+++
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
@@ -260,7 +260,7 @@ public class SessionUtils {
return endPointsList;
}
- private static TEndPoint parseNodeUrl(String nodeUrl) {
+ public static TEndPoint parseNodeUrl(String nodeUrl) {
TEndPoint endPoint = new TEndPoint();
String[] split = nodeUrl.split(":");
if (split.length != 2) {
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index 06c0b75bb8d..8103cffd2a4 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -36,6 +36,7 @@ import org.apache.iotdb.commons.schema.SchemaConstant;
import org.apache.iotdb.commons.utils.AuthUtils;
import org.apache.iotdb.commons.utils.StatusUtils;
import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.commons.utils.TimePartitionUtils;
import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
import org.apache.iotdb.confignode.conf.ConfigNodeConstant;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
@@ -116,6 +117,7 @@ import
org.apache.iotdb.confignode.rpc.thrift.TGetJarInListResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetLocationForTriggerResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetModelInfoReq;
import org.apache.iotdb.confignode.rpc.thrift.TGetModelInfoResp;
+import org.apache.iotdb.confignode.rpc.thrift.TGetPartitionParameterResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesReq;
import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetPipePluginTableResp;
@@ -168,6 +170,7 @@ import
org.apache.iotdb.confignode.rpc.thrift.TUpdateModelInfoReq;
import org.apache.iotdb.confignode.rpc.thrift.TUpdateModelStateReq;
import org.apache.iotdb.confignode.service.ConfigNode;
import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.queryengine.plan.statement.AuthorType;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -1068,4 +1071,11 @@ public class ConfigNodeRPCServiceProcessor implements
IConfigNodeRPCService.Ifac
public TThrottleQuotaResp getThrottleQuota() {
return configManager.getThrottleQuota();
}
+
+ @Override
+ public TGetPartitionParameterResp getPartitionParameter() {
+ return new TGetPartitionParameterResp(
+ TimePartitionUtils.getTimePartitionInterval(),
+ IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/audit/AuditLogger.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/audit/AuditLogger.java
index 7c465d34a6b..c36ecf5aefd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/audit/AuditLogger.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/audit/AuditLogger.java
@@ -30,8 +30,8 @@ import org.apache.iotdb.db.protocol.session.IClientSession;
import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeDevicePathCache;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.statement.Statement;
import org.apache.iotdb.db.queryengine.plan.statement.StatementType;
import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 6c9722ab1fd..e9cbb4d33d7 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -29,6 +29,7 @@ import org.apache.iotdb.db.audit.AuditLogOperation;
import org.apache.iotdb.db.audit.AuditLogStorage;
import org.apache.iotdb.db.exception.LoadConfigurationException;
import org.apache.iotdb.db.protocol.thrift.impl.ClientRPCServiceImpl;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.constant.CompactionValidationLevel;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.constant.CrossCompactionPerformer;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.constant.InnerSeqCompactionPerformer;
@@ -933,10 +934,7 @@ public class IoTDBConfig {
*/
private int maxClientNumForEachNode =
DefaultProperty.MAX_CLIENT_NUM_FOR_EACH_NODE;
- /**
- * Cache size of partition cache in {@link
- * org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher}
- */
+ /** Cache size of partition cache in {@link ClusterPartitionFetcher} */
private int partitionCacheSize = 1000;
private int devicePathCacheSize = 500_000;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
index 3414b7a9b29..f4536a707f8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
@@ -70,6 +70,17 @@ public class PipeConnectorConstant {
public static final String CONNECTOR_OPC_UA_HTTPS_BIND_PORT_KEY =
"connector.opcua.https.port";
public static final int CONNECTOR_OPC_UA_HTTPS_BIND_PORT_DEFAULT_VALUE =
8443;
+ public static final String CONNECTOR_LOCAL_SPLIT_ENABLE_KEY =
"connector.local-split.enable";
+ public static final String CONNECTOR_EXTERNAL_CONFIG_NODES_KEY =
+ "connector.external.config-nodes";
+ public static final String CONNECTOR_SPLIT_MAX_SIZE_KEY =
"connector.split.max-size";
+ public static final int CONNECTOR_SPLIT_MAX_SIZE_DEFAULT_VALUE = 64 * 1024 *
1024;
+ public static final String CONNECTOR_SPLIT_MAX_CONCURRENT_FILE_KEY =
+ "connector.split.max-concurrent-file";
+ public static final int CONNECTOR_SPLIT_MAX_CONCURRENT_FILE_DEFAULT_VALUE =
16;
+ public static final String CONNECTOR_EXTERNAL_USER_NAME_KEY =
"connector.external.user-name";
+ public static final String CONNECTOR_EXTERNAL_USER_NAME_DEFAULT_VALUE =
"root";
+
private PipeConnectorConstant() {
throw new IllegalStateException("Utility class");
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
index 84b8865604c..88e1f4d2b92 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
@@ -23,7 +23,10 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.ClientPoolFactory;
import org.apache.iotdb.commons.client.IClientManager;
import
org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient;
+import org.apache.iotdb.commons.client.property.ThriftClientProperty;
+import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.confignode.rpc.thrift.TGetPartitionParameterResp;
import
org.apache.iotdb.db.pipe.connector.payload.evolvable.builder.IoTDBThriftAsyncPipeTransferBatchReqBuilder;
import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferHandshakeReq;
import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferTabletBinaryReq;
@@ -39,19 +42,29 @@ import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import
org.apache.iotdb.db.pipe.event.common.tsfile.PipeBatchTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import
org.apache.iotdb.db.queryengine.execution.load.DataPartitionBatchFetcher;
+import org.apache.iotdb.db.queryengine.execution.load.TsFileSplitSender;
+import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ExternalPartitionFetcher;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileNode;
import org.apache.iotdb.pipe.api.PipeConnector;
import
org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileBatchInsertionEvent;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
import org.apache.iotdb.pipe.api.exception.PipeConnectionException;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp;
+import org.apache.iotdb.session.util.SessionUtils;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.thrift.TException;
@@ -62,7 +75,9 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Comparator;
+import java.util.List;
import java.util.Optional;
import java.util.PriorityQueue;
import java.util.concurrent.PriorityBlockingQueue;
@@ -70,6 +85,15 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_EXTERNAL_CONFIG_NODES_KEY;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_EXTERNAL_USER_NAME_DEFAULT_VALUE;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_EXTERNAL_USER_NAME_KEY;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_LOCAL_SPLIT_ENABLE_KEY;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_SPLIT_MAX_CONCURRENT_FILE_DEFAULT_VALUE;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_SPLIT_MAX_CONCURRENT_FILE_KEY;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_SPLIT_MAX_SIZE_DEFAULT_VALUE;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_SPLIT_MAX_SIZE_KEY;
+
public class IoTDBThriftAsyncConnector extends IoTDBConnector {
private static final Logger LOGGER =
LoggerFactory.getLogger(IoTDBThriftAsyncConnector.class);
@@ -82,6 +106,10 @@ public class IoTDBThriftAsyncConnector extends
IoTDBConnector {
ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER_HOLDER = new AtomicReference<>();
private final IClientManager<TEndPoint, AsyncPipeDataTransferServiceClient>
asyncPipeDataTransferClientManager;
+ private static final AtomicReference<IClientManager<TEndPoint,
SyncDataNodeInternalServiceClient>>
+ SYNC_DATA_NODE_CLIENT_MANAGER_HOLDER = new AtomicReference<>();
+ private final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient>
+ syncDataNodeClientManager;
private final IoTDBThriftSyncConnector retryConnector = new
IoTDBThriftSyncConnector();
private final PriorityBlockingQueue<Pair<Long, Event>> retryEventQueue =
@@ -94,6 +122,19 @@ public class IoTDBThriftAsyncConnector extends
IoTDBConnector {
private IoTDBThriftAsyncPipeTransferBatchReqBuilder tabletBatchBuilder;
+ // parameters for local split feature
+ // when enabled, TsFiles will be split in this node and sent to direct
managers in another
+ // cluster,
+ // using an optimized procedure
+ private boolean useLocalSplit;
+ private List<TEndPoint> targetConfigNodes = new ArrayList<>();
+ private ThriftClientProperty thriftClientProperty = new
ThriftClientProperty.Builder().build();
+ private int targetSeriesSlotNum;
+ private long targetPartitionInterval;
+ private int splitMaxSize;
+ private int maxConcurrentFileNum;
+ private String targetUserName;
+
public IoTDBThriftAsyncConnector() {
if (ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER_HOLDER.get() == null) {
synchronized (IoTDBThriftAsyncConnector.class) {
@@ -102,10 +143,15 @@ public class IoTDBThriftAsyncConnector extends
IoTDBConnector {
new IClientManager.Factory<TEndPoint,
AsyncPipeDataTransferServiceClient>()
.createClientManager(
new
ClientPoolFactory.AsyncPipeDataTransferServiceClientPoolFactory()));
+ SYNC_DATA_NODE_CLIENT_MANAGER_HOLDER.set(
+ new IClientManager.Factory<TEndPoint,
SyncDataNodeInternalServiceClient>()
+ .createClientManager(
+ new
ClientPoolFactory.SyncDataNodeInternalServiceClientPoolFactory()));
}
}
}
asyncPipeDataTransferClientManager =
ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER_HOLDER.get();
+ syncDataNodeClientManager = SYNC_DATA_NODE_CLIENT_MANAGER_HOLDER.get();
}
@Override
@@ -121,11 +167,44 @@ public class IoTDBThriftAsyncConnector extends
IoTDBConnector {
retryConnector.customize(parameters, configuration);
+ useLocalSplit =
parameters.getBooleanOrDefault(CONNECTOR_LOCAL_SPLIT_ENABLE_KEY, false);
+
+ if (useLocalSplit) {
+ String configNodeUrls =
+ parameters.getStringOrDefault(CONNECTOR_EXTERNAL_CONFIG_NODES_KEY,
"");
+ String[] urlSplits = configNodeUrls.split(",");
+ for (String urlSplit : urlSplits) {
+ targetConfigNodes.add(SessionUtils.parseNodeUrl(urlSplit));
+ }
+
+ queryTargetPartitionParameters();
+
+ splitMaxSize =
+ parameters.getIntOrDefault(
+ CONNECTOR_SPLIT_MAX_SIZE_KEY,
CONNECTOR_SPLIT_MAX_SIZE_DEFAULT_VALUE);
+ maxConcurrentFileNum =
+ parameters.getIntOrDefault(
+ CONNECTOR_SPLIT_MAX_CONCURRENT_FILE_KEY,
+ CONNECTOR_SPLIT_MAX_CONCURRENT_FILE_DEFAULT_VALUE);
+ targetUserName =
+ parameters.getStringOrDefault(
+ CONNECTOR_EXTERNAL_USER_NAME_KEY,
CONNECTOR_EXTERNAL_USER_NAME_DEFAULT_VALUE);
+ }
+
if (isTabletBatchModeEnabled) {
tabletBatchBuilder = new
IoTDBThriftAsyncPipeTransferBatchReqBuilder(parameters);
}
}
+ private void queryTargetPartitionParameters() throws TException {
+ try (ConfigNodeClient client =
+ new ConfigNodeClient(targetConfigNodes, thriftClientProperty, null)) {
+ TGetPartitionParameterResp partitionParameter =
client.getPartitionParameter();
+ targetPartitionInterval = partitionParameter.partitionInterval;
+ targetSeriesSlotNum = partitionParameter.seriesPartitionSlotNum;
+ }
+ }
+
@Override
// Synchronized to avoid close connector when transfer event
public synchronized void handshake() throws Exception {
@@ -314,6 +393,74 @@ public class IoTDBThriftAsyncConnector extends
IoTDBConnector {
transfer(requestCommitId, pipeTransferTsFileInsertionEventHandler);
}
+ @Override
+ public void transfer(TsFileBatchInsertionEvent tsFileInsertionEvent) throws
Exception {
+ transferQueuedEventsIfNecessary();
+ transferBatchedEventsIfNecessary();
+
+ if (!(tsFileInsertionEvent instanceof PipeBatchTsFileInsertionEvent)) {
+ LOGGER.warn(
+ "IoTDBThriftAsyncConnector only support
PipeBatchTsFileInsertionEvent. Current event: {}.",
+ tsFileInsertionEvent);
+ return;
+ }
+
+ if (((EnrichedEvent) tsFileInsertionEvent).shouldParsePatternOrTime()) {
+ try {
+ for (final TabletInsertionEvent event :
tsFileInsertionEvent.toTabletInsertionEvents()) {
+ transfer(event);
+ }
+ } finally {
+ tsFileInsertionEvent.close();
+ }
+ return;
+ }
+
+ final PipeBatchTsFileInsertionEvent pipeTsFileInsertionEvent =
+ (PipeBatchTsFileInsertionEvent) tsFileInsertionEvent;
+ pipeTsFileInsertionEvent.waitForTsFileClose();
+
+ if (!useLocalSplit) {
+ List<PipeTsFileInsertionEvent> pipeTsFileInsertionEvents =
+ pipeTsFileInsertionEvent.toSingleFileEvents();
+ for (PipeTsFileInsertionEvent fileInsertionEvent :
pipeTsFileInsertionEvents) {
+ transfer(fileInsertionEvent);
+ }
+ } else {
+ final long requestCommitId = commitIdGenerator.incrementAndGet();
+ transfer(requestCommitId, pipeTsFileInsertionEvent);
+ }
+ }
+
+ private void transfer(
+ long requestCommitId, PipeBatchTsFileInsertionEvent
pipeTsFileInsertionEvent)
+ throws IOException {
+ LoadTsFileNode loadTsFileNode =
+ new LoadTsFileNode(
+ new PlanNodeId("Pipe-" + requestCommitId),
pipeTsFileInsertionEvent.getResources());
+
+ IPartitionFetcher partitionFetcher =
+ new ExternalPartitionFetcher(targetConfigNodes, thriftClientProperty,
targetSeriesSlotNum);
+ DataPartitionBatchFetcher dataPartitionBatchFetcher =
+ new DataPartitionBatchFetcher(partitionFetcher);
+ TsFileSplitSender splitSender =
+ new TsFileSplitSender(
+ loadTsFileNode,
+ dataPartitionBatchFetcher,
+ targetPartitionInterval,
+ syncDataNodeClientManager,
+ false,
+ splitMaxSize,
+ maxConcurrentFileNum,
+ targetUserName);
+ splitSender.start();
+ LOGGER.info(
+ "Sending {} files to {} complete: {}",
+ pipeTsFileInsertionEvent.getTsFiles().size(),
+ targetConfigNodes,
+ splitSender.getStatistic());
+ }
+
private void transfer(
long requestCommitId,
PipeTransferTsFileInsertionEventHandler
pipeTransferTsFileInsertionEventHandler) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeBatchTsFileInsertionEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeBatchTsFileInsertionEvent.java
index 284c84096ce..222532c5673 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeBatchTsFileInsertionEvent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeBatchTsFileInsertionEvent.java
@@ -19,11 +19,6 @@
package org.apache.iotdb.db.pipe.event.common.tsfile;
-import java.io.File;
-import java.io.IOException;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Collectors;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
@@ -33,13 +28,20 @@ import
org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileBatchInsertionEvent;
-import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
import org.apache.iotdb.pipe.api.exception.PipeException;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class PipeBatchTsFileInsertionEvent extends EnrichedEvent implements
- TsFileBatchInsertionEvent {
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+
+public class PipeBatchTsFileInsertionEvent extends EnrichedEvent
+ implements TsFileBatchInsertionEvent {
private static final Logger LOGGER =
LoggerFactory.getLogger(PipeBatchTsFileInsertionEvent.class);
@@ -122,6 +124,10 @@ public class PipeBatchTsFileInsertionEvent extends
EnrichedEvent implements
return tsFiles;
}
+ public List<TsFileResource> getResources() {
+ return resources;
+ }
+
/////////////////////////// EnrichedEvent ///////////////////////////
@Override
@@ -165,9 +171,7 @@ public class PipeBatchTsFileInsertionEvent extends
EnrichedEvent implements
waitForTsFileClose();
return resources.get(resources.size() -
1).getMaxProgressIndexAfterClose();
} catch (InterruptedException e) {
- LOGGER.warn(
- String.format(
- "Interrupted when waiting for closing TsFiles %s.", resources));
+ LOGGER.warn(String.format("Interrupted when waiting for closing TsFiles
%s.", resources));
Thread.currentThread().interrupt();
return MinimumProgressIndex.INSTANCE;
}
@@ -202,8 +206,7 @@ public class PipeBatchTsFileInsertionEvent extends
EnrichedEvent implements
close();
final String errorMsg =
- String.format(
- "Interrupted when waiting for closing TsFiles %s.", resources);
+ String.format("Interrupted when waiting for closing TsFiles %s.",
resources);
LOGGER.warn(errorMsg, e);
throw new PipeException(errorMsg);
} catch (IOException e) {
@@ -237,4 +240,24 @@ public class PipeBatchTsFileInsertionEvent extends
EnrichedEvent implements
+ isClosed
+ '}';
}
+
+ public List<PipeTsFileInsertionEvent> toSingleFileEvents() {
+ List<PipeTsFileInsertionEvent> result = new ArrayList<>();
+ for (TsFileResource resource : resources) {
+ result.add(
+ new PipeTsFileInsertionEvent(
+ resource,
+ isGeneratedByPipe,
+ pipeTaskMeta,
+ getPattern(),
+ startTime,
+ endTime,
+ isTsFileResourceCoveredByTimeRange(resource)));
+ }
+ return result;
+ }
+
+ protected boolean isTsFileResourceCoveredByTimeRange(TsFileResource
resource) {
+ return startTime <= resource.getFileStartTime() && endTime >=
resource.getFileEndTime();
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileListInsertionDataContainer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileListInsertionDataContainer.java
index 6055f6f845d..e163243cc1e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileListInsertionDataContainer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileListInsertionDataContainer.java
@@ -19,14 +19,6 @@
package org.apache.iotdb.db.pipe.event.common.tsfile;
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.NoSuchElementException;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
@@ -43,13 +35,23 @@ import
org.apache.iotdb.tsfile.read.expression.impl.GlobalTimeExpression;
import org.apache.iotdb.tsfile.read.filter.TimeFilter;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.write.record.Tablet;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+
public class TsFileListInsertionDataContainer implements AutoCloseable {
- private static final Logger LOGGER = LoggerFactory.getLogger(
- TsFileListInsertionDataContainer.class);
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(TsFileListInsertionDataContainer.class);
private final String pattern; // used to filter data
private final IExpression timeFilterExpression; // used to filter data
@@ -65,9 +67,8 @@ public class TsFileListInsertionDataContainer implements
AutoCloseable {
private final List<Map<String, Boolean>> deviceIsAlignedMaps;
private final List<Map<String, TSDataType>> measurementDataTypeMaps;
- public TsFileListInsertionDataContainer(List<File> tsFiles, String pattern,
long startTime,
- long endTime)
- throws IOException {
+ public TsFileListInsertionDataContainer(
+ List<File> tsFiles, String pattern, long startTime, long endTime) throws
IOException {
this(tsFiles, pattern, startTime, endTime, null, null);
}
@@ -94,8 +95,8 @@ public class TsFileListInsertionDataContainer implements
AutoCloseable {
tsFileSequenceReaders = new ArrayList<>();
tsFileReaders = new ArrayList<>();
for (File tsFile : tsFiles) {
- TsFileSequenceReader tsFileSequenceReader = new TsFileSequenceReader(
- tsFile.getAbsolutePath(), true, true);
+ TsFileSequenceReader tsFileSequenceReader =
+ new TsFileSequenceReader(tsFile.getAbsolutePath(), true, true);
tsFileSequenceReaders.add(tsFileSequenceReader);
tsFileReaders.add(new TsFileReader(tsFileSequenceReader));
}
@@ -167,9 +168,7 @@ public class TsFileListInsertionDataContainer implements
AutoCloseable {
return deviceIsAlignedResultMap;
}
- /**
- * @return TabletInsertionEvent in a streaming way
- */
+ /** @return TabletInsertionEvent in a streaming way */
public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {
return () ->
new Iterator<TabletInsertionEvent>() {
@@ -189,8 +188,8 @@ public class TsFileListInsertionDataContainer implements
AutoCloseable {
}
}
- final Map.Entry<String, List<String>> entry =
deviceMeasurementsMapIterators.get(
- currFileIndex).next();
+ final Map.Entry<String, List<String>> entry =
+ deviceMeasurementsMapIterators.get(currFileIndex).next();
try {
tabletIterator =
@@ -217,8 +216,8 @@ public class TsFileListInsertionDataContainer implements
AutoCloseable {
}
final Tablet tablet = tabletIterator.next();
- final boolean isAligned = deviceIsAlignedMaps.get(currFileIndex)
- .getOrDefault(tablet.deviceId, false);
+ final boolean isAligned =
+
deviceIsAlignedMaps.get(currFileIndex).getOrDefault(tablet.deviceId, false);
final TabletInsertionEvent next;
if (!hasNext()) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/BatchedTsFileExtractor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/BatchedTsFileExtractor.java
index 463c529fedb..441b74705ab 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/BatchedTsFileExtractor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/BatchedTsFileExtractor.java
@@ -18,16 +18,18 @@
*/
package org.apache.iotdb.db.pipe.extractor.historical;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
import
org.apache.iotdb.db.pipe.event.common.tsfile.PipeBatchTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.resource.PipeResourceManager;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.pipe.api.event.Event;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
/**
* Similar to the base class, but it batches several files as an event to
enable further
* optimization during the latter transfer.
@@ -56,16 +58,17 @@ public class BatchedTsFileExtractor extends
PipeHistoricalDataRegionTsFileExtrac
tsFileResourceList.add(resource);
}
- final PipeBatchTsFileInsertionEvent event = new
PipeBatchTsFileInsertionEvent(
- tsFileResourceList,
- false,
- pipeTaskMeta,
- pattern,
- historicalDataExtractionStartTime,
- historicalDataExtractionEndTime,
- !(isTsFileResourceCoveredByTimeRange(tsFileResourceList.get(0))
- && isTsFileResourceCoveredByTimeRange(
- tsFileResourceList.get(tsFileResourceList.size() - 1))));
+ final PipeBatchTsFileInsertionEvent event =
+ new PipeBatchTsFileInsertionEvent(
+ tsFileResourceList,
+ false,
+ pipeTaskMeta,
+ pattern,
+ historicalDataExtractionStartTime,
+ historicalDataExtractionEndTime,
+ !(isTsFileResourceCoveredByTimeRange(tsFileResourceList.get(0))
+ && isTsFileResourceCoveredByTimeRange(
+ tsFileResourceList.get(tsFileResourceList.size() - 1))));
event.increaseReferenceCount(BatchedTsFileExtractor.class.getName());
@@ -81,5 +84,4 @@ public class BatchedTsFileExtractor extends
PipeHistoricalDataRegionTsFileExtrac
return event;
}
-
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/airgap/IoTDBAirGapReceiver.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/airgap/IoTDBAirGapReceiver.java
index 11dd9b29be5..7a4b6b8fa9c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/airgap/IoTDBAirGapReceiver.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/airgap/IoTDBAirGapReceiver.java
@@ -26,8 +26,8 @@ import
org.apache.iotdb.db.pipe.connector.payload.airgap.AirGapELanguageConstant
import org.apache.iotdb.db.pipe.connector.payload.airgap.AirGapOneByteResponse;
import
org.apache.iotdb.db.pipe.connector.payload.airgap.AirGapPseudoTPipeTransferRequest;
import org.apache.iotdb.db.pipe.receiver.thrift.IoTDBThriftReceiverAgent;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.rpc.TSStatusCode;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/legacy/loader/ILoader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/legacy/loader/ILoader.java
index eb759ab82ed..14aa3b459ba 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/legacy/loader/ILoader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/legacy/loader/ILoader.java
@@ -19,8 +19,8 @@
package org.apache.iotdb.db.pipe.receiver.legacy.loader;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
index 2cf05ec7838..29fd5c4a8a7 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
@@ -84,6 +84,7 @@ import
org.apache.iotdb.confignode.rpc.thrift.TGetJarInListResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetLocationForTriggerResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetModelInfoReq;
import org.apache.iotdb.confignode.rpc.thrift.TGetModelInfoResp;
+import org.apache.iotdb.confignode.rpc.thrift.TGetPartitionParameterResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesReq;
import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetPipePluginTableResp;
@@ -284,7 +285,12 @@ public class ConfigNodeClient implements
IConfigNodeRPCService.Iface, ThriftClie
@Override
public void close() {
- clientManager.returnClient(configRegionId, this);
+ if (clientManager != null) {
+ clientManager.returnClient(configRegionId, this);
+ } else {
+ // clients not managed by ClientManages will be destroyed after being
closed
+ invalidate();
+ }
}
@Override
@@ -294,7 +300,9 @@ public class ConfigNodeClient implements
IConfigNodeRPCService.Iface, ThriftClie
@Override
public void invalidateAll() {
- clientManager.clear(ConfigNodeInfo.CONFIG_REGION_ID);
+ if (clientManager != null) {
+ clientManager.clear(ConfigNodeInfo.CONFIG_REGION_ID);
+ }
}
@Override
@@ -1041,6 +1049,11 @@ public class ConfigNodeClient implements
IConfigNodeRPCService.Iface, ThriftClie
() -> client.getThrottleQuota(), resp ->
!updateConfigNodeLeader(resp.status));
}
+ @Override
+ public TGetPartitionParameterResp getPartitionParameter() throws TException {
+ return executeRemoteCallWithRetry(() -> client.getPartitionParameter(),
resp -> true);
+ }
+
public static class Factory extends ThriftClientFactory<ConfigRegionId,
ConfigNodeClient> {
public Factory(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/DataNodeInternalClient.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/DataNodeInternalClient.java
index 925a4b450f7..a76558b4ee5 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/DataNodeInternalClient.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/DataNodeInternalClient.java
@@ -29,8 +29,8 @@ import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.protocol.thrift.OperationType;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/MPPPublishHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/MPPPublishHandler.java
index 4063c7c44d1..eb960599a28 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/MPPPublishHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/MPPPublishHandler.java
@@ -26,9 +26,9 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.protocol.session.MqttClientSession;
import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeDevicePathCache;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/GrafanaApiServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/GrafanaApiServiceImpl.java
index 882dec8b86e..c9e614d308a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/GrafanaApiServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/GrafanaApiServiceImpl.java
@@ -31,8 +31,8 @@ import
org.apache.iotdb.db.protocol.rest.v1.model.ExpressionRequest;
import org.apache.iotdb.db.protocol.rest.v1.model.SQL;
import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/RestApiServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/RestApiServiceImpl.java
index e9cd7262f48..12a5ae0ae8a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/RestApiServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/RestApiServiceImpl.java
@@ -34,8 +34,8 @@ import
org.apache.iotdb.db.protocol.rest.v1.model.InsertTabletRequest;
import org.apache.iotdb.db.protocol.rest.v1.model.SQL;
import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/GrafanaApiServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/GrafanaApiServiceImpl.java
index a7eb24066e3..76d0ed64fce 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/GrafanaApiServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/GrafanaApiServiceImpl.java
@@ -31,8 +31,8 @@ import
org.apache.iotdb.db.protocol.rest.v2.model.ExpressionRequest;
import org.apache.iotdb.db.protocol.rest.v2.model.SQL;
import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java
index 643e3f91036..658ab43fab8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java
@@ -34,8 +34,8 @@ import
org.apache.iotdb.db.protocol.rest.v2.model.InsertTabletRequest;
import org.apache.iotdb.db.protocol.rest.v2.model.SQL;
import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
index 94db836be4a..a7bce86bd16 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
@@ -67,9 +67,9 @@ import
org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesAg
import
org.apache.iotdb.db.queryengine.execution.operator.source.SeriesAggregationScanOperator;
import
org.apache.iotdb.db.queryengine.execution.operator.source.SeriesScanOperator;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index 884469be6cf..a2a6289f74f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -76,10 +76,10 @@ import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceState;
import
org.apache.iotdb.db.queryengine.execution.operator.schema.source.ISchemaSource;
import
org.apache.iotdb.db.queryengine.execution.operator.schema.source.SchemaSourceFactory;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeDevicePathCache;
import
org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/MLNodeRPCServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/MLNodeRPCServiceImpl.java
index 46f1a966112..bf78e0422f4 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/MLNodeRPCServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/MLNodeRPCServiceImpl.java
@@ -27,8 +27,8 @@ import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.protocol.thrift.OperationType;
import org.apache.iotdb.db.queryengine.common.header.DatasetHeader;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analyzer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analyzer.java
index f7ca04f7d05..96031fbfc49 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analyzer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analyzer.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.queryengine.plan.analyze;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.statement.Statement;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ClusterPartitionFetcher.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/partition/BasicPartitionFetcher.java
similarity index 90%
rename from
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ClusterPartitionFetcher.java
rename to
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/partition/BasicPartitionFetcher.java
index 1069aca33b8..8536457fbf2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ClusterPartitionFetcher.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/partition/BasicPartitionFetcher.java
@@ -17,15 +17,13 @@
* under the License.
*/
-package org.apache.iotdb.db.queryengine.plan.analyze;
+package org.apache.iotdb.db.queryengine.plan.analyze.partition;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
-import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.exception.ClientManagerException;
-import org.apache.iotdb.commons.consensus.ConfigRegionId;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.partition.DataPartition;
import org.apache.iotdb.commons.partition.DataPartitionQueryParam;
@@ -44,15 +42,12 @@ import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.sql.StatementAnalyzeException;
import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
-import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
-import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.cache.partition.PartitionCache;
import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
@@ -63,40 +58,26 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
-public class ClusterPartitionFetcher implements IPartitionFetcher {
+public abstract class BasicPartitionFetcher implements IPartitionFetcher {
- private static final Logger logger =
LoggerFactory.getLogger(ClusterPartitionFetcher.class);
- private static final IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
+ protected static final IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
- private final SeriesPartitionExecutor partitionExecutor;
+ protected final SeriesPartitionExecutor partitionExecutor;
- private final PartitionCache partitionCache;
+ protected final PartitionCache partitionCache;
- private final IClientManager<ConfigRegionId, ConfigNodeClient>
configNodeClientManager =
- ConfigNodeClientManager.getInstance();
-
- private static final class ClusterPartitionFetcherHolder {
-
- private static final ClusterPartitionFetcher INSTANCE = new
ClusterPartitionFetcher();
-
- private ClusterPartitionFetcherHolder() {}
- }
-
- public static ClusterPartitionFetcher getInstance() {
- return ClusterPartitionFetcherHolder.INSTANCE;
- }
-
- private ClusterPartitionFetcher() {
+ protected BasicPartitionFetcher(int seriesPartitionSlotNum) {
this.partitionExecutor =
SeriesPartitionExecutor.getSeriesPartitionExecutor(
- config.getSeriesPartitionExecutorClass(),
config.getSeriesPartitionSlotNum());
+ config.getSeriesPartitionExecutorClass(), seriesPartitionSlotNum);
this.partitionCache = new PartitionCache();
}
+ protected abstract ConfigNodeClient getClient() throws
ClientManagerException, TException;
+
@Override
public SchemaPartition getSchemaPartition(PathPatternTree patternTree) {
- try (ConfigNodeClient client =
- configNodeClientManager.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID))
{
+ try (ConfigNodeClient client = getClient()) {
patternTree.constructTree();
List<String> devicePaths = patternTree.getAllDevicePatterns();
Map<String, List<String>> storageGroupToDeviceMap =
@@ -126,8 +107,7 @@ public class ClusterPartitionFetcher implements
IPartitionFetcher {
@Override
public SchemaPartition getOrCreateSchemaPartition(PathPatternTree
patternTree, String userName) {
- try (ConfigNodeClient client =
- configNodeClientManager.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID))
{
+ try (ConfigNodeClient client = getClient()) {
patternTree.constructTree();
List<String> devicePaths = patternTree.getAllDevicePatterns();
Map<String, List<String>> storageGroupToDeviceMap =
@@ -158,8 +138,7 @@ public class ClusterPartitionFetcher implements
IPartitionFetcher {
@Override
public SchemaNodeManagementPartition
getSchemaNodeManagementPartitionWithLevel(
PathPatternTree patternTree, PathPatternTree scope, Integer level) {
- try (ConfigNodeClient client =
- configNodeClientManager.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID))
{
+ try (ConfigNodeClient client = getClient()) {
patternTree.constructTree();
TSchemaNodeManagementResp schemaNodeManagementResp =
client.getSchemaNodeManagementPartition(
@@ -177,8 +156,7 @@ public class ClusterPartitionFetcher implements
IPartitionFetcher {
Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap) {
DataPartition dataPartition =
partitionCache.getDataPartition(sgNameToQueryParamsMap);
if (null == dataPartition) {
- try (ConfigNodeClient client =
-
configNodeClientManager.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+ try (ConfigNodeClient client = getClient()) {
TDataPartitionTableResp dataPartitionTableResp =
client.getDataPartitionTable(constructDataPartitionReqForQuery(sgNameToQueryParamsMap));
if (dataPartitionTableResp.getStatus().getCode()
@@ -204,8 +182,7 @@ public class ClusterPartitionFetcher implements
IPartitionFetcher {
// In this method, we must fetch from config node because it contains -oo
or +oo
// and there is no need to update cache because since we will never fetch
it from cache, the
// update operation will be only time waste
- try (ConfigNodeClient client =
- configNodeClientManager.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID))
{
+ try (ConfigNodeClient client = getClient()) {
TDataPartitionTableResp dataPartitionTableResp =
client.getDataPartitionTable(constructDataPartitionReqForQuery(sgNameToQueryParamsMap));
if (dataPartitionTableResp.getStatus().getCode()
@@ -228,8 +205,7 @@ public class ClusterPartitionFetcher implements
IPartitionFetcher {
DataPartition dataPartition =
partitionCache.getDataPartition(sgNameToQueryParamsMap);
if (null == dataPartition) {
// Do not use data partition cache
- try (ConfigNodeClient client =
-
configNodeClientManager.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+ try (ConfigNodeClient client = getClient()) {
TDataPartitionTableResp dataPartitionTableResp =
client.getOrCreateDataPartitionTable(constructDataPartitionReq(sgNameToQueryParamsMap));
if (dataPartitionTableResp.getStatus().getCode()
@@ -258,8 +234,7 @@ public class ClusterPartitionFetcher implements
IPartitionFetcher {
DataPartition dataPartition =
partitionCache.getDataPartition(splitDataPartitionQueryParams);
if (null == dataPartition) {
- try (ConfigNodeClient client =
-
configNodeClientManager.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+ try (ConfigNodeClient client = getClient()) {
TDataPartitionReq req =
constructDataPartitionReq(splitDataPartitionQueryParams);
TDataPartitionTableResp dataPartitionTableResp =
client.getOrCreateDataPartitionTable(req);
@@ -339,6 +314,7 @@ public class ClusterPartitionFetcher implements
IPartitionFetcher {
}
private static class ComplexTimeSlotList {
+
Set<TTimePartitionSlot> timeSlotList;
boolean needLeftAll;
boolean needRightAll;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/partition/ClusterPartitionFetcher.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/partition/ClusterPartitionFetcher.java
new file mode 100644
index 00000000000..f8d218bf71f
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/partition/ClusterPartitionFetcher.java
@@ -0,0 +1,52 @@
+/*
+ * 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.db.queryengine.plan.analyze.partition;
+
+import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.consensus.ConfigRegionId;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+
+public class ClusterPartitionFetcher extends BasicPartitionFetcher {
+ private final IClientManager<ConfigRegionId, ConfigNodeClient>
configNodeClientManager =
+ ConfigNodeClientManager.getInstance();
+
+ private static final class ClusterPartitionFetcherHolder {
+
+ private static final ClusterPartitionFetcher INSTANCE = new
ClusterPartitionFetcher();
+
+ private ClusterPartitionFetcherHolder() {}
+ }
+
+ public static ClusterPartitionFetcher getInstance() {
+ return ClusterPartitionFetcherHolder.INSTANCE;
+ }
+
+ private ClusterPartitionFetcher() {
+ super(config.getSeriesPartitionSlotNum());
+ }
+
+ @Override
+ protected ConfigNodeClient getClient() throws ClientManagerException {
+ return
configNodeClientManager.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/partition/ExternalPartitionFetcher.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/partition/ExternalPartitionFetcher.java
new file mode 100644
index 00000000000..ac1624c85c6
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/partition/ExternalPartitionFetcher.java
@@ -0,0 +1,52 @@
+/*
+ * 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.db.queryengine.plan.analyze.partition;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.client.property.ThriftClientProperty;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+
+import org.apache.thrift.TException;
+
+import java.util.List;
+
+/**
+ * ExternalPartitionFetcher fetches partition info from another cluster for
cross-cluster data
+ * communications.
+ */
+public class ExternalPartitionFetcher extends BasicPartitionFetcher {
+
+ private List<TEndPoint> externalConfigNodes;
+ private ThriftClientProperty property;
+
+ public ExternalPartitionFetcher(
+ List<TEndPoint> externalConfigNodes,
+ ThriftClientProperty property,
+ int seriesPartitionSlotNum) {
+ super(seriesPartitionSlotNum);
+ this.externalConfigNodes = externalConfigNodes;
+ this.property = property;
+ }
+
+ @Override
+ protected ConfigNodeClient getClient() throws TException {
+ return new ConfigNodeClient(externalConfigNodes, property, null);
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/AutoCreateSchemaExecutor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/AutoCreateSchemaExecutor.java
index 5e9e9964314..8abd996d8c1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/AutoCreateSchemaExecutor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/AutoCreateSchemaExecutor.java
@@ -34,8 +34,8 @@ import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.QueryType;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.MeasurementGroup;
import org.apache.iotdb.db.queryengine.plan.statement.Statement;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ClusterSchemaFetchExecutor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ClusterSchemaFetchExecutor.java
index 3bb32279e0a..834b14b10ff 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ClusterSchemaFetchExecutor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ClusterSchemaFetchExecutor.java
@@ -30,8 +30,8 @@ import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.QueryType;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
import org.apache.iotdb.db.queryengine.plan.statement.Statement;
import
org.apache.iotdb.db.queryengine.plan.statement.internal.SchemaFetchStatement;
diff --git
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
index e10cb01bd7d..12b944b8af8 100644
---
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
@@ -111,8 +111,8 @@ import
org.apache.iotdb.db.queryengine.common.schematree.ISchemaTree;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
import org.apache.iotdb.db.queryengine.plan.analyze.Analyzer;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.ExpressionAnalyzer;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java
index 8e08e14c633..0a561289fc2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java
@@ -37,8 +37,8 @@ import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
+import
org.apache.iotdb.db.queryengine.plan.analyze.partition.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
index 3fb9632a2d6..03755731723 100644
--- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
@@ -830,6 +830,11 @@ struct TShowThrottleReq{
1: optional string userName;
}
+struct TGetPartitionParameterResp{
+ 1: required i64 partitionInterval;
+ 2: required i32 seriesPartitionSlotNum;
+}
+
service IConfigNodeRPCService {
// ======================================================
@@ -1463,5 +1468,7 @@ service IConfigNodeRPCService {
/** Get throttle quota information */
TThrottleQuotaResp getThrottleQuota()
+
+ TGetPartitionParameterResp getPartitionParameter()
}
