You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2023/06/07 03:01:41 UTC
[iotdb] 01/01: Pipe: refactor package structure (#10062)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch 1.2-pipe-refactor
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e2b06927784f3be99fdb00925cf62390662f16a0
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Jun 7 10:26:00 2023 +0800
Pipe: refactor package structure (#10062)
* Pipe: refactor package structure (db-main)
* Pipe: refactor package structure (db-test)
* Pipe: refactor package structure (pipe-api)
* Pipe: refactor package structure (confignode)
* Pipe: refactor package structure (node-commons)
(cherry picked from commit 37a539eb089778bf7f203d84e39388b390bf47e6)
---
.../consensus/response/pipe/PipeResp.java | 48 -------------------
.../consensus/response/pipe/PipeSinkResp.java | 48 -------------------
.../persistence/pipe/PipePluginInfo.java | 16 +++----
.../confignode/persistence/pipe/PipeTaskInfo.java | 23 +++++----
.../{task => }/AbstractOperatePipeProcedureV2.java | 8 ++--
.../impl}/pipe/PipeTaskOperation.java | 2 +-
.../pipe/plugin/CreatePipePluginProcedure.java | 8 ++--
.../impl/pipe/plugin/DropPipePluginProcedure.java | 6 +--
.../runtime/PipeHandleLeaderChangeProcedure.java | 10 ++--
.../runtime/PipeHandleMetaChangeProcedure.java | 8 ++--
.../impl/pipe/runtime/PipeMetaSyncProcedure.java | 4 +-
.../impl/pipe/task/CreatePipeProcedureV2.java | 18 ++++---
.../impl/pipe/task/DropPipeProcedureV2.java | 16 +++----
.../impl/pipe/task/StartPipeProcedureV2.java | 20 ++++----
.../impl/pipe/task/StopPipeProcedureV2.java | 20 ++++----
.../org/apache/iotdb/pipe/api/event/EventType.java | 26 ----------
.../event/dml/insertion/TsFileInsertionEvent.java | 8 ----
.../api/exception/PipeManagementException.java | 32 -------------
.../service/PipePluginExecutableManager.java | 6 +--
.../commons/pipe/utils/PipeBinaryTransformer.java | 32 -------------
.../pipe/{task/meta => }/PipeMetaDeSerTest.java | 6 ++-
.../db/engine/storagegroup/TsFileProcessor.java | 2 +-
.../db/pipe/agent/plugin/PipePluginAgent.java | 19 ++++----
.../pipe/agent/receiver/IoTDBThriftReceiver.java | 2 +-
.../db/pipe/agent/receiver/PipeReceiverAgent.java | 4 +-
.../db/pipe/agent/runtime/PipeAgentLauncher.java | 4 +-
.../collector/IoTDBDataRegionCollector.java | 22 ++++-----
.../PipeHistoricalDataRegionCollector.java | 2 +-
.../PipeHistoricalDataRegionTsFileCollector.java | 4 +-
.../realtime/PipeRealtimeDataRegionCollector.java | 6 +--
.../PipeRealtimeDataRegionFakeCollector.java | 4 +-
.../PipeRealtimeDataRegionHybridCollector.java | 8 ++--
.../PipeRealtimeDataRegionLogCollector.java | 8 ++--
.../PipeRealtimeDataRegionTsFileCollector.java | 8 ++--
.../realtime/assigner/DisruptorQueue.java | 2 +-
.../realtime/assigner/PipeDataRegionAssigner.java | 10 ++--
.../realtime/epoch}/TsFileEpoch.java | 4 +-
.../realtime/epoch}/TsFileEpochManager.java | 7 +--
.../realtime/epoch}/TsFileEpochStateMigrator.java | 2 +-
.../listener/PipeInsertionDataNodeListener.java | 8 ++--
.../matcher/CachedSchemaPatternMatcher.java | 6 +--
.../realtime/matcher/PipeDataRegionMatcher.java | 6 +--
.../IoTDBThriftConnectorClient.java | 2 +-
.../IoTDBThriftConnectorVersion.java | 2 +-
.../v1/IoTDBThriftConnectorV1.java | 22 ++++-----
.../v1/IoTDBThriftReceiverV1.java | 16 +++----
.../iotdb => connector}/v1/PipeRequestType.java | 2 +-
.../v1/reponse/PipeTransferFilePieceResp.java | 2 +-
.../v1/request/PipeTransferFilePieceReq.java | 6 +--
.../v1/request/PipeTransferFileSealReq.java | 6 +--
.../v1/request/PipeTransferHandshakeReq.java | 6 +--
.../v1/request/PipeTransferInsertNodeReq.java | 6 +--
.../v1/request/PipeTransferTabletReq.java | 6 +--
.../PipeHistoricalDataRegionFakeCollector.java | 56 ----------------------
.../db/pipe/{core => }/event/EnrichedEvent.java | 2 +-
.../event/common/row}/PipeDataTypeTransformer.java | 40 +++-------------
.../view/access => event/common/row}/PipeRow.java | 3 +-
.../common/row}/PipeRowCollector.java | 5 +-
.../PipeInsertNodeTabletInsertionEvent.java | 5 +-
.../tablet}/PipeRawTabletInsertionEvent.java | 3 +-
.../tablet}/TabletInsertionDataContainer.java | 6 +--
.../common/tsfile}/PipeTsFileInsertionEvent.java | 10 +---
.../tsfile}/TsFileInsertionDataContainer.java | 4 +-
.../tsfile}/TsFileInsertionDataTabletIterator.java | 2 +-
.../event/realtime/PipeRealtimeCollectEvent.java | 5 +-
.../realtime/PipeRealtimeCollectEventFactory.java | 7 +--
.../executor/PipeConnectorSubtaskExecutor.java | 2 +-
.../executor/PipeProcessorSubtaskExecutor.java | 2 +-
.../processor/PipeDoNothingProcessor.java | 4 +-
.../BlockingPendingQueue.java | 2 +-
.../BoundedBlockingPendingQueue.java | 2 +-
.../task/{queue => connection}/EventSupplier.java | 2 +-
.../connection}/PipeEventCollector.java | 5 +-
.../UnboundedBlockingPendingQueue.java | 2 +-
.../db/pipe/task/stage/PipeTaskCollectorStage.java | 6 +--
.../db/pipe/task/stage/PipeTaskConnectorStage.java | 4 +-
.../db/pipe/task/stage/PipeTaskProcessorStage.java | 10 ++--
.../db/pipe/task/subtask/PipeConnectorSubtask.java | 4 +-
.../subtask}/PipeConnectorSubtaskLifeCycle.java | 5 +-
.../subtask}/PipeConnectorSubtaskManager.java | 10 ++--
.../db/pipe/task/subtask/PipeProcessorSubtask.java | 2 +-
.../iotdb/db/pipe/task/subtask/PipeSubtask.java | 2 +-
.../collector/CachedSchemaPatternMatcherTest.java | 8 ++--
.../collector/PipeRealtimeCollectTest.java | 10 ++--
.../connector/PipeThriftRequestTest.java | 12 ++---
.../event/PipeTabletInsertionEventTest.java | 10 ++--
.../event/TsFileInsertionDataContainerTest.java | 6 +--
.../PipeConnectorSubtaskExecutorTest.java | 5 +-
.../PipeProcessorSubtaskExecutorTest.java | 5 +-
.../{executor => }/PipeSubtaskExecutorTest.java | 3 +-
90 files changed, 279 insertions(+), 571 deletions(-)
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeResp.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeResp.java
deleted file mode 100644
index 26bd8b3290f..00000000000
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeResp.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.confignode.consensus.response.pipe;
-
-import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.commons.sync.pipe.PipeInfo;
-import org.apache.iotdb.consensus.common.DataSet;
-
-import java.util.List;
-
-public class PipeResp implements DataSet {
-
- TSStatus status;
- List<PipeInfo> pipeInfoList;
-
- public TSStatus getStatus() {
- return status;
- }
-
- public void setStatus(TSStatus status) {
- this.status = status;
- }
-
- public List<PipeInfo> getPipeInfoList() {
- return pipeInfoList;
- }
-
- public void setPipeInfoList(List<PipeInfo> pipeInfoList) {
- this.pipeInfoList = pipeInfoList;
- }
-}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeSinkResp.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeSinkResp.java
deleted file mode 100644
index 850165a923a..00000000000
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeSinkResp.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.confignode.consensus.response.pipe;
-
-import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.commons.sync.pipesink.PipeSink;
-import org.apache.iotdb.consensus.common.DataSet;
-
-import java.util.List;
-
-public class PipeSinkResp implements DataSet {
-
- TSStatus status;
- List<PipeSink> pipeSinkList;
-
- public TSStatus getStatus() {
- return status;
- }
-
- public void setStatus(TSStatus status) {
- this.status = status;
- }
-
- public List<PipeSink> getPipeSinkList() {
- return pipeSinkList;
- }
-
- public void setPipeSinkList(List<PipeSink> pipeSinkList) {
- this.pipeSinkList = pipeSinkList;
- }
-}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipePluginInfo.java b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipePluginInfo.java
index f839f6251dc..22f0cb97943 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipePluginInfo.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipePluginInfo.java
@@ -38,7 +38,7 @@ import org.apache.iotdb.db.pipe.config.PipeCollectorConstant;
import org.apache.iotdb.db.pipe.config.PipeConnectorConstant;
import org.apache.iotdb.db.pipe.config.PipeProcessorConstant;
import org.apache.iotdb.pipe.api.customizer.PipeParameters;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.TSStatusCode;
import org.slf4j.Logger;
@@ -93,14 +93,14 @@ public class PipePluginInfo implements SnapshotProcessor {
public void validateBeforeCreatingPipePlugin(String pluginName, String jarName, String jarMD5) {
// both build-in and user defined pipe plugin should be unique
if (pipePluginMetaKeeper.containsPipePlugin(pluginName)) {
- throw new PipeManagementException(
+ throw new PipeException(
String.format(
"Failed to create PipePlugin [%s], the same name PipePlugin has been created",
pluginName));
}
if (pipePluginMetaKeeper.jarNameExistsAndMatchesMd5(jarName, jarMD5)) {
- throw new PipeManagementException(
+ throw new PipeException(
String.format(
"Failed to create PipePlugin [%s], the same name Jar [%s] but different MD5 [%s] has existed",
pluginName, jarName, jarMD5));
@@ -110,7 +110,7 @@ public class PipePluginInfo implements SnapshotProcessor {
public void validateBeforeDroppingPipePlugin(String pluginName) {
if (pipePluginMetaKeeper.containsPipePlugin(pluginName)
&& pipePluginMetaKeeper.getPipePluginMeta(pluginName).isBuiltin()) {
- throw new PipeManagementException(
+ throw new PipeException(
String.format(
"Failed to drop PipePlugin [%s], the PipePlugin is a built-in PipePlugin",
pluginName));
@@ -133,7 +133,7 @@ public class PipePluginInfo implements SnapshotProcessor {
"Failed to create pipe, the pipe collector plugin %s does not exist",
collectorPluginName);
LOGGER.warn(exceptionMessage);
- throw new PipeManagementException(exceptionMessage);
+ throw new PipeException(exceptionMessage);
}
final PipeParameters processorParameters =
@@ -147,7 +147,7 @@ public class PipePluginInfo implements SnapshotProcessor {
"Failed to create pipe, the pipe processor plugin %s does not exist",
processorPluginName);
LOGGER.warn(exceptionMessage);
- throw new PipeManagementException(exceptionMessage);
+ throw new PipeException(exceptionMessage);
}
final PipeParameters connectorParameters =
@@ -156,7 +156,7 @@ public class PipePluginInfo implements SnapshotProcessor {
final String exceptionMessage =
"Failed to create pipe, the pipe connector plugin is not specified";
LOGGER.warn(exceptionMessage);
- throw new PipeManagementException(exceptionMessage);
+ throw new PipeException(exceptionMessage);
}
final String connectorPluginName =
connectorParameters.getString(PipeConnectorConstant.CONNECTOR_KEY);
@@ -166,7 +166,7 @@ public class PipePluginInfo implements SnapshotProcessor {
"Failed to create pipe, the pipe connector plugin %s does not exist",
connectorPluginName);
LOGGER.warn(exceptionMessage);
- throw new PipeManagementException(exceptionMessage);
+ throw new PipeException(exceptionMessage);
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
index 8aaa0948f80..d2580f87a1d 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
@@ -34,7 +34,7 @@ import org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStat
import org.apache.iotdb.confignode.consensus.response.pipe.task.PipeTableResp;
import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
import org.apache.iotdb.consensus.common.DataSet;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.TSStatusCode;
import org.slf4j.Logger;
@@ -74,8 +74,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
/////////////////////////////// Validator ///////////////////////////////
- public void checkBeforeCreatePipe(TCreatePipeReq createPipeRequest)
- throws PipeManagementException {
+ public void checkBeforeCreatePipe(TCreatePipeReq createPipeRequest) throws PipeException {
if (!isPipeExisted(createPipeRequest.getPipeName())) {
return;
}
@@ -85,15 +84,15 @@ public class PipeTaskInfo implements SnapshotProcessor {
"Failed to create pipe %s, the pipe with the same name has been created",
createPipeRequest.getPipeName());
LOGGER.info(exceptionMessage);
- throw new PipeManagementException(exceptionMessage);
+ throw new PipeException(exceptionMessage);
}
- public void checkBeforeStartPipe(String pipeName) throws PipeManagementException {
+ public void checkBeforeStartPipe(String pipeName) throws PipeException {
if (!isPipeExisted(pipeName)) {
final String exceptionMessage =
String.format("Failed to start pipe %s, the pipe does not exist", pipeName);
LOGGER.info(exceptionMessage);
- throw new PipeManagementException(exceptionMessage);
+ throw new PipeException(exceptionMessage);
}
final PipeStatus pipeStatus = getPipeStatus(pipeName);
@@ -101,22 +100,22 @@ public class PipeTaskInfo implements SnapshotProcessor {
final String exceptionMessage =
String.format("Failed to start pipe %s, the pipe is already running", pipeName);
LOGGER.info(exceptionMessage);
- throw new PipeManagementException(exceptionMessage);
+ throw new PipeException(exceptionMessage);
}
if (pipeStatus == PipeStatus.DROPPED) {
final String exceptionMessage =
String.format("Failed to start pipe %s, the pipe is already dropped", pipeName);
LOGGER.info(exceptionMessage);
- throw new PipeManagementException(exceptionMessage);
+ throw new PipeException(exceptionMessage);
}
}
- public void checkBeforeStopPipe(String pipeName) throws PipeManagementException {
+ public void checkBeforeStopPipe(String pipeName) throws PipeException {
if (!isPipeExisted(pipeName)) {
final String exceptionMessage =
String.format("Failed to stop pipe %s, the pipe does not exist", pipeName);
LOGGER.info(exceptionMessage);
- throw new PipeManagementException(exceptionMessage);
+ throw new PipeException(exceptionMessage);
}
final PipeStatus pipeStatus = getPipeStatus(pipeName);
@@ -124,13 +123,13 @@ public class PipeTaskInfo implements SnapshotProcessor {
final String exceptionMessage =
String.format("Failed to stop pipe %s, the pipe is already stop", pipeName);
LOGGER.info(exceptionMessage);
- throw new PipeManagementException(exceptionMessage);
+ throw new PipeException(exceptionMessage);
}
if (pipeStatus == PipeStatus.DROPPED) {
final String exceptionMessage =
String.format("Failed to stop pipe %s, the pipe is already dropped", pipeName);
LOGGER.info(exceptionMessage);
- throw new PipeManagementException(exceptionMessage);
+ throw new PipeException(exceptionMessage);
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AbstractOperatePipeProcedureV2.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
similarity index 95%
rename from confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AbstractOperatePipeProcedureV2.java
rename to confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
index 133a74854be..200a71e8f6f 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AbstractOperatePipeProcedureV2.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
@@ -16,18 +16,16 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.confignode.procedure.impl.pipe.task;
+package org.apache.iotdb.confignode.procedure.impl.pipe;
-import org.apache.iotdb.commons.exception.sync.PipeException;
import org.apache.iotdb.commons.pipe.task.meta.PipeMeta;
-import org.apache.iotdb.confignode.persistence.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedException;
import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
import org.apache.iotdb.confignode.procedure.impl.node.AbstractNodeProcedure;
import org.apache.iotdb.confignode.procedure.state.pipe.task.OperatePipeTaskState;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
@@ -196,7 +194,7 @@ public abstract class AbstractOperatePipeProcedureV2
if (RpcUtils.squashResponseStatusList(env.pushPipeMetaToDataNodes(pipeMetaBinaryList)).getCode()
!= TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- throw new PipeManagementException("Failed to push pipe meta list to data nodes");
+ throw new PipeException("Failed to push pipe meta list to data nodes");
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskOperation.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/PipeTaskOperation.java
similarity index 94%
rename from confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskOperation.java
rename to confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/PipeTaskOperation.java
index 6531f84e049..74a31ac223f 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskOperation.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/PipeTaskOperation.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.confignode.persistence.pipe;
+package org.apache.iotdb.confignode.procedure.impl.pipe;
public enum PipeTaskOperation {
CREATE_PIPE,
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/CreatePipePluginProcedure.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/CreatePipePluginProcedure.java
index 88f40bdc114..cdefb0501b2 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/CreatePipePluginProcedure.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/CreatePipePluginProcedure.java
@@ -36,7 +36,7 @@ import org.apache.iotdb.confignode.procedure.impl.node.RemoveDataNodeProcedure;
import org.apache.iotdb.confignode.procedure.state.pipe.plugin.CreatePipePluginState;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.utils.Binary;
@@ -128,7 +128,7 @@ public class CreatePipePluginProcedure extends AbstractNodeProcedure<CreatePipeP
pipePluginMeta.getPluginName(),
pipePluginMeta.getJarName(),
pipePluginMeta.getJarMD5());
- } catch (PipeManagementException e) {
+ } catch (PipeException e) {
// The pipe plugin has already created, we should end the procedure
LOGGER.warn(
"Pipe plugin {} is already created, end the CreatePipePluginProcedure({})",
@@ -162,7 +162,7 @@ public class CreatePipePluginProcedure extends AbstractNodeProcedure<CreatePipeP
final ConsensusWriteResponse response =
configNodeManager.getConsensusManager().write(createPluginPlan);
if (!response.isSuccessful()) {
- throw new PipeManagementException(response.getErrorMessage());
+ throw new PipeException(response.getErrorMessage());
}
setNextState(CreatePipePluginState.CREATE_ON_DATA_NODES);
@@ -181,7 +181,7 @@ public class CreatePipePluginProcedure extends AbstractNodeProcedure<CreatePipeP
return Flow.HAS_MORE_STATE;
}
- throw new PipeManagementException(
+ throw new PipeException(
String.format(
"Failed to create pipe plugin instance [%s] on data nodes",
pipePluginMeta.getPluginName()));
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/DropPipePluginProcedure.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/DropPipePluginProcedure.java
index d44df9ac580..5b469725038 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/DropPipePluginProcedure.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/plugin/DropPipePluginProcedure.java
@@ -31,7 +31,7 @@ import org.apache.iotdb.confignode.procedure.impl.node.RemoveConfigNodeProcedure
import org.apache.iotdb.confignode.procedure.impl.node.RemoveDataNodeProcedure;
import org.apache.iotdb.confignode.procedure.state.pipe.plugin.DropPipePluginState;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
@@ -109,7 +109,7 @@ public class DropPipePluginProcedure extends AbstractNodeProcedure<DropPipePlugi
try {
pipePluginCoordinator.getPipePluginInfo().validateBeforeDroppingPipePlugin(pluginName);
- } catch (PipeManagementException e) {
+ } catch (PipeException e) {
// if the pipe plugin is a built-in plugin, we should not drop it
LOGGER.warn(e.getMessage());
setFailure(new ProcedureException(e.getMessage()));
@@ -133,7 +133,7 @@ public class DropPipePluginProcedure extends AbstractNodeProcedure<DropPipePlugi
return Flow.HAS_MORE_STATE;
}
- throw new PipeManagementException(
+ throw new PipeException(
String.format("Failed to drop pipe plugin %s on data nodes", pluginName));
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleLeaderChangeProcedure.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleLeaderChangeProcedure.java
index 3cbcebe6f6b..c07982381cd 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleLeaderChangeProcedure.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleLeaderChangeProcedure.java
@@ -22,12 +22,12 @@ package org.apache.iotdb.confignode.procedure.impl.pipe.runtime;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.confignode.consensus.request.write.pipe.runtime.PipeHandleLeaderChangePlan;
-import org.apache.iotdb.confignode.persistence.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
-import org.apache.iotdb.confignode.procedure.impl.pipe.task.AbstractOperatePipeProcedureV2;
+import org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2;
+import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
@@ -95,7 +95,7 @@ public class PipeHandleLeaderChangeProcedure extends AbstractOperatePipeProcedur
final ConsensusWriteResponse response =
env.getConfigManager().getConsensusManager().write(pipeHandleLeaderChangePlan);
if (!response.isSuccessful()) {
- throw new PipeManagementException(response.getErrorMessage());
+ throw new PipeException(response.getErrorMessage());
}
}
@@ -137,7 +137,7 @@ public class PipeHandleLeaderChangeProcedure extends AbstractOperatePipeProcedur
final ConsensusWriteResponse response =
env.getConfigManager().getConsensusManager().write(pipeHandleLeaderChangePlan);
if (!response.isSuccessful()) {
- throw new PipeManagementException(response.getErrorMessage());
+ throw new PipeException(response.getErrorMessage());
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleMetaChangeProcedure.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleMetaChangeProcedure.java
index 39948e9fbb9..31ebb405907 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleMetaChangeProcedure.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleMetaChangeProcedure.java
@@ -29,12 +29,12 @@ import org.apache.iotdb.commons.pipe.task.meta.PipeStatus;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.apache.iotdb.confignode.consensus.request.write.pipe.runtime.PipeHandleMetaChangePlan;
import org.apache.iotdb.confignode.consensus.response.pipe.task.PipeTableResp;
-import org.apache.iotdb.confignode.persistence.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
-import org.apache.iotdb.confignode.procedure.impl.pipe.task.AbstractOperatePipeProcedureV2;
+import org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2;
+import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
@@ -233,7 +233,7 @@ public class PipeHandleMetaChangeProcedure extends AbstractOperatePipeProcedureV
.getConsensusManager()
.write(new PipeHandleMetaChangePlan(pipeMetaList));
if (!response.isSuccessful()) {
- throw new PipeManagementException(response.getErrorMessage());
+ throw new PipeException(response.getErrorMessage());
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
index a7faa989ab9..65ecdaae628 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
@@ -19,9 +19,9 @@
package org.apache.iotdb.confignode.procedure.impl.pipe.runtime;
-import org.apache.iotdb.confignode.persistence.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
-import org.apache.iotdb.confignode.procedure.impl.pipe.task.AbstractOperatePipeProcedureV2;
+import org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2;
+import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
import org.slf4j.Logger;
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
index b7ab5ef62d5..297b1c97a79 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
@@ -27,13 +27,13 @@ import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.apache.iotdb.confignode.consensus.request.write.pipe.task.CreatePipePlanV2;
import org.apache.iotdb.confignode.consensus.request.write.pipe.task.DropPipePlanV2;
import org.apache.iotdb.confignode.manager.pipe.PipeManager;
-import org.apache.iotdb.confignode.persistence.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2;
+import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
import org.apache.iotdb.pipe.api.exception.PipeException;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.slf4j.Logger;
@@ -70,8 +70,7 @@ public class CreatePipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
}
@Override
- protected void executeFromValidateTask(ConfigNodeProcedureEnv env)
- throws PipeManagementException {
+ protected void executeFromValidateTask(ConfigNodeProcedureEnv env) throws PipeException {
LOGGER.info(
"CreatePipeProcedureV2: executeFromValidateTask({})", createPipeRequest.getPipeName());
@@ -84,8 +83,7 @@ public class CreatePipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
}
@Override
- protected void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env)
- throws PipeManagementException {
+ protected void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) throws PipeException {
LOGGER.info(
"CreatePipeProcedureV2: executeFromCalculateInfoForTask({})",
createPipeRequest.getPipeName());
@@ -111,7 +109,7 @@ public class CreatePipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
@Override
protected void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env)
- throws PipeManagementException {
+ throws PipeException {
LOGGER.info(
"CreatePipeProcedureV2: executeFromWriteConfigNodeConsensus({})",
createPipeRequest.getPipeName());
@@ -121,13 +119,13 @@ public class CreatePipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
.getConsensusManager()
.write(new CreatePipePlanV2(pipeStaticMeta, pipeRuntimeMeta));
if (!response.isSuccessful()) {
- throw new PipeManagementException(response.getErrorMessage());
+ throw new PipeException(response.getErrorMessage());
}
}
@Override
protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
- throws PipeManagementException, IOException {
+ throws PipeException, IOException {
LOGGER.info(
"CreatePipeProcedureV2: executeFromOperateOnDataNodes({})",
createPipeRequest.getPipeName());
@@ -161,7 +159,7 @@ public class CreatePipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
.getConsensusManager()
.write(new DropPipePlanV2(createPipeRequest.getPipeName()));
if (!response.isSuccessful()) {
- throw new PipeManagementException(response.getErrorMessage());
+ throw new PipeException(response.getErrorMessage());
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/DropPipeProcedureV2.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/DropPipeProcedureV2.java
index 5a715ae3534..bae85b9bf74 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/DropPipeProcedureV2.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/DropPipeProcedureV2.java
@@ -19,12 +19,12 @@
package org.apache.iotdb.confignode.procedure.impl.pipe.task;
import org.apache.iotdb.confignode.consensus.request.write.pipe.task.DropPipePlanV2;
-import org.apache.iotdb.confignode.persistence.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2;
+import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
import org.apache.iotdb.pipe.api.exception.PipeException;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.slf4j.Logger;
@@ -55,8 +55,7 @@ public class DropPipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
}
@Override
- protected void executeFromValidateTask(ConfigNodeProcedureEnv env)
- throws PipeManagementException {
+ protected void executeFromValidateTask(ConfigNodeProcedureEnv env) throws PipeException {
LOGGER.info("DropPipeProcedureV2: executeFromValidateTask({})", pipeName);
env.getConfigManager()
@@ -67,27 +66,26 @@ public class DropPipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
}
@Override
- protected void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env)
- throws PipeManagementException {
+ protected void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) throws PipeException {
LOGGER.info("DropPipeProcedureV2: executeFromCalculateInfoForTask({})", pipeName);
// Do nothing
}
@Override
protected void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env)
- throws PipeManagementException {
+ throws PipeException {
LOGGER.info("DropPipeProcedureV2: executeFromWriteConfigNodeConsensus({})", pipeName);
final ConsensusWriteResponse response =
env.getConfigManager().getConsensusManager().write(new DropPipePlanV2(pipeName));
if (!response.isSuccessful()) {
- throw new PipeManagementException(response.getErrorMessage());
+ throw new PipeException(response.getErrorMessage());
}
}
@Override
protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
- throws PipeManagementException, IOException {
+ throws PipeException, IOException {
LOGGER.info("DropPipeProcedureV2: executeFromOperateOnDataNodes({})", pipeName);
pushPipeMetaToDataNodes(env);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java
index ecd4cdc57c0..ef277a4fcc8 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java
@@ -20,12 +20,12 @@ package org.apache.iotdb.confignode.procedure.impl.pipe.task;
import org.apache.iotdb.commons.pipe.task.meta.PipeStatus;
import org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusPlanV2;
-import org.apache.iotdb.confignode.persistence.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2;
+import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
import org.apache.iotdb.pipe.api.exception.PipeException;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.slf4j.Logger;
@@ -56,8 +56,7 @@ public class StartPipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
}
@Override
- protected void executeFromValidateTask(ConfigNodeProcedureEnv env)
- throws PipeManagementException {
+ protected void executeFromValidateTask(ConfigNodeProcedureEnv env) throws PipeException {
LOGGER.info("StartPipeProcedureV2: executeFromValidateTask({})", pipeName);
env.getConfigManager()
@@ -68,15 +67,14 @@ public class StartPipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
}
@Override
- protected void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env)
- throws PipeManagementException {
+ protected void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) throws PipeException {
LOGGER.info("StartPipeProcedureV2: executeFromCalculateInfoForTask({})", pipeName);
// Do nothing
}
@Override
protected void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env)
- throws PipeManagementException {
+ throws PipeException {
LOGGER.info("StartPipeProcedureV2: executeFromWriteConfigNodeConsensus({})", pipeName);
final ConsensusWriteResponse response =
@@ -84,13 +82,13 @@ public class StartPipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
.getConsensusManager()
.write(new SetPipeStatusPlanV2(pipeName, PipeStatus.RUNNING));
if (!response.isSuccessful()) {
- throw new PipeManagementException(response.getErrorMessage());
+ throw new PipeException(response.getErrorMessage());
}
}
@Override
protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
- throws PipeManagementException, IOException {
+ throws PipeException, IOException {
LOGGER.info("StartPipeProcedureV2: executeFromOperateOnDataNodes({})", pipeName);
pushPipeMetaToDataNodes(env);
@@ -117,13 +115,13 @@ public class StartPipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
.getConsensusManager()
.write(new SetPipeStatusPlanV2(pipeName, PipeStatus.STOPPED));
if (!response.isSuccessful()) {
- throw new PipeManagementException(response.getErrorMessage());
+ throw new PipeException(response.getErrorMessage());
}
}
@Override
protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
- throws PipeManagementException, IOException {
+ throws PipeException, IOException {
LOGGER.info("StartPipeProcedureV2: rollbackFromOperateOnDataNodes({})", pipeName);
pushPipeMetaToDataNodes(env);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StopPipeProcedureV2.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StopPipeProcedureV2.java
index fcd015fb940..7c98678a608 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StopPipeProcedureV2.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StopPipeProcedureV2.java
@@ -20,12 +20,12 @@ package org.apache.iotdb.confignode.procedure.impl.pipe.task;
import org.apache.iotdb.commons.pipe.task.meta.PipeStatus;
import org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusPlanV2;
-import org.apache.iotdb.confignode.persistence.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2;
+import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
import org.apache.iotdb.pipe.api.exception.PipeException;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.slf4j.Logger;
@@ -56,8 +56,7 @@ public class StopPipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
}
@Override
- protected void executeFromValidateTask(ConfigNodeProcedureEnv env)
- throws PipeManagementException {
+ protected void executeFromValidateTask(ConfigNodeProcedureEnv env) throws PipeException {
LOGGER.info("StopPipeProcedureV2: executeFromValidateTask({})", pipeName);
env.getConfigManager()
@@ -68,15 +67,14 @@ public class StopPipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
}
@Override
- protected void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env)
- throws PipeManagementException {
+ protected void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) throws PipeException {
LOGGER.info("StopPipeProcedureV2: executeFromCalculateInfoForTask({})", pipeName);
// Do nothing
}
@Override
protected void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env)
- throws PipeManagementException {
+ throws PipeException {
LOGGER.info("StopPipeProcedureV2: executeFromWriteConfigNodeConsensus({})", pipeName);
final ConsensusWriteResponse response =
@@ -84,13 +82,13 @@ public class StopPipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
.getConsensusManager()
.write(new SetPipeStatusPlanV2(pipeName, PipeStatus.STOPPED));
if (!response.isSuccessful()) {
- throw new PipeManagementException(response.getErrorMessage());
+ throw new PipeException(response.getErrorMessage());
}
}
@Override
protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
- throws PipeManagementException, IOException {
+ throws PipeException, IOException {
LOGGER.info("StopPipeProcedureV2: executeFromOperateOnDataNodes({})", pipeName);
pushPipeMetaToDataNodes(env);
@@ -117,13 +115,13 @@ public class StopPipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
.getConsensusManager()
.write(new SetPipeStatusPlanV2(pipeName, PipeStatus.RUNNING));
if (!response.isSuccessful()) {
- throw new PipeManagementException(response.getErrorMessage());
+ throw new PipeException(response.getErrorMessage());
}
}
@Override
protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
- throws PipeManagementException, IOException {
+ throws PipeException, IOException {
LOGGER.info("StopPipeProcedureV2: rollbackFromOperateOnDataNodes({})", pipeName);
pushPipeMetaToDataNodes(env);
diff --git a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/EventType.java b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/EventType.java
deleted file mode 100644
index 20be75367d5..00000000000
--- a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/EventType.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * 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.pipe.api.event;
-
-public enum EventType {
- TABLET_INSERTION,
- TSFILE_INSERTION,
- DELETION,
-}
diff --git a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/dml/insertion/TsFileInsertionEvent.java b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/dml/insertion/TsFileInsertionEvent.java
index 325f0683cb7..a3e44fd55ed 100644
--- a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/dml/insertion/TsFileInsertionEvent.java
+++ b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/dml/insertion/TsFileInsertionEvent.java
@@ -33,12 +33,4 @@ public interface TsFileInsertionEvent extends Event {
* @return the list of TsFileInsertionEvent
*/
Iterable<TabletInsertionEvent> toTabletInsertionEvents();
-
- /**
- * The method is used to compact several TabletInsertionEvents into one TsFileInsertionEvent. The
- * underlying data in TabletInsertionEvents will be stored into a TsFile.
- *
- * @return TsFileInsertionEvent
- */
- TsFileInsertionEvent toTsFileInsertionEvent(Iterable<TabletInsertionEvent> iterable);
}
diff --git a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/exception/PipeManagementException.java b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/exception/PipeManagementException.java
deleted file mode 100644
index 5ec3b2039b0..00000000000
--- a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/exception/PipeManagementException.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * 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.pipe.api.exception;
-
-public class PipeManagementException extends PipeException {
-
- public PipeManagementException(String message, Throwable cause) {
- super(message);
- this.initCause(cause);
- }
-
- public PipeManagementException(String message) {
- super(message);
- }
-}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/service/PipePluginExecutableManager.java b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/service/PipePluginExecutableManager.java
index 843df30c6ba..5d9d7da27e5 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/service/PipePluginExecutableManager.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/plugin/service/PipePluginExecutableManager.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.commons.pipe.plugin.service;
import org.apache.iotdb.commons.executable.ExecutableManager;
import org.apache.iotdb.commons.file.SystemFileFactory;
import org.apache.iotdb.commons.pipe.plugin.meta.PipePluginMeta;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.commons.codec.digest.DigestUtils;
import org.slf4j.Logger;
@@ -41,7 +41,7 @@ public class PipePluginExecutableManager extends ExecutableManager {
super(temporaryLibRoot, libRoot);
}
- public boolean isLocalJarMatched(PipePluginMeta pipePluginMeta) throws PipeManagementException {
+ public boolean isLocalJarMatched(PipePluginMeta pipePluginMeta) throws PipeException {
final String pluginName = pipePluginMeta.getPluginName();
final String md5FilePath = pluginName + ".txt";
@@ -69,7 +69,7 @@ public class PipePluginExecutableManager extends ExecutableManager {
+ "because error occurred when trying to compute md5 of jar file for function %s ",
pluginName, pluginName);
LOGGER.warn(errorMessage, e);
- throw new PipeManagementException(errorMessage);
+ throw new PipeException(errorMessage);
}
}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/utils/PipeBinaryTransformer.java b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/utils/PipeBinaryTransformer.java
deleted file mode 100644
index 125de0e9206..00000000000
--- a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/utils/PipeBinaryTransformer.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * 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.commons.pipe.utils;
-
-public class PipeBinaryTransformer {
- public static org.apache.iotdb.tsfile.utils.Binary transformToBinary(
- org.apache.iotdb.pipe.api.type.Binary binary) {
- return binary == null ? null : new org.apache.iotdb.tsfile.utils.Binary(binary.getValues());
- }
-
- public static org.apache.iotdb.pipe.api.type.Binary transformToPipeBinary(
- org.apache.iotdb.tsfile.utils.Binary binary) {
- return binary == null ? null : new org.apache.iotdb.pipe.api.type.Binary(binary.getValues());
- }
-}
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/meta/PipeMetaDeSerTest.java b/node-commons/src/test/java/org/apache/iotdb/commons/pipe/PipeMetaDeSerTest.java
similarity index 91%
rename from node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/meta/PipeMetaDeSerTest.java
rename to node-commons/src/test/java/org/apache/iotdb/commons/pipe/PipeMetaDeSerTest.java
index b9fc0ef7e5d..c5813959938 100644
--- a/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/meta/PipeMetaDeSerTest.java
+++ b/node-commons/src/test/java/org/apache/iotdb/commons/pipe/PipeMetaDeSerTest.java
@@ -17,11 +17,15 @@
* under the License.
*/
-package org.apache.iotdb.commons.pipe.task.meta;
+package org.apache.iotdb.commons.pipe;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
+import org.apache.iotdb.commons.pipe.task.meta.PipeMeta;
+import org.apache.iotdb.commons.pipe.task.meta.PipeRuntimeMeta;
+import org.apache.iotdb.commons.pipe.task.meta.PipeStaticMeta;
+import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.junit.Assert;
import org.junit.Test;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index cbc609e3605..656e30b40f1 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -53,7 +53,7 @@ import org.apache.iotdb.db.mpp.metric.QueryResourceMetricSet;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode;
-import org.apache.iotdb.db.pipe.core.collector.realtime.listener.PipeInsertionDataNodeListener;
+import org.apache.iotdb.db.pipe.collector.realtime.listener.PipeInsertionDataNodeListener;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.rescon.MemTableManager;
import org.apache.iotdb.db.rescon.PrimitiveArrayManager;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginAgent.java b/server/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginAgent.java
index b0fe93b4533..454cf5c1788 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginAgent.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipePluginAgent.java
@@ -33,7 +33,7 @@ import org.apache.iotdb.pipe.api.PipeConnector;
import org.apache.iotdb.pipe.api.PipePlugin;
import org.apache.iotdb.pipe.api.PipeProcessor;
import org.apache.iotdb.pipe.api.customizer.PipeParameters;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -82,7 +82,7 @@ public class PipePluginAgent {
}
}
- private void checkIfRegistered(PipePluginMeta pipePluginMeta) throws PipeManagementException {
+ private void checkIfRegistered(PipePluginMeta pipePluginMeta) throws PipeException {
final String pluginName = pipePluginMeta.getPluginName();
final PipePluginMeta information = pipePluginMetaKeeper.getPipePluginMeta(pluginName);
if (information == null) {
@@ -95,7 +95,7 @@ public class PipePluginAgent {
"Failed to register PipePlugin %s, because the given PipePlugin name is the same as a built-in PipePlugin name.",
pluginName);
LOGGER.warn(errorMessage);
- throw new PipeManagementException(errorMessage);
+ throw new PipeException(errorMessage);
}
if (PipePluginExecutableManager.getInstance()
@@ -107,7 +107,7 @@ public class PipePluginAgent {
+ "because existed md5 of jar file for pipe plugin %s is different from the new jar file.",
pluginName, pluginName);
LOGGER.warn(errMsg);
- throw new PipeManagementException(errMsg);
+ throw new PipeException(errMsg);
}
// if the pipe plugin is already registered and the jar file is the same, do nothing
@@ -125,10 +125,9 @@ public class PipePluginAgent {
* the PipePluginClassLoader and its instance will be created to ensure that it can be loaded.
*
* @param pipePluginMeta the meta information of the PipePlugin
- * @throws PipeManagementException if the PipePlugin can not be loaded or its instance can not be
- * created
+ * @throws PipeException if the PipePlugin can not be loaded or its instance can not be created
*/
- public void doRegister(PipePluginMeta pipePluginMeta) throws PipeManagementException {
+ public void doRegister(PipePluginMeta pipePluginMeta) throws PipeException {
final String pluginName = pipePluginMeta.getPluginName();
final String className = pipePluginMeta.getClassName();
@@ -155,7 +154,7 @@ public class PipePluginAgent {
"Failed to register PipePlugin %s(%s), because its instance can not be constructed successfully. Exception: %s",
pluginName.toUpperCase(), className, e);
LOGGER.warn(errorMessage, e);
- throw new PipeManagementException(errorMessage);
+ throw new PipeException(errorMessage);
}
}
@@ -175,7 +174,7 @@ public class PipePluginAgent {
String errorMessage =
String.format("Failed to deregister builtin PipePlugin %s.", pluginName);
LOGGER.warn(errorMessage);
- throw new PipeManagementException(errorMessage);
+ throw new PipeException(errorMessage);
}
// remove anyway
@@ -211,7 +210,7 @@ public class PipePluginAgent {
public PipeConnector reflectConnector(PipeParameters connectorParameters) {
if (!connectorParameters.hasAttribute(PipeConnectorConstant.CONNECTOR_KEY)) {
- throw new PipeManagementException(
+ throw new PipeException(
"Failed to reflect PipeConnector instance because 'connector' is not specified in the parameters.");
}
return (PipeConnector)
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/IoTDBThriftReceiver.java b/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/IoTDBThriftReceiver.java
index 81d6fef23d0..4230ee73a34 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/IoTDBThriftReceiver.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/IoTDBThriftReceiver.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.pipe.agent.receiver;
import org.apache.iotdb.db.mpp.plan.analyze.IPartitionFetcher;
import org.apache.iotdb.db.mpp.plan.analyze.schema.ISchemaFetcher;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.IoTDBThriftConnectorVersion;
+import org.apache.iotdb.db.pipe.connector.IoTDBThriftConnectorVersion;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/PipeReceiverAgent.java b/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/PipeReceiverAgent.java
index ab48a9f6149..fdeda91ea56 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/PipeReceiverAgent.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/agent/receiver/PipeReceiverAgent.java
@@ -21,8 +21,8 @@ package org.apache.iotdb.db.pipe.agent.receiver;
import org.apache.iotdb.db.mpp.plan.analyze.IPartitionFetcher;
import org.apache.iotdb.db.mpp.plan.analyze.schema.ISchemaFetcher;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.IoTDBThriftConnectorVersion;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.IoTDBThriftReceiverV1;
+import org.apache.iotdb.db.pipe.connector.IoTDBThriftConnectorVersion;
+import org.apache.iotdb.db.pipe.connector.v1.IoTDBThriftReceiverV1;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeAgentLauncher.java b/server/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeAgentLauncher.java
index 16f3716bc13..69786989982 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeAgentLauncher.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeAgentLauncher.java
@@ -35,7 +35,7 @@ import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.service.ResourcesInformationHolder;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.thrift.TException;
@@ -122,7 +122,7 @@ class PipeAgentLauncher {
if (!PipePluginExecutableManager.getInstance().isLocalJarMatched(pipePluginMeta)) {
pipePluginMetaList.add(pipePluginMeta);
}
- } catch (PipeManagementException e) {
+ } catch (PipeException e) {
pipePluginMetaList.add(pipePluginMeta);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/IoTDBDataRegionCollector.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/IoTDBDataRegionCollector.java
similarity index 91%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/collector/IoTDBDataRegionCollector.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/IoTDBDataRegionCollector.java
index 757599f2b24..bb247981bf6 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/IoTDBDataRegionCollector.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/IoTDBDataRegionCollector.java
@@ -17,26 +17,26 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector;
+package org.apache.iotdb.db.pipe.collector;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.pipe.collector.historical.PipeHistoricalDataRegionCollector;
+import org.apache.iotdb.db.pipe.collector.historical.PipeHistoricalDataRegionTsFileCollector;
+import org.apache.iotdb.db.pipe.collector.realtime.PipeRealtimeDataRegionCollector;
+import org.apache.iotdb.db.pipe.collector.realtime.PipeRealtimeDataRegionFakeCollector;
+import org.apache.iotdb.db.pipe.collector.realtime.PipeRealtimeDataRegionHybridCollector;
+import org.apache.iotdb.db.pipe.collector.realtime.PipeRealtimeDataRegionLogCollector;
+import org.apache.iotdb.db.pipe.collector.realtime.PipeRealtimeDataRegionTsFileCollector;
import org.apache.iotdb.db.pipe.config.PipeCollectorConstant;
-import org.apache.iotdb.db.pipe.core.collector.historical.PipeHistoricalDataRegionCollector;
-import org.apache.iotdb.db.pipe.core.collector.historical.PipeHistoricalDataRegionTsFileCollector;
-import org.apache.iotdb.db.pipe.core.collector.realtime.PipeRealtimeDataRegionCollector;
-import org.apache.iotdb.db.pipe.core.collector.realtime.PipeRealtimeDataRegionFakeCollector;
-import org.apache.iotdb.db.pipe.core.collector.realtime.PipeRealtimeDataRegionHybridCollector;
-import org.apache.iotdb.db.pipe.core.collector.realtime.PipeRealtimeDataRegionLogCollector;
-import org.apache.iotdb.db.pipe.core.collector.realtime.PipeRealtimeDataRegionTsFileCollector;
-import org.apache.iotdb.db.pipe.task.queue.UnboundedBlockingPendingQueue;
+import org.apache.iotdb.db.pipe.task.connection.UnboundedBlockingPendingQueue;
import org.apache.iotdb.pipe.api.PipeCollector;
import org.apache.iotdb.pipe.api.customizer.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.PipeParameters;
import org.apache.iotdb.pipe.api.customizer.collector.PipeCollectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.event.Event;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -228,7 +228,7 @@ public class IoTDBDataRegionCollector implements PipeCollector {
private void rethrowExceptionIfAny(AtomicReference<Exception> exceptionHolder) {
if (exceptionHolder.get() != null) {
- throw new PipeManagementException("failed to start collectors.", exceptionHolder.get());
+ throw new PipeException("failed to start collectors.", exceptionHolder.get());
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/historical/PipeHistoricalDataRegionCollector.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/historical/PipeHistoricalDataRegionCollector.java
similarity index 94%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/collector/historical/PipeHistoricalDataRegionCollector.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/historical/PipeHistoricalDataRegionCollector.java
index b5eb71c8df0..beb96f8f26b 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/historical/PipeHistoricalDataRegionCollector.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/historical/PipeHistoricalDataRegionCollector.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector.historical;
+package org.apache.iotdb.db.pipe.collector.historical;
import org.apache.iotdb.pipe.api.PipeCollector;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/historical/PipeHistoricalDataRegionTsFileCollector.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/historical/PipeHistoricalDataRegionTsFileCollector.java
similarity index 98%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/collector/historical/PipeHistoricalDataRegionTsFileCollector.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/historical/PipeHistoricalDataRegionTsFileCollector.java
index 583cbd5d853..8a6f21603e9 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/historical/PipeHistoricalDataRegionTsFileCollector.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/historical/PipeHistoricalDataRegionTsFileCollector.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector.historical;
+package org.apache.iotdb.db.pipe.collector.historical;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
@@ -28,7 +28,7 @@ import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.pipe.config.PipeCollectorConstant;
-import org.apache.iotdb.db.pipe.core.event.impl.PipeTsFileInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.utils.DateTimeUtils;
import org.apache.iotdb.pipe.api.customizer.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.PipeParameters;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionCollector.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionCollector.java
similarity index 92%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionCollector.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionCollector.java
index 46d81b05ed0..658da872795 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionCollector.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionCollector.java
@@ -17,12 +17,12 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector.realtime;
+package org.apache.iotdb.db.pipe.collector.realtime;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
+import org.apache.iotdb.db.pipe.collector.realtime.listener.PipeInsertionDataNodeListener;
import org.apache.iotdb.db.pipe.config.PipeCollectorConstant;
-import org.apache.iotdb.db.pipe.core.collector.realtime.listener.PipeInsertionDataNodeListener;
-import org.apache.iotdb.db.pipe.core.event.realtime.PipeRealtimeCollectEvent;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeCollectEvent;
import org.apache.iotdb.pipe.api.PipeCollector;
import org.apache.iotdb.pipe.api.customizer.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.PipeParameters;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionFakeCollector.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionFakeCollector.java
similarity index 93%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionFakeCollector.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionFakeCollector.java
index a559472f290..bd0913132b5 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionFakeCollector.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionFakeCollector.java
@@ -17,10 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector.realtime;
+package org.apache.iotdb.db.pipe.collector.realtime;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
-import org.apache.iotdb.db.pipe.core.event.realtime.PipeRealtimeCollectEvent;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeCollectEvent;
import org.apache.iotdb.pipe.api.customizer.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.PipeParameters;
import org.apache.iotdb.pipe.api.customizer.collector.PipeCollectorRuntimeConfiguration;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionHybridCollector.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionHybridCollector.java
similarity index 96%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionHybridCollector.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionHybridCollector.java
index cf01a1ea15a..50349097d50 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionHybridCollector.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionHybridCollector.java
@@ -17,15 +17,15 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector.realtime;
+package org.apache.iotdb.db.pipe.collector.realtime;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeNonCriticalException;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
-import org.apache.iotdb.db.pipe.core.event.realtime.PipeRealtimeCollectEvent;
-import org.apache.iotdb.db.pipe.core.event.realtime.TsFileEpoch;
-import org.apache.iotdb.db.pipe.task.queue.UnboundedBlockingPendingQueue;
+import org.apache.iotdb.db.pipe.collector.realtime.epoch.TsFileEpoch;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeCollectEvent;
+import org.apache.iotdb.db.pipe.task.connection.UnboundedBlockingPendingQueue;
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.TsFileInsertionEvent;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionLogCollector.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionLogCollector.java
similarity index 93%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionLogCollector.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionLogCollector.java
index 3263e52c1e6..845637b5f85 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionLogCollector.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionLogCollector.java
@@ -17,14 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector.realtime;
+package org.apache.iotdb.db.pipe.collector.realtime;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeNonCriticalException;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
-import org.apache.iotdb.db.pipe.core.event.realtime.PipeRealtimeCollectEvent;
-import org.apache.iotdb.db.pipe.core.event.realtime.TsFileEpoch;
-import org.apache.iotdb.db.pipe.task.queue.UnboundedBlockingPendingQueue;
+import org.apache.iotdb.db.pipe.collector.realtime.epoch.TsFileEpoch;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeCollectEvent;
+import org.apache.iotdb.db.pipe.task.connection.UnboundedBlockingPendingQueue;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionTsFileCollector.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionTsFileCollector.java
similarity index 93%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionTsFileCollector.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionTsFileCollector.java
index 7b287b10545..da0c1fb90c3 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/PipeRealtimeDataRegionTsFileCollector.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/PipeRealtimeDataRegionTsFileCollector.java
@@ -17,14 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector.realtime;
+package org.apache.iotdb.db.pipe.collector.realtime;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeNonCriticalException;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
-import org.apache.iotdb.db.pipe.core.event.realtime.PipeRealtimeCollectEvent;
-import org.apache.iotdb.db.pipe.core.event.realtime.TsFileEpoch;
-import org.apache.iotdb.db.pipe.task.queue.UnboundedBlockingPendingQueue;
+import org.apache.iotdb.db.pipe.collector.realtime.epoch.TsFileEpoch;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeCollectEvent;
+import org.apache.iotdb.db.pipe.task.connection.UnboundedBlockingPendingQueue;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/assigner/DisruptorQueue.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/assigner/DisruptorQueue.java
similarity index 98%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/assigner/DisruptorQueue.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/assigner/DisruptorQueue.java
index f0dc0a461be..885913a44cf 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/assigner/DisruptorQueue.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/assigner/DisruptorQueue.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector.realtime.assigner;
+package org.apache.iotdb.db.pipe.collector.realtime.assigner;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/assigner/PipeDataRegionAssigner.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/assigner/PipeDataRegionAssigner.java
similarity index 87%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/assigner/PipeDataRegionAssigner.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/assigner/PipeDataRegionAssigner.java
index 4b7c0cf8dbf..fe482147e04 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/assigner/PipeDataRegionAssigner.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/assigner/PipeDataRegionAssigner.java
@@ -17,12 +17,12 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector.realtime.assigner;
+package org.apache.iotdb.db.pipe.collector.realtime.assigner;
-import org.apache.iotdb.db.pipe.core.collector.realtime.PipeRealtimeDataRegionCollector;
-import org.apache.iotdb.db.pipe.core.collector.realtime.matcher.CachedSchemaPatternMatcher;
-import org.apache.iotdb.db.pipe.core.collector.realtime.matcher.PipeDataRegionMatcher;
-import org.apache.iotdb.db.pipe.core.event.realtime.PipeRealtimeCollectEvent;
+import org.apache.iotdb.db.pipe.collector.realtime.PipeRealtimeDataRegionCollector;
+import org.apache.iotdb.db.pipe.collector.realtime.matcher.CachedSchemaPatternMatcher;
+import org.apache.iotdb.db.pipe.collector.realtime.matcher.PipeDataRegionMatcher;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeCollectEvent;
import com.lmax.disruptor.dsl.ProducerType;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/TsFileEpoch.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/epoch/TsFileEpoch.java
similarity index 93%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/TsFileEpoch.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/epoch/TsFileEpoch.java
index b6c71f45d4e..f20bf4cc6dc 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/TsFileEpoch.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/epoch/TsFileEpoch.java
@@ -17,9 +17,9 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event.realtime;
+package org.apache.iotdb.db.pipe.collector.realtime.epoch;
-import org.apache.iotdb.db.pipe.core.collector.realtime.PipeRealtimeDataRegionCollector;
+import org.apache.iotdb.db.pipe.collector.realtime.PipeRealtimeDataRegionCollector;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/TsFileEpochManager.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/epoch/TsFileEpochManager.java
similarity index 89%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/TsFileEpochManager.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/epoch/TsFileEpochManager.java
index 5a33649aad2..095bac51c2b 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/TsFileEpochManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/epoch/TsFileEpochManager.java
@@ -17,12 +17,13 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event.realtime;
+package org.apache.iotdb.db.pipe.collector.realtime.epoch;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertNode;
-import org.apache.iotdb.db.pipe.core.event.impl.PipeInsertNodeTabletInsertionEvent;
-import org.apache.iotdb.db.pipe.core.event.impl.PipeTsFileInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeCollectEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/TsFileEpochStateMigrator.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/epoch/TsFileEpochStateMigrator.java
similarity index 94%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/TsFileEpochStateMigrator.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/epoch/TsFileEpochStateMigrator.java
index f75f7c87e5d..b95d3349b47 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/TsFileEpochStateMigrator.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/epoch/TsFileEpochStateMigrator.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event.realtime;
+package org.apache.iotdb.db.pipe.collector.realtime.epoch;
@FunctionalInterface
public interface TsFileEpochStateMigrator {
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/listener/PipeInsertionDataNodeListener.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/listener/PipeInsertionDataNodeListener.java
similarity index 94%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/listener/PipeInsertionDataNodeListener.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/listener/PipeInsertionDataNodeListener.java
index ae9cce9a498..6fd82618ff9 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/listener/PipeInsertionDataNodeListener.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/listener/PipeInsertionDataNodeListener.java
@@ -17,14 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector.realtime.listener;
+package org.apache.iotdb.db.pipe.collector.realtime.listener;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
-import org.apache.iotdb.db.pipe.core.collector.realtime.PipeRealtimeDataRegionCollector;
-import org.apache.iotdb.db.pipe.core.collector.realtime.assigner.PipeDataRegionAssigner;
-import org.apache.iotdb.db.pipe.core.event.realtime.PipeRealtimeCollectEventFactory;
+import org.apache.iotdb.db.pipe.collector.realtime.PipeRealtimeDataRegionCollector;
+import org.apache.iotdb.db.pipe.collector.realtime.assigner.PipeDataRegionAssigner;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeCollectEventFactory;
import org.apache.iotdb.db.wal.utils.WALEntryHandler;
import java.util.concurrent.ConcurrentHashMap;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/matcher/CachedSchemaPatternMatcher.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/matcher/CachedSchemaPatternMatcher.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/matcher/CachedSchemaPatternMatcher.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/matcher/CachedSchemaPatternMatcher.java
index b68a7e99c0f..cc3dd94a67c 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/matcher/CachedSchemaPatternMatcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/matcher/CachedSchemaPatternMatcher.java
@@ -17,11 +17,11 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector.realtime.matcher;
+package org.apache.iotdb.db.pipe.collector.realtime.matcher;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
-import org.apache.iotdb.db.pipe.core.collector.realtime.PipeRealtimeDataRegionCollector;
-import org.apache.iotdb.db.pipe.core.event.realtime.PipeRealtimeCollectEvent;
+import org.apache.iotdb.db.pipe.collector.realtime.PipeRealtimeDataRegionCollector;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeCollectEvent;
import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
import com.github.benmanes.caffeine.cache.Cache;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/matcher/PipeDataRegionMatcher.java b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/matcher/PipeDataRegionMatcher.java
similarity index 88%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/matcher/PipeDataRegionMatcher.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/matcher/PipeDataRegionMatcher.java
index 6411d7cead1..6e84cdf0ce2 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/realtime/matcher/PipeDataRegionMatcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/collector/realtime/matcher/PipeDataRegionMatcher.java
@@ -17,10 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector.realtime.matcher;
+package org.apache.iotdb.db.pipe.collector.realtime.matcher;
-import org.apache.iotdb.db.pipe.core.collector.realtime.PipeRealtimeDataRegionCollector;
-import org.apache.iotdb.db.pipe.core.event.realtime.PipeRealtimeCollectEvent;
+import org.apache.iotdb.db.pipe.collector.realtime.PipeRealtimeDataRegionCollector;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeCollectEvent;
import java.util.Set;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/IoTDBThriftConnectorClient.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/IoTDBThriftConnectorClient.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/IoTDBThriftConnectorClient.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/connector/IoTDBThriftConnectorClient.java
index 16ebfb2d74d..db0047372db 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/IoTDBThriftConnectorClient.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/IoTDBThriftConnectorClient.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.connector.impl.iotdb;
+package org.apache.iotdb.db.pipe.connector;
import org.apache.iotdb.commons.client.ThriftClient;
import org.apache.iotdb.commons.client.property.ThriftClientProperty;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/IoTDBThriftConnectorVersion.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/IoTDBThriftConnectorVersion.java
similarity index 94%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/IoTDBThriftConnectorVersion.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/connector/IoTDBThriftConnectorVersion.java
index d05fd374544..7dac858ae3f 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/IoTDBThriftConnectorVersion.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/IoTDBThriftConnectorVersion.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.connector.impl.iotdb;
+package org.apache.iotdb.db.pipe.connector;
public enum IoTDBThriftConnectorVersion {
VERSION_ONE((byte) 1),
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/IoTDBThriftConnectorV1.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
similarity index 91%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/IoTDBThriftConnectorV1.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
index 4c0b1c0e54e..04719ad2f44 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/IoTDBThriftConnectorV1.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftConnectorV1.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1;
+package org.apache.iotdb.db.pipe.connector.v1;
import org.apache.iotdb.commons.client.property.ThriftClientProperty;
import org.apache.iotdb.commons.conf.CommonConfig;
@@ -26,16 +26,16 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.config.PipeConnectorConstant;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.IoTDBThriftConnectorClient;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.reponse.PipeTransferFilePieceResp;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request.PipeTransferFilePieceReq;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request.PipeTransferFileSealReq;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request.PipeTransferHandshakeReq;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request.PipeTransferInsertNodeReq;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request.PipeTransferTabletReq;
-import org.apache.iotdb.db.pipe.core.event.impl.PipeInsertNodeTabletInsertionEvent;
-import org.apache.iotdb.db.pipe.core.event.impl.PipeRawTabletInsertionEvent;
-import org.apache.iotdb.db.pipe.core.event.impl.PipeTsFileInsertionEvent;
+import org.apache.iotdb.db.pipe.connector.IoTDBThriftConnectorClient;
+import org.apache.iotdb.db.pipe.connector.v1.reponse.PipeTransferFilePieceResp;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferFilePieceReq;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferFileSealReq;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferHandshakeReq;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferInsertNodeReq;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferTabletReq;
+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.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.wal.exception.WALPipeException;
import org.apache.iotdb.pipe.api.PipeConnector;
import org.apache.iotdb.pipe.api.customizer.PipeParameterValidator;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/IoTDBThriftReceiverV1.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftReceiverV1.java
similarity index 94%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/IoTDBThriftReceiverV1.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftReceiverV1.java
index b7fcb594c20..dfd6765b68d 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/IoTDBThriftReceiverV1.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftReceiverV1.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1;
+package org.apache.iotdb.db.pipe.connector.v1;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.db.conf.IoTDBConfig;
@@ -30,13 +30,13 @@ import org.apache.iotdb.db.mpp.plan.statement.Statement;
import org.apache.iotdb.db.mpp.plan.statement.crud.InsertTabletStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.LoadTsFileStatement;
import org.apache.iotdb.db.pipe.agent.receiver.IoTDBThriftReceiver;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.IoTDBThriftConnectorVersion;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.reponse.PipeTransferFilePieceResp;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request.PipeTransferFilePieceReq;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request.PipeTransferFileSealReq;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request.PipeTransferHandshakeReq;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request.PipeTransferInsertNodeReq;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request.PipeTransferTabletReq;
+import org.apache.iotdb.db.pipe.connector.IoTDBThriftConnectorVersion;
+import org.apache.iotdb.db.pipe.connector.v1.reponse.PipeTransferFilePieceResp;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferFilePieceReq;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferFileSealReq;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferHandshakeReq;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferInsertNodeReq;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferTabletReq;
import org.apache.iotdb.db.query.control.SessionManager;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/PipeRequestType.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/PipeRequestType.java
similarity index 96%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/PipeRequestType.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/PipeRequestType.java
index c62034353d0..cdbb5a59eef 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/PipeRequestType.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/PipeRequestType.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1;
+package org.apache.iotdb.db.pipe.connector.v1;
import java.util.Arrays;
import java.util.HashMap;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/reponse/PipeTransferFilePieceResp.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/reponse/PipeTransferFilePieceResp.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/reponse/PipeTransferFilePieceResp.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/reponse/PipeTransferFilePieceResp.java
index 56c3b8a4cd5..8b96e0242a2 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/reponse/PipeTransferFilePieceResp.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/reponse/PipeTransferFilePieceResp.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.reponse;
+package org.apache.iotdb.db.pipe.connector.v1.reponse;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferFilePieceReq.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferFilePieceReq.java
similarity index 92%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferFilePieceReq.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferFilePieceReq.java
index 6f6266fa1e9..897ee2a29e7 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferFilePieceReq.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferFilePieceReq.java
@@ -17,10 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request;
+package org.apache.iotdb.db.pipe.connector.v1.request;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.IoTDBThriftConnectorVersion;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.PipeRequestType;
+import org.apache.iotdb.db.pipe.connector.IoTDBThriftConnectorVersion;
+import org.apache.iotdb.db.pipe.connector.v1.PipeRequestType;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.PublicBAOS;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferFileSealReq.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferFileSealReq.java
similarity index 91%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferFileSealReq.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferFileSealReq.java
index 935131db950..0e73b95af07 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferFileSealReq.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferFileSealReq.java
@@ -17,10 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request;
+package org.apache.iotdb.db.pipe.connector.v1.request;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.IoTDBThriftConnectorVersion;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.PipeRequestType;
+import org.apache.iotdb.db.pipe.connector.IoTDBThriftConnectorVersion;
+import org.apache.iotdb.db.pipe.connector.v1.PipeRequestType;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferHandshakeReq.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferHandshakeReq.java
similarity index 91%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferHandshakeReq.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferHandshakeReq.java
index f258c5036f9..da03903fd79 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferHandshakeReq.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferHandshakeReq.java
@@ -17,10 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request;
+package org.apache.iotdb.db.pipe.connector.v1.request;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.IoTDBThriftConnectorVersion;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.PipeRequestType;
+import org.apache.iotdb.db.pipe.connector.IoTDBThriftConnectorVersion;
+import org.apache.iotdb.db.pipe.connector.v1.PipeRequestType;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferInsertNodeReq.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferInsertNodeReq.java
similarity index 94%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferInsertNodeReq.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferInsertNodeReq.java
index 5b4fc83ced8..10098a9721e 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferInsertNodeReq.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferInsertNodeReq.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request;
+package org.apache.iotdb.db.pipe.connector.v1.request;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertNode;
@@ -26,8 +26,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.mpp.plan.statement.Statement;
import org.apache.iotdb.db.mpp.plan.statement.crud.InsertRowStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.InsertTabletStatement;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.IoTDBThriftConnectorVersion;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.PipeRequestType;
+import org.apache.iotdb.db.pipe.connector.IoTDBThriftConnectorVersion;
+import org.apache.iotdb.db.pipe.connector.v1.PipeRequestType;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
public class PipeTransferInsertNodeReq extends TPipeTransferReq {
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferTabletReq.java b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferTabletReq.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferTabletReq.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferTabletReq.java
index ad54dd267e5..758013d6c72 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/impl/iotdb/v1/request/PipeTransferTabletReq.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/connector/v1/request/PipeTransferTabletReq.java
@@ -17,14 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request;
+package org.apache.iotdb.db.pipe.connector.v1.request;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.utils.PathUtils;
import org.apache.iotdb.db.mpp.plan.parser.StatementGenerator;
import org.apache.iotdb.db.mpp.plan.statement.crud.InsertTabletStatement;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.IoTDBThriftConnectorVersion;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.PipeRequestType;
+import org.apache.iotdb.db.pipe.connector.IoTDBThriftConnectorVersion;
+import org.apache.iotdb.db.pipe.connector.v1.PipeRequestType;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
import org.apache.iotdb.service.rpc.thrift.TSInsertTabletReq;
import org.apache.iotdb.session.util.SessionUtils;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/historical/PipeHistoricalDataRegionFakeCollector.java b/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/historical/PipeHistoricalDataRegionFakeCollector.java
deleted file mode 100644
index fb61a9b4cc1..00000000000
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/collector/historical/PipeHistoricalDataRegionFakeCollector.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * 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.pipe.core.collector.historical;
-
-import org.apache.iotdb.pipe.api.customizer.PipeParameterValidator;
-import org.apache.iotdb.pipe.api.customizer.PipeParameters;
-import org.apache.iotdb.pipe.api.customizer.collector.PipeCollectorRuntimeConfiguration;
-import org.apache.iotdb.pipe.api.event.Event;
-
-public class PipeHistoricalDataRegionFakeCollector extends PipeHistoricalDataRegionCollector {
-
- @Override
- public void validate(PipeParameterValidator validator) {}
-
- @Override
- public void customize(
- PipeParameters parameters, PipeCollectorRuntimeConfiguration configuration) {}
-
- @Override
- public void start() {}
-
- @Override
- public Event supply() {
- return null;
- }
-
- @Override
- public boolean hasConsumedAll() {
- return true;
- }
-
- @Override
- public void close() {}
-
- @Override
- public String toString() {
- return "PipeHistoricalDataRegionFakeCollector{}";
- }
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/EnrichedEvent.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java
similarity index 99%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/EnrichedEvent.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java
index e972104ce5d..4b181f83549 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/EnrichedEvent.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event;
+package org.apache.iotdb.db.pipe.event;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/utils/PipeDataTypeTransformer.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/row/PipeDataTypeTransformer.java
similarity index 69%
rename from node-commons/src/main/java/org/apache/iotdb/commons/pipe/utils/PipeDataTypeTransformer.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/event/common/row/PipeDataTypeTransformer.java
index 7e810f41c28..e233c3a8431 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/utils/PipeDataTypeTransformer.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/row/PipeDataTypeTransformer.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.commons.pipe.utils;
+package org.apache.iotdb.db.pipe.event.common.row;
import org.apache.iotdb.pipe.api.type.Type;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -30,9 +30,6 @@ import java.util.stream.Collectors;
* org.apache.iotdb.pipe.api.type.Type}
*/
public class PipeDataTypeTransformer {
- public static Type transformToPipeDataType(TSDataType tsDataType) {
- return tsDataType == null ? null : getPipeDataType(tsDataType.getType());
- }
public static List<Type> transformToPipeDataTypeList(List<TSDataType> tsDataTypeList) {
return tsDataTypeList == null
@@ -42,6 +39,10 @@ public class PipeDataTypeTransformer {
.collect(Collectors.toList());
}
+ public static Type transformToPipeDataType(TSDataType tsDataType) {
+ return tsDataType == null ? null : getPipeDataType(tsDataType.getType());
+ }
+
private static Type getPipeDataType(byte type) {
switch (type) {
case 0:
@@ -61,34 +62,7 @@ public class PipeDataTypeTransformer {
}
}
- public static TSDataType transformToTsDataType(Type type) {
- return type == null ? null : getTsDataType(type.getType());
- }
-
- public static List<TSDataType> transformToTsDataTypeList(List<Type> typeList) {
- return typeList == null
- ? null
- : typeList.stream()
- .map(PipeDataTypeTransformer::transformToTsDataType)
- .collect(Collectors.toList());
- }
-
- private static TSDataType getTsDataType(byte type) {
- switch (type) {
- case 0:
- return TSDataType.BOOLEAN;
- case 1:
- return TSDataType.INT32;
- case 2:
- return TSDataType.INT64;
- case 3:
- return TSDataType.FLOAT;
- case 4:
- return TSDataType.DOUBLE;
- case 5:
- return TSDataType.TEXT;
- default:
- throw new IllegalArgumentException("Invalid input: " + type);
- }
+ private PipeDataTypeTransformer() {
+ // util class
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/access/PipeRow.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/row/PipeRow.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/access/PipeRow.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/event/common/row/PipeRow.java
index 85f4a210ccb..8b9e10ebf91 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/access/PipeRow.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/row/PipeRow.java
@@ -17,9 +17,8 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event.view.access;
+package org.apache.iotdb.db.pipe.event.common.row;
-import org.apache.iotdb.commons.pipe.utils.PipeDataTypeTransformer;
import org.apache.iotdb.pipe.api.access.Row;
import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException;
import org.apache.iotdb.pipe.api.type.Binary;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/collector/PipeRowCollector.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/row/PipeRowCollector.java
similarity index 93%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/collector/PipeRowCollector.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/event/common/row/PipeRowCollector.java
index 8010b67e57e..983a7877713 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/collector/PipeRowCollector.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/row/PipeRowCollector.java
@@ -17,10 +17,9 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event.view.collector;
+package org.apache.iotdb.db.pipe.event.common.row;
-import org.apache.iotdb.db.pipe.core.event.impl.PipeRawTabletInsertionEvent;
-import org.apache.iotdb.db.pipe.core.event.view.access.PipeRow;
+import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.pipe.api.access.Row;
import org.apache.iotdb.pipe.api.collector.RowCollector;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/impl/PipeInsertNodeTabletInsertionEvent.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
similarity index 96%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/impl/PipeInsertNodeTabletInsertionEvent.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
index dd65ad4c319..a86842875d4 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/impl/PipeInsertNodeTabletInsertionEvent.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
@@ -17,13 +17,12 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event.impl;
+package org.apache.iotdb.db.pipe.event.common.tablet;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertNode;
-import org.apache.iotdb.db.pipe.core.event.EnrichedEvent;
-import org.apache.iotdb.db.pipe.core.event.view.datastructure.TabletInsertionDataContainer;
+import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.resource.PipeResourceManager;
import org.apache.iotdb.db.wal.exception.WALPipeException;
import org.apache.iotdb.db.wal.utils.WALEntryHandler;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/impl/PipeRawTabletInsertionEvent.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
similarity index 95%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/impl/PipeRawTabletInsertionEvent.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
index abb33534eb8..8629eb52ba4 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/impl/PipeRawTabletInsertionEvent.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
@@ -17,10 +17,9 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event.impl;
+package org.apache.iotdb.db.pipe.event.common.tablet;
import org.apache.iotdb.db.pipe.config.PipeCollectorConstant;
-import org.apache.iotdb.db.pipe.core.event.view.datastructure.TabletInsertionDataContainer;
import org.apache.iotdb.pipe.api.access.Row;
import org.apache.iotdb.pipe.api.collector.RowCollector;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/datastructure/TabletInsertionDataContainer.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/TabletInsertionDataContainer.java
similarity index 98%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/datastructure/TabletInsertionDataContainer.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/TabletInsertionDataContainer.java
index 542847074c0..49275556cd2 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/datastructure/TabletInsertionDataContainer.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/TabletInsertionDataContainer.java
@@ -17,14 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event.view.datastructure;
+package org.apache.iotdb.db.pipe.event.common.tablet;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode;
-import org.apache.iotdb.db.pipe.core.event.view.access.PipeRow;
-import org.apache.iotdb.db.pipe.core.event.view.collector.PipeRowCollector;
+import org.apache.iotdb.db.pipe.event.common.row.PipeRow;
+import org.apache.iotdb.db.pipe.event.common.row.PipeRowCollector;
import org.apache.iotdb.pipe.api.access.Row;
import org.apache.iotdb.pipe.api.collector.RowCollector;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/impl/PipeTsFileInsertionEvent.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
similarity index 93%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/impl/PipeTsFileInsertionEvent.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
index 2de0cd31d19..785948b66a4 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/impl/PipeTsFileInsertionEvent.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
@@ -17,15 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event.impl;
+package org.apache.iotdb.db.pipe.event.common.tsfile;
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;
import org.apache.iotdb.db.engine.storagegroup.TsFileProcessor;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.pipe.core.event.EnrichedEvent;
-import org.apache.iotdb.db.pipe.core.event.view.datastructure.TsFileInsertionDataContainer;
+import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.resource.PipeResourceManager;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
@@ -166,11 +165,6 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileIns
}
}
- @Override
- public TsFileInsertionEvent toTsFileInsertionEvent(Iterable<TabletInsertionEvent> iterable) {
- throw new UnsupportedOperationException("Not implemented yet");
- }
-
/////////////////////////// Object ///////////////////////////
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/datastructure/TsFileInsertionDataContainer.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/datastructure/TsFileInsertionDataContainer.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java
index 79274e48fc7..6426e8dccad 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/datastructure/TsFileInsertionDataContainer.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java
@@ -17,9 +17,9 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event.view.datastructure;
+package org.apache.iotdb.db.pipe.event.common.tsfile;
-import org.apache.iotdb.db.pipe.core.event.impl.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/datastructure/TsFileInsertionDataTabletIterator.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataTabletIterator.java
similarity index 98%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/datastructure/TsFileInsertionDataTabletIterator.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataTabletIterator.java
index 15959adb7ee..8b3b1e14fe5 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/datastructure/TsFileInsertionDataTabletIterator.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataTabletIterator.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event.view.datastructure;
+package org.apache.iotdb.db.pipe.event.common.tsfile;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/PipeRealtimeCollectEvent.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeCollectEvent.java
similarity index 96%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/PipeRealtimeCollectEvent.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeCollectEvent.java
index d91eb07a4b0..06410b26a02 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/PipeRealtimeCollectEvent.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeCollectEvent.java
@@ -17,11 +17,12 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event.realtime;
+package org.apache.iotdb.db.pipe.event.realtime;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
-import org.apache.iotdb.db.pipe.core.event.EnrichedEvent;
+import org.apache.iotdb.db.pipe.collector.realtime.epoch.TsFileEpoch;
+import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import org.apache.iotdb.pipe.api.event.Event;
import java.util.Map;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/PipeRealtimeCollectEventFactory.java b/server/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeCollectEventFactory.java
similarity index 86%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/PipeRealtimeCollectEventFactory.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeCollectEventFactory.java
index b2a622f7c05..274553301f3 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/realtime/PipeRealtimeCollectEventFactory.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeCollectEventFactory.java
@@ -17,12 +17,13 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event.realtime;
+package org.apache.iotdb.db.pipe.event.realtime;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertNode;
-import org.apache.iotdb.db.pipe.core.event.impl.PipeInsertNodeTabletInsertionEvent;
-import org.apache.iotdb.db.pipe.core.event.impl.PipeTsFileInsertionEvent;
+import org.apache.iotdb.db.pipe.collector.realtime.epoch.TsFileEpochManager;
+import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.wal.utils.WALEntryHandler;
public class PipeRealtimeCollectEventFactory {
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/execution/executor/PipeConnectorSubtaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/pipe/execution/executor/PipeConnectorSubtaskExecutor.java
index b6a26508ddc..83ce8c8febd 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/execution/executor/PipeConnectorSubtaskExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/execution/executor/PipeConnectorSubtaskExecutor.java
@@ -24,7 +24,7 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig;
public class PipeConnectorSubtaskExecutor extends PipeSubtaskExecutor {
- PipeConnectorSubtaskExecutor() {
+ public PipeConnectorSubtaskExecutor() {
super(
PipeConfig.getInstance().getPipeSubtaskExecutorMaxThreadNum(),
ThreadName.PIPE_CONNECTOR_EXECUTOR_POOL);
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/execution/executor/PipeProcessorSubtaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/pipe/execution/executor/PipeProcessorSubtaskExecutor.java
index f806e287fc4..3d3d5598699 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/execution/executor/PipeProcessorSubtaskExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/execution/executor/PipeProcessorSubtaskExecutor.java
@@ -24,7 +24,7 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig;
public class PipeProcessorSubtaskExecutor extends PipeSubtaskExecutor {
- PipeProcessorSubtaskExecutor() {
+ public PipeProcessorSubtaskExecutor() {
super(
PipeConfig.getInstance().getPipeSubtaskExecutorMaxThreadNum(),
ThreadName.PIPE_PROCESSOR_EXECUTOR_POOL);
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/processor/PipeDoNothingProcessor.java b/server/src/main/java/org/apache/iotdb/db/pipe/processor/PipeDoNothingProcessor.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/processor/PipeDoNothingProcessor.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/processor/PipeDoNothingProcessor.java
index c9774b4618a..a976358611e 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/processor/PipeDoNothingProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/processor/PipeDoNothingProcessor.java
@@ -17,10 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.processor;
+package org.apache.iotdb.db.pipe.processor;
import org.apache.iotdb.db.pipe.config.PipeCollectorConstant;
-import org.apache.iotdb.db.pipe.core.event.EnrichedEvent;
+import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import org.apache.iotdb.pipe.api.PipeProcessor;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.customizer.PipeParameterValidator;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/task/queue/BlockingPendingQueue.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/connection/BlockingPendingQueue.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/pipe/task/queue/BlockingPendingQueue.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/task/connection/BlockingPendingQueue.java
index 07465e5fa36..9896bfee3e9 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/task/queue/BlockingPendingQueue.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/connection/BlockingPendingQueue.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.task.queue;
+package org.apache.iotdb.db.pipe.task.connection;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.pipe.api.event.Event;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/task/queue/BoundedBlockingPendingQueue.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/connection/BoundedBlockingPendingQueue.java
similarity index 95%
rename from server/src/main/java/org/apache/iotdb/db/pipe/task/queue/BoundedBlockingPendingQueue.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/task/connection/BoundedBlockingPendingQueue.java
index 0940fee2753..3dcd76a441d 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/task/queue/BoundedBlockingPendingQueue.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/connection/BoundedBlockingPendingQueue.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.task.queue;
+package org.apache.iotdb.db.pipe.task.connection;
import org.apache.iotdb.pipe.api.event.Event;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/task/queue/EventSupplier.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/connection/EventSupplier.java
similarity index 96%
rename from server/src/main/java/org/apache/iotdb/db/pipe/task/queue/EventSupplier.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/task/connection/EventSupplier.java
index ea056dc22aa..0efb7a2ae67 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/task/queue/EventSupplier.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/connection/EventSupplier.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.task.queue;
+package org.apache.iotdb.db.pipe.task.connection;
import org.apache.iotdb.pipe.api.event.Event;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/collector/PipeEventCollector.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
similarity index 92%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/collector/PipeEventCollector.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
index 6a8697eec8b..4da053d9526 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/event/view/collector/PipeEventCollector.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
@@ -17,10 +17,9 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event.view.collector;
+package org.apache.iotdb.db.pipe.task.connection;
-import org.apache.iotdb.db.pipe.core.event.EnrichedEvent;
-import org.apache.iotdb.db.pipe.task.queue.BoundedBlockingPendingQueue;
+import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.event.Event;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/task/queue/UnboundedBlockingPendingQueue.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/connection/UnboundedBlockingPendingQueue.java
similarity index 95%
rename from server/src/main/java/org/apache/iotdb/db/pipe/task/queue/UnboundedBlockingPendingQueue.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/task/connection/UnboundedBlockingPendingQueue.java
index 3a94393dd08..dafb567e902 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/task/queue/UnboundedBlockingPendingQueue.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/connection/UnboundedBlockingPendingQueue.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.task.queue;
+package org.apache.iotdb.db.pipe.task.connection;
import org.apache.iotdb.pipe.api.event.Event;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskCollectorStage.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskCollectorStage.java
index a73a9da8b9e..9e03458bfa0 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskCollectorStage.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskCollectorStage.java
@@ -23,10 +23,10 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.commons.pipe.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
+import org.apache.iotdb.db.pipe.collector.IoTDBDataRegionCollector;
import org.apache.iotdb.db.pipe.config.PipeCollectorConstant;
-import org.apache.iotdb.db.pipe.core.collector.IoTDBDataRegionCollector;
-import org.apache.iotdb.db.pipe.task.queue.EventSupplier;
-import org.apache.iotdb.db.pipe.task.queue.UnboundedBlockingPendingQueue;
+import org.apache.iotdb.db.pipe.task.connection.EventSupplier;
+import org.apache.iotdb.db.pipe.task.connection.UnboundedBlockingPendingQueue;
import org.apache.iotdb.pipe.api.PipeCollector;
import org.apache.iotdb.pipe.api.customizer.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.PipeParameters;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskConnectorStage.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskConnectorStage.java
index 5615c36f560..6b96097fd6f 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskConnectorStage.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskConnectorStage.java
@@ -19,9 +19,9 @@
package org.apache.iotdb.db.pipe.task.stage;
-import org.apache.iotdb.db.pipe.core.connector.manager.PipeConnectorSubtaskManager;
import org.apache.iotdb.db.pipe.execution.executor.PipeSubtaskExecutorManager;
-import org.apache.iotdb.db.pipe.task.queue.BoundedBlockingPendingQueue;
+import org.apache.iotdb.db.pipe.task.connection.BoundedBlockingPendingQueue;
+import org.apache.iotdb.db.pipe.task.subtask.PipeConnectorSubtaskManager;
import org.apache.iotdb.pipe.api.customizer.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.exception.PipeException;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskProcessorStage.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskProcessorStage.java
index b21a3a85c34..571b04773a4 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskProcessorStage.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskProcessorStage.java
@@ -23,13 +23,13 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.commons.pipe.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.pipe.config.PipeProcessorConstant;
-import org.apache.iotdb.db.pipe.core.event.view.collector.PipeEventCollector;
-import org.apache.iotdb.db.pipe.core.processor.PipeDoNothingProcessor;
import org.apache.iotdb.db.pipe.execution.executor.PipeProcessorSubtaskExecutor;
import org.apache.iotdb.db.pipe.execution.executor.PipeSubtaskExecutorManager;
-import org.apache.iotdb.db.pipe.task.queue.BlockingPendingQueue;
-import org.apache.iotdb.db.pipe.task.queue.BoundedBlockingPendingQueue;
-import org.apache.iotdb.db.pipe.task.queue.EventSupplier;
+import org.apache.iotdb.db.pipe.processor.PipeDoNothingProcessor;
+import org.apache.iotdb.db.pipe.task.connection.BlockingPendingQueue;
+import org.apache.iotdb.db.pipe.task.connection.BoundedBlockingPendingQueue;
+import org.apache.iotdb.db.pipe.task.connection.EventSupplier;
+import org.apache.iotdb.db.pipe.task.connection.PipeEventCollector;
import org.apache.iotdb.db.pipe.task.subtask.PipeProcessorSubtask;
import org.apache.iotdb.pipe.api.PipeProcessor;
import org.apache.iotdb.pipe.api.customizer.PipeParameterValidator;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtask.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtask.java
index f34b2b3f961..ce2b6403a2f 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtask.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtask.java
@@ -21,8 +21,8 @@ package org.apache.iotdb.db.pipe.task.subtask;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeConnectorCriticalException;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
-import org.apache.iotdb.db.pipe.core.event.EnrichedEvent;
-import org.apache.iotdb.db.pipe.task.queue.BoundedBlockingPendingQueue;
+import org.apache.iotdb.db.pipe.event.EnrichedEvent;
+import org.apache.iotdb.db.pipe.task.connection.BoundedBlockingPendingQueue;
import org.apache.iotdb.pipe.api.PipeConnector;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/manager/PipeConnectorSubtaskLifeCycle.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtaskLifeCycle.java
similarity index 93%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/connector/manager/PipeConnectorSubtaskLifeCycle.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtaskLifeCycle.java
index 6936e4610cb..a0814043431 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/manager/PipeConnectorSubtaskLifeCycle.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtaskLifeCycle.java
@@ -17,11 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.connector.manager;
+package org.apache.iotdb.db.pipe.task.subtask;
import org.apache.iotdb.db.pipe.execution.executor.PipeConnectorSubtaskExecutor;
-import org.apache.iotdb.db.pipe.task.queue.BoundedBlockingPendingQueue;
-import org.apache.iotdb.db.pipe.task.subtask.PipeConnectorSubtask;
+import org.apache.iotdb.db.pipe.task.connection.BoundedBlockingPendingQueue;
import org.apache.iotdb.pipe.api.event.Event;
public class PipeConnectorSubtaskLifeCycle implements AutoCloseable {
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/manager/PipeConnectorSubtaskManager.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtaskManager.java
similarity index 94%
rename from server/src/main/java/org/apache/iotdb/db/pipe/core/connector/manager/PipeConnectorSubtaskManager.java
rename to server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtaskManager.java
index 548a48fdc0b..a6e338aae2e 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/core/connector/manager/PipeConnectorSubtaskManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeConnectorSubtaskManager.java
@@ -17,23 +17,21 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.connector.manager;
+package org.apache.iotdb.db.pipe.task.subtask;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.pipe.config.PipeConnectorConstant;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.IoTDBThriftConnectorV1;
+import org.apache.iotdb.db.pipe.connector.v1.IoTDBThriftConnectorV1;
import org.apache.iotdb.db.pipe.execution.executor.PipeConnectorSubtaskExecutor;
-import org.apache.iotdb.db.pipe.task.queue.BoundedBlockingPendingQueue;
-import org.apache.iotdb.db.pipe.task.subtask.PipeConnectorSubtask;
+import org.apache.iotdb.db.pipe.task.connection.BoundedBlockingPendingQueue;
import org.apache.iotdb.pipe.api.PipeConnector;
import org.apache.iotdb.pipe.api.customizer.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.PipeParameters;
import org.apache.iotdb.pipe.api.customizer.connector.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.exception.PipeException;
-import org.apache.iotdb.pipe.api.exception.PipeManagementException;
import java.util.HashMap;
import java.util.Map;
@@ -69,7 +67,7 @@ public class PipeConnectorSubtaskManager {
// TODO: use runtimeConfiguration to configure PipeConnector
pipeConnector.handshake();
} catch (Exception e) {
- throw new PipeManagementException(
+ throw new PipeException(
"Failed to construct PipeConnector, because of " + e.getMessage(), e);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeProcessorSubtask.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeProcessorSubtask.java
index 6a76beb02b3..2472a29eefe 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeProcessorSubtask.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeProcessorSubtask.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.db.pipe.task.subtask;
-import org.apache.iotdb.db.pipe.task.queue.EventSupplier;
+import org.apache.iotdb.db.pipe.task.connection.EventSupplier;
import org.apache.iotdb.pipe.api.PipeProcessor;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.event.Event;
diff --git a/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java b/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java
index 524e9abbcfa..9f5c6e4a37a 100644
--- a/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java
+++ b/server/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.pipe.task.subtask;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
-import org.apache.iotdb.db.pipe.core.event.EnrichedEvent;
+import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.execution.scheduler.PipeSubtaskScheduler;
import org.apache.iotdb.pipe.api.event.Event;
diff --git a/server/src/test/java/org/apache/iotdb/db/pipe/core/collector/CachedSchemaPatternMatcherTest.java b/server/src/test/java/org/apache/iotdb/db/pipe/collector/CachedSchemaPatternMatcherTest.java
similarity index 95%
rename from server/src/test/java/org/apache/iotdb/db/pipe/core/collector/CachedSchemaPatternMatcherTest.java
rename to server/src/test/java/org/apache/iotdb/db/pipe/collector/CachedSchemaPatternMatcherTest.java
index 4bc74e6d929..ef8e2e67e3a 100644
--- a/server/src/test/java/org/apache/iotdb/db/pipe/core/collector/CachedSchemaPatternMatcherTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/pipe/collector/CachedSchemaPatternMatcherTest.java
@@ -17,13 +17,13 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector;
+package org.apache.iotdb.db.pipe.collector;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
+import org.apache.iotdb.db.pipe.collector.realtime.PipeRealtimeDataRegionCollector;
+import org.apache.iotdb.db.pipe.collector.realtime.matcher.CachedSchemaPatternMatcher;
import org.apache.iotdb.db.pipe.config.PipeCollectorConstant;
-import org.apache.iotdb.db.pipe.core.collector.realtime.PipeRealtimeDataRegionCollector;
-import org.apache.iotdb.db.pipe.core.collector.realtime.matcher.CachedSchemaPatternMatcher;
-import org.apache.iotdb.db.pipe.core.event.realtime.PipeRealtimeCollectEvent;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeCollectEvent;
import org.apache.iotdb.pipe.api.customizer.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
diff --git a/server/src/test/java/org/apache/iotdb/db/pipe/core/collector/PipeRealtimeCollectTest.java b/server/src/test/java/org/apache/iotdb/db/pipe/collector/PipeRealtimeCollectTest.java
similarity index 96%
rename from server/src/test/java/org/apache/iotdb/db/pipe/core/collector/PipeRealtimeCollectTest.java
rename to server/src/test/java/org/apache/iotdb/db/pipe/collector/PipeRealtimeCollectTest.java
index d071c1ae5ef..202c1527bd2 100644
--- a/server/src/test/java/org/apache/iotdb/db/pipe/core/collector/PipeRealtimeCollectTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/pipe/collector/PipeRealtimeCollectTest.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.collector;
+package org.apache.iotdb.db.pipe.collector;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.path.PartialPath;
@@ -25,11 +25,11 @@ import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
+import org.apache.iotdb.db.pipe.collector.realtime.PipeRealtimeDataRegionCollector;
+import org.apache.iotdb.db.pipe.collector.realtime.PipeRealtimeDataRegionHybridCollector;
+import org.apache.iotdb.db.pipe.collector.realtime.listener.PipeInsertionDataNodeListener;
import org.apache.iotdb.db.pipe.config.PipeCollectorConstant;
-import org.apache.iotdb.db.pipe.core.collector.realtime.PipeRealtimeDataRegionCollector;
-import org.apache.iotdb.db.pipe.core.collector.realtime.PipeRealtimeDataRegionHybridCollector;
-import org.apache.iotdb.db.pipe.core.collector.realtime.listener.PipeInsertionDataNodeListener;
-import org.apache.iotdb.db.pipe.task.queue.UnboundedBlockingPendingQueue;
+import org.apache.iotdb.db.pipe.task.connection.UnboundedBlockingPendingQueue;
import org.apache.iotdb.db.wal.utils.WALEntryHandler;
import org.apache.iotdb.pipe.api.customizer.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
diff --git a/server/src/test/java/org/apache/iotdb/db/pipe/core/connector/PipeThriftRequestTest.java b/server/src/test/java/org/apache/iotdb/db/pipe/connector/PipeThriftRequestTest.java
similarity index 89%
rename from server/src/test/java/org/apache/iotdb/db/pipe/core/connector/PipeThriftRequestTest.java
rename to server/src/test/java/org/apache/iotdb/db/pipe/connector/PipeThriftRequestTest.java
index a3b65a66bfd..72e5075cb43 100644
--- a/server/src/test/java/org/apache/iotdb/db/pipe/core/connector/PipeThriftRequestTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/pipe/connector/PipeThriftRequestTest.java
@@ -17,16 +17,16 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.connector;
+package org.apache.iotdb.db.pipe.connector;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.reponse.PipeTransferFilePieceResp;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request.PipeTransferFilePieceReq;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request.PipeTransferFileSealReq;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request.PipeTransferHandshakeReq;
-import org.apache.iotdb.db.pipe.core.connector.impl.iotdb.v1.request.PipeTransferInsertNodeReq;
+import org.apache.iotdb.db.pipe.connector.v1.reponse.PipeTransferFilePieceResp;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferFilePieceReq;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferFileSealReq;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferHandshakeReq;
+import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferInsertNodeReq;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
diff --git a/server/src/test/java/org/apache/iotdb/db/pipe/core/event/PipeTabletInsertionEventTest.java b/server/src/test/java/org/apache/iotdb/db/pipe/event/PipeTabletInsertionEventTest.java
similarity index 95%
rename from server/src/test/java/org/apache/iotdb/db/pipe/core/event/PipeTabletInsertionEventTest.java
rename to server/src/test/java/org/apache/iotdb/db/pipe/event/PipeTabletInsertionEventTest.java
index 6a68b5ad7cd..abd0cf9b056 100644
--- a/server/src/test/java/org/apache/iotdb/db/pipe/core/event/PipeTabletInsertionEventTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/pipe/event/PipeTabletInsertionEventTest.java
@@ -17,15 +17,15 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event;
+package org.apache.iotdb.db.pipe.event;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode;
-import org.apache.iotdb.db.pipe.core.event.impl.PipeRawTabletInsertionEvent;
-import org.apache.iotdb.db.pipe.core.event.view.datastructure.TabletInsertionDataContainer;
+import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.tablet.TabletInsertionDataContainer;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.BitMap;
@@ -40,8 +40,8 @@ import java.util.Arrays;
public class PipeTabletInsertionEventTest {
- InsertRowNode insertRowNode;
- InsertTabletNode insertTabletNode;
+ private InsertRowNode insertRowNode;
+ private InsertTabletNode insertTabletNode;
final String deviceId = "root.sg.d1";
final long[] times = new long[] {110L, 111L, 112L, 113L, 114L};
diff --git a/server/src/test/java/org/apache/iotdb/db/pipe/core/event/TsFileInsertionDataContainerTest.java b/server/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
similarity index 99%
rename from server/src/test/java/org/apache/iotdb/db/pipe/core/event/TsFileInsertionDataContainerTest.java
rename to server/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
index d74f790fd15..aa0b7506cbf 100644
--- a/server/src/test/java/org/apache/iotdb/db/pipe/core/event/TsFileInsertionDataContainerTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
@@ -17,10 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.core.event;
+package org.apache.iotdb.db.pipe.event;
-import org.apache.iotdb.db.pipe.core.event.impl.PipeRawTabletInsertionEvent;
-import org.apache.iotdb.db.pipe.core.event.view.datastructure.TsFileInsertionDataContainer;
+import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.tsfile.TsFileInsertionDataContainer;
import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.utils.TsFileGeneratorUtils;
diff --git a/server/src/test/java/org/apache/iotdb/db/pipe/execution/executor/PipeConnectorSubtaskExecutorTest.java b/server/src/test/java/org/apache/iotdb/db/pipe/execution/PipeConnectorSubtaskExecutorTest.java
similarity index 87%
rename from server/src/test/java/org/apache/iotdb/db/pipe/execution/executor/PipeConnectorSubtaskExecutorTest.java
rename to server/src/test/java/org/apache/iotdb/db/pipe/execution/PipeConnectorSubtaskExecutorTest.java
index 802afcf16d9..08ec913c32a 100644
--- a/server/src/test/java/org/apache/iotdb/db/pipe/execution/executor/PipeConnectorSubtaskExecutorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/pipe/execution/PipeConnectorSubtaskExecutorTest.java
@@ -17,9 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.execution.executor;
+package org.apache.iotdb.db.pipe.execution;
-import org.apache.iotdb.db.pipe.task.queue.BoundedBlockingPendingQueue;
+import org.apache.iotdb.db.pipe.execution.executor.PipeConnectorSubtaskExecutor;
+import org.apache.iotdb.db.pipe.task.connection.BoundedBlockingPendingQueue;
import org.apache.iotdb.db.pipe.task.subtask.PipeConnectorSubtask;
import org.apache.iotdb.pipe.api.PipeConnector;
diff --git a/server/src/test/java/org/apache/iotdb/db/pipe/execution/executor/PipeProcessorSubtaskExecutorTest.java b/server/src/test/java/org/apache/iotdb/db/pipe/execution/PipeProcessorSubtaskExecutorTest.java
similarity index 89%
rename from server/src/test/java/org/apache/iotdb/db/pipe/execution/executor/PipeProcessorSubtaskExecutorTest.java
rename to server/src/test/java/org/apache/iotdb/db/pipe/execution/PipeProcessorSubtaskExecutorTest.java
index d0a5208d537..b1a454c732c 100644
--- a/server/src/test/java/org/apache/iotdb/db/pipe/execution/executor/PipeProcessorSubtaskExecutorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/pipe/execution/PipeProcessorSubtaskExecutorTest.java
@@ -17,9 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.execution.executor;
+package org.apache.iotdb.db.pipe.execution;
-import org.apache.iotdb.db.pipe.task.queue.EventSupplier;
+import org.apache.iotdb.db.pipe.execution.executor.PipeProcessorSubtaskExecutor;
+import org.apache.iotdb.db.pipe.task.connection.EventSupplier;
import org.apache.iotdb.db.pipe.task.subtask.PipeProcessorSubtask;
import org.apache.iotdb.pipe.api.PipeProcessor;
import org.apache.iotdb.pipe.api.collector.EventCollector;
diff --git a/server/src/test/java/org/apache/iotdb/db/pipe/execution/executor/PipeSubtaskExecutorTest.java b/server/src/test/java/org/apache/iotdb/db/pipe/execution/PipeSubtaskExecutorTest.java
similarity index 97%
rename from server/src/test/java/org/apache/iotdb/db/pipe/execution/executor/PipeSubtaskExecutorTest.java
rename to server/src/test/java/org/apache/iotdb/db/pipe/execution/PipeSubtaskExecutorTest.java
index d24efc3fb7f..883804cf69d 100644
--- a/server/src/test/java/org/apache/iotdb/db/pipe/execution/executor/PipeSubtaskExecutorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/pipe/execution/PipeSubtaskExecutorTest.java
@@ -17,8 +17,9 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.execution.executor;
+package org.apache.iotdb.db.pipe.execution;
+import org.apache.iotdb.db.pipe.execution.executor.PipeSubtaskExecutor;
import org.apache.iotdb.db.pipe.task.subtask.PipeSubtask;
import org.junit.After;