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 2025/01/13 17:26:12 UTC
(iotdb) branch master updated: Pipe Plugin: intro class annotations for pipe plugin & tree/table model isolation for drop/show pipe plugin operations (#14673)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 97a08c880fb Pipe Plugin: intro class annotations for pipe plugin & tree/table model isolation for drop/show pipe plugin operations (#14673)
97a08c880fb is described below
commit 97a08c880fb854c6e5c3e33ab04f0d29dcd9db3c
Author: VGalaxies <vg...@apache.org>
AuthorDate: Tue Jan 14 01:26:05 2025 +0800
Pipe Plugin: intro class annotations for pipe plugin & tree/table model isolation for drop/show pipe plugin operations (#14673)
Co-authored-by: Steve Yurong Su <ro...@apache.org>
---
.../java/org/apache/iotdb/CountPointProcessor.java | 2 +
.../java/org/apache/iotdb/pipe/api/PipePlugin.java | 37 ++++++
.../iotdb/pipe/api/annotation/TableModel.java | 23 ++--
.../iotdb/pipe/api/annotation/TreeModel.java | 23 ++--
.../response/pipe/plugin/PipePluginTableResp.java | 27 ++++-
.../response/pipe/task/PipeTableResp.java | 2 +-
.../iotdb/confignode/manager/ConfigManager.java | 9 ++
.../apache/iotdb/confignode/manager/IManager.java | 4 +
.../protocol/IoTDBConfigRegionAirGapConnector.java | 4 +
.../protocol/IoTDBConfigRegionConnector.java | 4 +
.../coordinator/plugin/PipePluginCoordinator.java | 29 +++++
.../pipe/extractor/IoTDBConfigRegionExtractor.java | 4 +
.../persistence/pipe/PipePluginInfo.java | 39 +++++-
.../confignode/persistence/pipe/PipeTaskInfo.java | 7 +-
.../thrift/ConfigNodeRPCServiceProcessor.java | 6 +
.../response/pipe/PipePluginTableRespTest.java | 4 +-
.../pipe/annotation/PipePluginAnnotationTest.java} | 23 ++--
.../pipe/agent/plugin/PipeDataNodePluginAgent.java | 20 +--
.../dataregion/PipeDataRegionPluginAgent.java | 49 ++++++++
.../airgap/IoTDBDataNodeAirGapConnector.java | 4 +
.../airgap/IoTDBDataRegionAirGapConnector.java | 4 +
.../airgap/IoTDBSchemaRegionAirGapConnector.java | 4 +
.../protocol/legacy/IoTDBLegacyPipeConnector.java | 2 +
.../connector/protocol/opcua/OpcUaConnector.java | 4 +
.../pipeconsensus/PipeConsensusAsyncConnector.java | 4 +
.../pipeconsensus/PipeConsensusSyncConnector.java | 4 +
.../async/IoTDBDataRegionAsyncConnector.java | 4 +
.../thrift/sync/IoTDBDataNodeSyncConnector.java | 4 +
.../thrift/sync/IoTDBDataRegionSyncConnector.java | 4 +
.../thrift/sync/IoTDBSchemaRegionConnector.java | 4 +
.../protocol/websocket/WebSocketConnector.java | 2 +
.../protocol/writeback/WriteBackConnector.java | 4 +
.../pipe/consensus/deletion/DeletionResource.java | 2 +-
.../dataregion/IoTDBDataRegionExtractor.java | 4 +
...oricalDataRegionTsFileAndDeletionExtractor.java | 2 +-
.../schemaregion/IoTDBSchemaRegionExtractor.java | 4 +
.../processor/aggregate/AggregateProcessor.java | 2 +
.../StandardStatisticsOperatorProcessor.java | 2 +
.../processor/TumblingWindowingProcessor.java | 2 +
.../changing/ChangingValueSamplingProcessor.java | 2 +
.../sdt/SwingingDoorTrendingSamplingProcessor.java | 2 +
.../tumbling/TumblingTimeSamplingProcessor.java | 2 +
.../pipeconsensus/PipeConsensusProcessor.java | 4 +
.../schemachange/RenameDatabaseProcessor.java | 2 +
.../twostage/plugin/TwoStageCountProcessor.java | 2 +
.../iotdb/db/protocol/client/ConfigNodeClient.java | 8 ++
.../execution/config/TableConfigTaskVisitor.java | 2 +-
.../execution/config/TreeConfigTaskVisitor.java | 2 +-
.../config/executor/ClusterConfigTaskExecutor.java | 58 ++++++---
.../config/executor/IConfigTaskExecutor.java | 4 +-
.../config/metadata/DropPipePluginTask.java | 13 +-
.../config/metadata/ShowPipePluginsTask.java | 20 ++-
.../metadata/pipe/DropPipePluginStatement.java | 17 ++-
.../metadata/pipe/ShowPipePluginsStatement.java | 23 ++--
.../dataregion/tsfile/TsFileResource.java | 2 +-
.../annotation/PipePluginAnnotationTest.java} | 23 ++--
iotdb-core/node-commons/pom.xml | 4 +
.../commons/pipe/agent/plugin/PipePluginAgent.java | 17 ++-
.../builtin/connector/PlaceholderConnector.java | 4 +
.../connector/donothing/DoNothingConnector.java | 4 +
.../iotdb/airgap/IoTDBAirGapConnector.java | 4 +
.../consensus/PipeConsensusAsyncConnector.java | 4 +
.../iotdb/thrift/IoTDBLegacyPipeConnector.java | 2 +
.../iotdb/thrift/IoTDBThriftAsyncConnector.java | 5 +
.../iotdb/thrift/IoTDBThriftConnector.java | 4 +
.../iotdb/thrift/IoTDBThriftSslConnector.java | 4 +
.../iotdb/thrift/IoTDBThriftSyncConnector.java | 5 +
.../builtin/connector/opcua/OpcUaConnector.java | 4 +
.../connector/websocket/WebSocketConnector.java | 2 +
.../connector/writeback/WriteBackConnector.java | 4 +
.../extractor/donothing/DoNothingExtractor.java | 4 +
.../builtin/extractor/iotdb/IoTDBExtractor.java | 4 +
.../processor/donothing/DoNothingProcessor.java | 4 +
.../throwing/ThrowingExceptionProcessor.java | 4 +
.../agent/plugin/meta/PipePluginMetaKeeper.java | 47 +++++--
.../commons/pipe/agent/task/PipeTaskAgent.java | 2 +-
.../commons/pipe/agent/task/meta/PipeMeta.java | 40 ------
.../pipe/agent/task/meta/PipeMetaKeeper.java | 6 +-
.../pipe/agent/task/meta/PipeStaticMeta.java | 14 +++
.../connector/protocol/IoTDBAirGapConnector.java | 4 +
.../pipe/connector/protocol/IoTDBConnector.java | 4 +
.../connector/protocol/IoTDBSslSyncConnector.java | 4 +
.../{ => resource}/PersistentResource.java | 2 +-
.../Result.java} | 39 ++++--
.../pipe/datastructure/visibility/Visibility.java | 9 +-
.../visibility/VisibilityTestUtils.java | 135 +++++++++++++++++++++
.../datastructure/visibility/VisibilityUtils.java | 133 ++++++++++++++++++++
.../commons/pipe/extractor/IoTDBExtractor.java | 4 +
.../extractor/IoTDBNonDataRegionExtractor.java | 4 +
.../annotation/PipePluginAnnotationTest.java} | 23 ++--
.../src/main/thrift/confignode.thrift | 10 ++
91 files changed, 943 insertions(+), 188 deletions(-)
diff --git a/example/pipe-count-point-processor/src/main/java/org/apache/iotdb/CountPointProcessor.java b/example/pipe-count-point-processor/src/main/java/org/apache/iotdb/CountPointProcessor.java
index 8444ad5e1eb..8d773deee96 100644
--- a/example/pipe-count-point-processor/src/main/java/org/apache/iotdb/CountPointProcessor.java
+++ b/example/pipe-count-point-processor/src/main/java/org/apache/iotdb/CountPointProcessor.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.pipe.api.PipeProcessor;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -37,6 +38,7 @@ import org.apache.tsfile.write.schema.MeasurementSchema;
import java.util.Collections;
import java.util.concurrent.atomic.AtomicLong;
+@TreeModel
public class CountPointProcessor implements PipeProcessor {
private static final String AGGREGATE_SERIES_KEY = "aggregate-series";
private static final AtomicLong writePointCount = new AtomicLong(0);
diff --git a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipePlugin.java b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipePlugin.java
index ec06f5de31e..276c5c8b300 100644
--- a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipePlugin.java
+++ b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipePlugin.java
@@ -19,4 +19,41 @@
package org.apache.iotdb.pipe.api;
+/**
+ * {@link PipePlugin}
+ *
+ * <p>{@link PipePlugin} represents a customizable component that can serve as a data extraction
+ * plugin, data processing plugin, or data sending plugin within a pipeline framework.
+ *
+ * <p>Developers can implement different plugin functionalities according to specific requirements,
+ * such as collecting data from various sources, transforming the data, or forwarding the data to
+ * external systems.
+ *
+ * <p>Usage Model:
+ *
+ * <ul>
+ * <li>By default, a {@link PipePlugin} can operate under a tree model only, if no model
+ * annotation under {@link org.apache.iotdb.pipe.api.annotation} is specified.
+ * <li>To extend its applicability, a {@link PipePlugin} can be configured to support table model,
+ * or both tree and table models concurrently.
+ * </ul>
+ *
+ * <p>Lifecycle:
+ *
+ * <ul>
+ * <li>When the pipeline framework loads, the plugin's configuration is parsed and validated.
+ * <li>As part of the setup, methods can be provided to prepare connections or resources required
+ * by the plugin (e.g., reading external configurations, establishing data routes).
+ * <li>During data processing, the plugin performs its core functionality (extraction,
+ * transformation, or sending).
+ * <li>When the pipeline is stopped or destroyed, any allocated resources must be released
+ * accurately, and {@link #close()} will be invoked to ensure a clean shutdown.
+ * </ul>
+ *
+ * <p>Example: {@link org.apache.iotdb.CountPointProcessor}
+ *
+ * <p>Implementations of {@link PipePlugin} should follow best practices for resource management and
+ * gracefully handle exceptions, especially when running in long-lived or continuously operating
+ * environments.
+ */
public interface PipePlugin extends AutoCloseable {}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/consensus/PipeConsensusAsyncConnector.java b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/annotation/TableModel.java
similarity index 52%
copy from iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/consensus/PipeConsensusAsyncConnector.java
copy to iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/annotation/TableModel.java
index fb55bc71a10..8a20d3b6513 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/consensus/PipeConsensusAsyncConnector.java
+++ b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/annotation/TableModel.java
@@ -17,14 +17,23 @@
* under the License.
*/
-package org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.consensus;
+package org.apache.iotdb.pipe.api.annotation;
-import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderConnector;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
/**
- * This class is a placeholder and should not be initialized. It represents the
- * PipeConsensusAsyncConnector. There is a real implementation in the server module but cannot be
- * imported here. The pipe agent in the server module will replace this class with the real
- * implementation when initializing the PipeConsensusAsyncConnector.
+ * Indicates that a plugin can be used in table model environments.
+ *
+ * <p>When implementing a custom {@link org.apache.iotdb.pipe.api.PipePlugin} that needs to operate
+ * under table model settings, declare this annotation on the plugin class. Through the {@code
+ * CREATE PIPEPLUGIN} statement, a plugin annotated with {@link TableModel} is valid for both tree
+ * model connections and table model connections.
+ *
+ * @since 2.0.0
*/
-public class PipeConsensusAsyncConnector extends PlaceholderConnector {}
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface TableModel {}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/consensus/PipeConsensusAsyncConnector.java b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/annotation/TreeModel.java
similarity index 52%
copy from iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/consensus/PipeConsensusAsyncConnector.java
copy to iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/annotation/TreeModel.java
index fb55bc71a10..bc0fca6c52e 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/consensus/PipeConsensusAsyncConnector.java
+++ b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/annotation/TreeModel.java
@@ -17,14 +17,23 @@
* under the License.
*/
-package org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.consensus;
+package org.apache.iotdb.pipe.api.annotation;
-import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderConnector;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
/**
- * This class is a placeholder and should not be initialized. It represents the
- * PipeConsensusAsyncConnector. There is a real implementation in the server module but cannot be
- * imported here. The pipe agent in the server module will replace this class with the real
- * implementation when initializing the PipeConsensusAsyncConnector.
+ * Indicates that a plugin can be used in tree model environments.
+ *
+ * <p>When implementing a custom {@link org.apache.iotdb.pipe.api.PipePlugin} that needs to operate
+ * under tree model settings, declare this annotation on the plugin class. Through the {@code CREATE
+ * PIPEPLUGIN} statement, a plugin annotated with {@link TreeModel} is valid for both tree model
+ * connections and tree model connections.
+ *
+ * @since 2.0.0
*/
-public class PipeConsensusAsyncConnector extends PlaceholderConnector {}
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface TreeModel {}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/plugin/PipePluginTableResp.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/plugin/PipePluginTableResp.java
index 44f6f1d779f..534b824ef47 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/plugin/PipePluginTableResp.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/plugin/PipePluginTableResp.java
@@ -21,6 +21,8 @@ package org.apache.iotdb.confignode.consensus.response.pipe.plugin;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.pipe.agent.plugin.meta.PipePluginMeta;
+import org.apache.iotdb.commons.pipe.datastructure.visibility.Visibility;
+import org.apache.iotdb.commons.pipe.datastructure.visibility.VisibilityUtils;
import org.apache.iotdb.confignode.rpc.thrift.TGetPipePluginTableResp;
import org.apache.iotdb.consensus.common.DataSet;
@@ -28,15 +30,22 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
public class PipePluginTableResp implements DataSet {
private final TSStatus status;
private final List<PipePluginMeta> allPipePluginMeta;
+ private final Map<String, Visibility> pipePluginNameToVisibilityMap;
- public PipePluginTableResp(TSStatus status, List<PipePluginMeta> allPipePluginMeta) {
+ public PipePluginTableResp(
+ TSStatus status,
+ List<PipePluginMeta> allPipePluginMeta,
+ Map<String, Visibility> pipePluginNameToVisibilityMap) {
this.status = status;
this.allPipePluginMeta = allPipePluginMeta;
+ this.pipePluginNameToVisibilityMap = pipePluginNameToVisibilityMap;
}
public TGetPipePluginTableResp convertToThriftResponse() throws IOException {
@@ -46,4 +55,20 @@ public class PipePluginTableResp implements DataSet {
}
return new TGetPipePluginTableResp(status, pipePluginInformationByteBuffers);
}
+
+ public PipePluginTableResp filter(final boolean isTableModel) {
+ return new PipePluginTableResp(
+ status,
+ allPipePluginMeta.stream()
+ .filter(
+ meta -> {
+ final String pipePluginName = meta.getPluginName();
+ final Visibility visibility =
+ pipePluginNameToVisibilityMap.getOrDefault(
+ pipePluginName, Visibility.TREE_ONLY);
+ return VisibilityUtils.isCompatible(visibility, isTableModel);
+ })
+ .collect(Collectors.toList()),
+ pipePluginNameToVisibilityMap);
+ }
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java
index 20cd76e4a41..4d854be05a3 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java
@@ -97,7 +97,7 @@ public class PipeTableResp implements DataSet {
public PipeTableResp filter(
final Boolean whereClause, final String pipeName, final boolean isTableModel) {
final PipeTableResp resp = filter(whereClause, pipeName);
- resp.allPipeMeta.removeIf(meta -> !meta.visibleUnder(isTableModel));
+ resp.allPipeMeta.removeIf(meta -> !meta.getStaticMeta().visibleUnder(isTableModel));
return resp;
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index 030ff801bb5..94be1f938b7 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -215,6 +215,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowModelReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowModelResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowPipePluginReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionReq;
@@ -1609,6 +1610,14 @@ public class ConfigManager implements IManager {
: new TGetPipePluginTableResp(status, Collections.emptyList());
}
+ @Override
+ public TGetPipePluginTableResp getPipePluginTableExtended(TShowPipePluginReq req) {
+ TSStatus status = confirmLeader();
+ return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ ? pipeManager.getPipePluginCoordinator().getPipePluginTableExtended(req)
+ : new TGetPipePluginTableResp(status, Collections.emptyList());
+ }
+
@Override
public TGetJarInListResp getPipePluginJar(TGetJarInListReq req) {
TSStatus status = confirmLeader();
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
index 82dae052ef2..41a3971755f 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
@@ -138,6 +138,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowModelReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowModelResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowPipePluginReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionReq;
@@ -556,6 +557,9 @@ public interface IManager {
/** Show pipe plugins. */
TGetPipePluginTableResp getPipePluginTable();
+ /** Show pipe plugins. */
+ TGetPipePluginTableResp getPipePluginTableExtended(TShowPipePluginReq req);
+
/** Get pipe plugin jar. */
TGetJarInListResp getPipePluginJar(TGetJarInListReq req);
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionAirGapConnector.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionAirGapConnector.java
index 419a7a75de4..f20a412a7c0 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionAirGapConnector.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionAirGapConnector.java
@@ -33,6 +33,8 @@ import org.apache.iotdb.confignode.manager.pipe.event.PipeConfigRegionSnapshotEv
import org.apache.iotdb.confignode.manager.pipe.event.PipeConfigRegionWritePlanEvent;
import org.apache.iotdb.confignode.service.ConfigNode;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
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;
@@ -48,6 +50,8 @@ import java.io.IOException;
import java.util.HashMap;
import java.util.Objects;
+@TreeModel
+@TableModel
public class IoTDBConfigRegionAirGapConnector extends IoTDBAirGapConnector {
private static final Logger LOGGER =
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionConnector.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionConnector.java
index 9777e22a53e..3fb90f03cf0 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionConnector.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionConnector.java
@@ -33,6 +33,8 @@ import org.apache.iotdb.confignode.manager.pipe.connector.payload.PipeTransferCo
import org.apache.iotdb.confignode.manager.pipe.event.PipeConfigRegionSnapshotEvent;
import org.apache.iotdb.confignode.manager.pipe.event.PipeConfigRegionWritePlanEvent;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
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;
@@ -52,6 +54,8 @@ import java.util.Collections;
import java.util.List;
import java.util.Objects;
+@TreeModel
+@TableModel
public class IoTDBConfigRegionConnector extends IoTDBSslSyncConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBConfigRegionConnector.class);
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/plugin/PipePluginCoordinator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/plugin/PipePluginCoordinator.java
index b26e8ab3ce1..8a53fd84b90 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/plugin/PipePluginCoordinator.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/plugin/PipePluginCoordinator.java
@@ -32,7 +32,9 @@ import org.apache.iotdb.confignode.rpc.thrift.TDropPipePluginReq;
import org.apache.iotdb.confignode.rpc.thrift.TGetJarInListReq;
import org.apache.iotdb.confignode.rpc.thrift.TGetJarInListResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetPipePluginTableResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowPipePluginReq;
import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.slf4j.Logger;
@@ -82,6 +84,18 @@ public class PipePluginCoordinator {
}
public TSStatus dropPipePlugin(TDropPipePluginReq req) {
+ final String pluginName = req.getPluginName();
+ final boolean isSetIfExistsCondition =
+ req.isSetIfExistsCondition() && req.isIfExistsCondition();
+ if (!pipePluginInfo.isPipePluginExisted(pluginName, req.isTableModel)) {
+ return isSetIfExistsCondition
+ ? RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)
+ : RpcUtils.getStatus(
+ TSStatusCode.DROP_PIPE_PLUGIN_ERROR,
+ String.format(
+ "Failed to drop pipe plugin %s. Failures: %s does not exist.",
+ pluginName, pluginName));
+ }
return configManager.getProcedureManager().dropPipePlugin(req);
}
@@ -99,6 +113,21 @@ public class PipePluginCoordinator {
}
}
+ public TGetPipePluginTableResp getPipePluginTableExtended(TShowPipePluginReq req) {
+ try {
+ return ((PipePluginTableResp)
+ configManager.getConsensusManager().read(new GetPipePluginTablePlan()))
+ .filter(req.isTableModel)
+ .convertToThriftResponse();
+ } catch (IOException | ConsensusException e) {
+ LOGGER.error("Fail to get PipePluginTable", e);
+ return new TGetPipePluginTableResp(
+ new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage(e.getMessage()),
+ Collections.emptyList());
+ }
+ }
+
public TGetJarInListResp getPipePluginJar(TGetJarInListReq req) {
try {
return ((JarResp)
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/IoTDBConfigRegionExtractor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/IoTDBConfigRegionExtractor.java
index 537539ec55b..9387d67bebe 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/IoTDBConfigRegionExtractor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/IoTDBConfigRegionExtractor.java
@@ -43,6 +43,8 @@ import org.apache.iotdb.confignode.manager.pipe.metric.PipeConfigRegionExtractor
import org.apache.iotdb.confignode.service.ConfigNode;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.exception.PipeException;
@@ -52,6 +54,8 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Set;
+@TreeModel
+@TableModel
public class IoTDBConfigRegionExtractor extends IoTDBNonDataRegionExtractor {
public static final PipeConfigPhysicalPlanTreePatternParseVisitor TREE_PATTERN_PARSE_VISITOR =
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipePluginInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipePluginInfo.java
index 818cb585fa4..baecc040735 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipePluginInfo.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipePluginInfo.java
@@ -23,10 +23,13 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.executable.ExecutableManager;
import org.apache.iotdb.commons.pipe.agent.plugin.meta.ConfigNodePipePluginMetaKeeper;
import org.apache.iotdb.commons.pipe.agent.plugin.meta.PipePluginMeta;
+import org.apache.iotdb.commons.pipe.agent.plugin.service.PipePluginClassLoader;
+import org.apache.iotdb.commons.pipe.agent.plugin.service.PipePluginClassLoaderManager;
import org.apache.iotdb.commons.pipe.agent.plugin.service.PipePluginExecutableManager;
import org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant;
import org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant;
import org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant;
+import org.apache.iotdb.commons.pipe.datastructure.visibility.VisibilityUtils;
import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
@@ -57,6 +60,8 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
import static org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin.DO_NOTHING_PROCESSOR;
import static org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin.IOTDB_EXTRACTOR;
@@ -73,12 +78,15 @@ public class PipePluginInfo implements SnapshotProcessor {
private final ConfigNodePipePluginMetaKeeper pipePluginMetaKeeper;
private final PipePluginExecutableManager pipePluginExecutableManager;
+ private final PipePluginClassLoaderManager classLoaderManager;
public PipePluginInfo() throws IOException {
this.pipePluginMetaKeeper = new ConfigNodePipePluginMetaKeeper();
this.pipePluginExecutableManager =
PipePluginExecutableManager.setupAndGetInstance(
CONFIG_NODE_CONF.getPipeTemporaryLibDir(), CONFIG_NODE_CONF.getPipeDir());
+ this.classLoaderManager =
+ PipePluginClassLoaderManager.setupAndGetInstance(CONFIG_NODE_CONF.getPipeDir());
}
/////////////////////////////// Lock ///////////////////////////////
@@ -138,6 +146,16 @@ public class PipePluginInfo implements SnapshotProcessor {
return false;
}
+ public boolean isPipePluginExisted(final String pipePluginName, final boolean isTableModel) {
+ acquirePipePluginInfoLock();
+ try {
+ return pipePluginMetaKeeper.containsPipePlugin(pipePluginName)
+ && pipePluginMetaKeeper.visibleUnder(pipePluginName, isTableModel);
+ } finally {
+ releasePipePluginInfoLock();
+ }
+ }
+
public boolean isJarNeededToBeSavedWhenCreatingPipePlugin(final String jarName) {
return !pipePluginMetaKeeper.containsJar(jarName);
}
@@ -193,19 +211,28 @@ public class PipePluginInfo implements SnapshotProcessor {
public TSStatus createPipePlugin(final CreatePipePluginPlan createPipePluginPlan) {
try {
final PipePluginMeta pipePluginMeta = createPipePluginPlan.getPipePluginMeta();
+ final String pluginName = pipePluginMeta.getPluginName();
// try to drop the old pipe plugin if exists to reduce the effect of the inconsistency
- dropPipePlugin(new DropPipePluginPlan(pipePluginMeta.getPluginName()));
+ dropPipePlugin(new DropPipePluginPlan(pluginName));
- pipePluginMetaKeeper.addPipePluginMeta(pipePluginMeta.getPluginName(), pipePluginMeta);
+ pipePluginMetaKeeper.addPipePluginMeta(pluginName, pipePluginMeta);
pipePluginMetaKeeper.addJarNameAndMd5(
pipePluginMeta.getJarName(), pipePluginMeta.getJarMD5());
if (createPipePluginPlan.getJarFile() != null) {
pipePluginExecutableManager.savePluginToInstallDir(
ByteBuffer.wrap(createPipePluginPlan.getJarFile().getValues()),
- createPipePluginPlan.getPipePluginMeta().getPluginName(),
+ pluginName,
pipePluginMeta.getJarName());
+ final String pluginDirPath = pipePluginExecutableManager.getPluginsDirPath(pluginName);
+ final PipePluginClassLoader pipePluginClassLoader =
+ classLoaderManager.createPipePluginClassLoader(pluginDirPath);
+ final Class<?> pluginClass =
+ Class.forName(pipePluginMeta.getClassName(), true, pipePluginClassLoader);
+ pipePluginMetaKeeper.addPipePluginVisibility(
+ pluginName, VisibilityUtils.calculateFromPluginClass(pluginClass));
+ classLoaderManager.addPluginAndClassLoader(pluginName, pipePluginClassLoader);
}
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
@@ -227,9 +254,11 @@ public class PipePluginInfo implements SnapshotProcessor {
String jarName = pipePluginMetaKeeper.getPipePluginMeta(pluginName).getJarName();
pipePluginMetaKeeper.removeJarNameAndMd5IfPossible(jarName);
pipePluginMetaKeeper.removePipePluginMeta(pluginName);
+ pipePluginMetaKeeper.removePipePluginVisibility(pluginName);
try {
pipePluginExecutableManager.removePluginFileUnderLibRoot(pluginName, jarName);
+ classLoaderManager.removePluginClassLoader(pluginName);
} catch (IOException e) {
final String errorMessage =
String.format(
@@ -247,7 +276,9 @@ public class PipePluginInfo implements SnapshotProcessor {
public DataSet showPipePlugins() {
return new PipePluginTableResp(
new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()),
- Arrays.asList(pipePluginMetaKeeper.getAllPipePluginMeta()));
+ StreamSupport.stream(pipePluginMetaKeeper.getAllPipePluginMeta().spliterator(), false)
+ .collect(Collectors.toList()),
+ pipePluginMetaKeeper.getPipePluginNameToVisibilityMap());
}
public JarResp getPipePluginJar(final GetPipePluginJarPlan getPipePluginJarPlan) {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
index 8fe6659dc0c..44562aa8bbb 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
@@ -440,9 +440,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
public TSStatus createPipe(final CreatePipePlanV2 plan) {
acquireWriteLock();
try {
- pipeMetaKeeper.addPipeMeta(
- plan.getPipeStaticMeta().getPipeName(),
- new PipeMeta(plan.getPipeStaticMeta(), plan.getPipeRuntimeMeta()));
+ pipeMetaKeeper.addPipeMeta(new PipeMeta(plan.getPipeStaticMeta(), plan.getPipeRuntimeMeta()));
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
} finally {
releaseWriteLock();
@@ -502,7 +500,6 @@ public class PipeTaskInfo implements SnapshotProcessor {
pipeMetaKeeper.getPipeMeta(plan.getPipeStaticMeta().getPipeName()).getTemporaryMeta();
pipeMetaKeeper.removePipeMeta(plan.getPipeStaticMeta().getPipeName());
pipeMetaKeeper.addPipeMeta(
- plan.getPipeStaticMeta().getPipeName(),
new PipeMeta(plan.getPipeStaticMeta(), plan.getPipeRuntimeMeta(), temporaryMeta));
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
} finally {
@@ -663,7 +660,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
plan.getPipeMetaList()
.forEach(
pipeMeta -> {
- pipeMetaKeeper.addPipeMeta(pipeMeta.getStaticMeta().getPipeName(), pipeMeta);
+ pipeMetaKeeper.addPipeMeta(pipeMeta);
logger.ifPresent(l -> l.info("Recording pipe meta: {}", pipeMeta));
});
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index 219b68813e6..c434d44aec7 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -186,6 +186,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowModelReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowModelResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowPipePluginReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq;
@@ -905,6 +906,11 @@ public class ConfigNodeRPCServiceProcessor implements IConfigNodeRPCService.Ifac
return configManager.getPipePluginTable();
}
+ @Override
+ public TGetPipePluginTableResp getPipePluginTableExtended(TShowPipePluginReq req) {
+ return configManager.getPipePluginTableExtended(req);
+ }
+
@Override
public TGetJarInListResp getPipePluginJar(TGetJarInListReq req) {
return configManager.getPipePluginJar(req);
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipePluginTableRespTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipePluginTableRespTest.java
index 45454d56ca5..2dc09056bea 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipePluginTableRespTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipePluginTableRespTest.java
@@ -32,6 +32,7 @@ import org.junit.Test;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
public class PipePluginTableRespTest {
@@ -43,7 +44,8 @@ public class PipePluginTableRespTest {
pipePluginMetaList.add(new PipePluginMeta("iotdb-extractor", IoTDBExtractor.class.getName()));
pipePluginMetaList.add(
new PipePluginMeta("do-nothing-processor", DoNothingProcessor.class.getName()));
- PipePluginTableResp pipePluginTableResp = new PipePluginTableResp(status, pipePluginMetaList);
+ PipePluginTableResp pipePluginTableResp =
+ new PipePluginTableResp(status, pipePluginMetaList, Collections.emptyMap());
final List<ByteBuffer> pipePluginByteBuffers = new ArrayList<>();
for (PipePluginMeta pipePluginMeta : pipePluginMetaList) {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/pipe/annotation/PipePluginAnnotationTest.java
similarity index 59%
copy from iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java
copy to iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/pipe/annotation/PipePluginAnnotationTest.java
index 8d562f7552c..1cd3adb8f0e 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/pipe/annotation/PipePluginAnnotationTest.java
@@ -17,16 +17,23 @@
* under the License.
*/
-package org.apache.iotdb.commons.pipe.datastructure;
+package org.apache.iotdb.confignode.pipe.annotation;
-import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import org.apache.iotdb.commons.pipe.datastructure.result.Result;
+import org.apache.iotdb.commons.pipe.datastructure.visibility.VisibilityTestUtils;
-public interface PersistentResource {
- ProgressIndex getProgressIndex();
+import org.junit.Test;
- /** Used in PipeHistoricalDataRegionTsFileAndDeletionExtractor to sort PersistentResource */
- long getFileStartTime();
+import static org.junit.Assert.fail;
- /** Used in PipeHistoricalDataRegionTsFileAndDeletionExtractor to sort PersistentResource */
- long getFileEndTime();
+public class PipePluginAnnotationTest {
+
+ @Test
+ public void testPipePluginVisibility() {
+ final Result<Void, String> result =
+ VisibilityTestUtils.testVisibilityCompatibilityEntry("org.apache.iotdb.confignode");
+ if (result.isErr()) {
+ fail(result.getErr());
+ }
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeDataNodePluginAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeDataNodePluginAgent.java
index 31f404de9d2..088d19966b4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeDataNodePluginAgent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeDataNodePluginAgent.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.commons.pipe.agent.plugin.meta.PipePluginMeta;
import org.apache.iotdb.commons.pipe.agent.plugin.service.PipePluginClassLoader;
import org.apache.iotdb.commons.pipe.agent.plugin.service.PipePluginClassLoaderManager;
import org.apache.iotdb.commons.pipe.agent.plugin.service.PipePluginExecutableManager;
+import org.apache.iotdb.commons.pipe.datastructure.visibility.VisibilityUtils;
import org.apache.iotdb.db.pipe.agent.plugin.dataregion.PipeDataRegionPluginAgent;
import org.apache.iotdb.db.pipe.agent.plugin.schemaregion.PipeSchemaRegionPluginAgent;
import org.apache.iotdb.pipe.api.PipePlugin;
@@ -151,6 +152,8 @@ public class PipeDataNodePluginAgent {
final PipePlugin ignored = (PipePlugin) pluginClass.getDeclaredConstructor().newInstance();
pipePluginMetaKeeper.addPipePluginMeta(pluginName, pipePluginMeta);
+ pipePluginMetaKeeper.addPipePluginVisibility(
+ pluginName, VisibilityUtils.calculateFromPluginClass(pluginClass));
classLoaderManager.addPluginAndClassLoader(pluginName, pipePluginClassLoader);
} catch (IOException
| InstantiationException
@@ -183,6 +186,7 @@ public class PipeDataNodePluginAgent {
// remove anyway
pipePluginMetaKeeper.removePipePluginMeta(pluginName);
+ pipePluginMetaKeeper.removePipePluginVisibility(pluginName);
PipePluginClassLoaderManager.getInstance().removePluginClassLoader(pluginName);
// if it is needed to delete jar file of the pipe plugin, delete both jar file and md5
@@ -211,20 +215,4 @@ public class PipeDataNodePluginAgent {
schemaRegionAgent.validate(
pipeName, extractorAttributes, processorAttributes, connectorAttributes);
}
-
- public void validateExtractor(Map<String, String> extractorAttributes) throws Exception {
- dataRegionAgent.validateExtractor(extractorAttributes);
- schemaRegionAgent.validateExtractor(extractorAttributes);
- }
-
- public void validateProcessor(Map<String, String> processorAttributes) throws Exception {
- dataRegionAgent.validateProcessor(processorAttributes);
- schemaRegionAgent.validateProcessor(processorAttributes);
- }
-
- public void validateConnector(String pipeName, Map<String, String> connectorAttributes)
- throws Exception {
- dataRegionAgent.validateConnector(pipeName, connectorAttributes);
- schemaRegionAgent.validateConnector(pipeName, connectorAttributes);
- }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionPluginAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionPluginAgent.java
index 22d9a75fba1..43326eb72a7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionPluginAgent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionPluginAgent.java
@@ -25,6 +25,15 @@ import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeExtractorConst
import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeProcessorConstructor;
import org.apache.iotdb.commons.pipe.agent.plugin.meta.DataNodePipePluginMetaKeeper;
import org.apache.iotdb.commons.pipe.agent.plugin.meta.PipePluginMetaKeeper;
+import org.apache.iotdb.commons.pipe.datastructure.visibility.Visibility;
+import org.apache.iotdb.commons.pipe.datastructure.visibility.VisibilityUtils;
+import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.PipeExtractor;
+import org.apache.iotdb.pipe.api.PipeProcessor;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException;
+
+import java.util.Map;
public class PipeDataRegionPluginAgent extends PipePluginAgent {
@@ -52,4 +61,44 @@ public class PipeDataRegionPluginAgent extends PipePluginAgent {
return new PipeDataRegionConnectorConstructor(
(DataNodePipePluginMetaKeeper) pipePluginMetaKeeper);
}
+
+ @Override
+ public void validate(
+ String pipeName,
+ Map<String, String> extractorAttributes,
+ Map<String, String> processorAttributes,
+ Map<String, String> connectorAttributes)
+ throws Exception {
+ PipeExtractor temporaryExtractor = validateExtractor(extractorAttributes);
+ PipeProcessor temporaryProcessor = validateProcessor(processorAttributes);
+ PipeConnector temporaryConnector = validateConnector(pipeName, connectorAttributes);
+
+ // validate visibility
+ // TODO: validate visibility for schema region and config region
+ Visibility pipeVisibility =
+ VisibilityUtils.calculateFromExtractorParameters(new PipeParameters(extractorAttributes));
+ Visibility extractorVisibility =
+ VisibilityUtils.calculateFromPluginClass(temporaryExtractor.getClass());
+ Visibility processorVisibility =
+ VisibilityUtils.calculateFromPluginClass(temporaryProcessor.getClass());
+ Visibility connectorVisibility =
+ VisibilityUtils.calculateFromPluginClass(temporaryConnector.getClass());
+ if (!VisibilityUtils.isCompatible(
+ pipeVisibility, extractorVisibility, processorVisibility, connectorVisibility)) {
+ throw new PipeParameterNotValidException(
+ String.format(
+ "The visibility of the pipe (%s, %s) is not compatible with the visibility of the extractor (%s, %s, %s), processor (%s, %s, %s), and connector (%s, %s, %s).",
+ pipeName,
+ pipeVisibility,
+ extractorAttributes,
+ temporaryExtractor.getClass().getName(),
+ extractorVisibility,
+ processorAttributes,
+ temporaryProcessor.getClass().getName(),
+ processorVisibility,
+ connectorAttributes,
+ temporaryConnector.getClass().getName(),
+ connectorVisibility));
+ }
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBDataNodeAirGapConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBDataNodeAirGapConnector.java
index 46f8a7383f6..591db25fc6a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBDataNodeAirGapConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBDataNodeAirGapConnector.java
@@ -28,6 +28,8 @@ import org.apache.iotdb.commons.utils.NodeUrlUtils;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferDataNodeHandshakeV1Req;
import org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferDataNodeHandshakeV2Req;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.slf4j.Logger;
@@ -39,6 +41,8 @@ import java.util.HashMap;
import java.util.Set;
import java.util.stream.Collectors;
+@TreeModel
+@TableModel
public abstract class IoTDBDataNodeAirGapConnector extends IoTDBAirGapConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBDataNodeAirGapConnector.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBDataRegionAirGapConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBDataRegionAirGapConnector.java
index ec92b2e9ddd..0f2e47dab0c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBDataRegionAirGapConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBDataRegionAirGapConnector.java
@@ -36,6 +36,8 @@ import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
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;
@@ -50,6 +52,8 @@ import java.io.File;
import java.io.IOException;
import java.util.Objects;
+@TreeModel
+@TableModel
public class IoTDBDataRegionAirGapConnector extends IoTDBDataNodeAirGapConnector {
private static final Logger LOGGER =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBSchemaRegionAirGapConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBSchemaRegionAirGapConnector.java
index 61c1a42d133..22a127905ef 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBSchemaRegionAirGapConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBSchemaRegionAirGapConnector.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransfer
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionSnapshotEvent;
import org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionWritePlanEvent;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
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;
@@ -41,6 +43,8 @@ import java.io.File;
import java.io.IOException;
import java.util.Objects;
+@TreeModel
+@TableModel
public class IoTDBSchemaRegionAirGapConnector extends IoTDBDataNodeAirGapConnector {
private static final Logger LOGGER =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
index f4bc9cc4c6e..0327b665107 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
@@ -39,6 +39,7 @@ import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -91,6 +92,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstan
import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_USERNAME_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_USER_KEY;
+@TreeModel
public class IoTDBLegacyPipeConnector implements PipeConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBLegacyPipeConnector.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaConnector.java
index ae96b7f5916..d9e5ed979f8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaConnector.java
@@ -26,6 +26,8 @@ import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -81,6 +83,8 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstan
* tablets, then eventNodes to send to the subscriber clients. Notice that there is no namespace
* since the eventNodes do not need to be saved.
*/
+@TreeModel
+@TableModel
public class OpcUaConnector implements PipeConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(OpcUaConnector.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java
index c0fa792da79..3573440472a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java
@@ -52,6 +52,8 @@ import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertio
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.queryengine.plan.planner.plan.node.write.InsertNode;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -77,6 +79,8 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstan
// TODO: Optimize the network and disk io for TsFile onComplete
// TODO: support Tablet Batch
+@TreeModel
+@TableModel
public class PipeConsensusAsyncConnector extends IoTDBConnector implements ConsensusPipeConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeConsensusAsyncConnector.class);
private static final String ENQUEUE_EXCEPTION_MSG =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusSyncConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusSyncConnector.java
index c00527ec38d..01d3fbabcd1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusSyncConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusSyncConnector.java
@@ -46,6 +46,8 @@ import org.apache.iotdb.db.pipe.event.common.deletion.PipeDeleteDataNodeEvent;
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.queryengine.plan.planner.plan.node.write.InsertNode;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
@@ -66,6 +68,8 @@ import java.util.List;
import java.util.stream.Collectors;
/** This connector is used for PipeConsensus to transfer queued event. */
+@TreeModel
+@TableModel
public class PipeConsensusSyncConnector extends IoTDBConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeConsensusSyncConnector.class);
private static final String PIPE_CONSENSUS_SYNC_CONNECTION_FAILED_FORMAT =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBDataRegionAsyncConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBDataRegionAsyncConnector.java
index 1dfebff416c..72b6f29494f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBDataRegionAsyncConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBDataRegionAsyncConnector.java
@@ -46,6 +46,8 @@ import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -80,6 +82,8 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstan
import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_LEADER_CACHE_ENABLE_KEY;
+@TreeModel
+@TableModel
public class IoTDBDataRegionAsyncConnector extends IoTDBConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBDataRegionAsyncConnector.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBDataNodeSyncConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBDataNodeSyncConnector.java
index 03754dff436..d3386a3eb93 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBDataNodeSyncConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBDataNodeSyncConnector.java
@@ -26,6 +26,8 @@ import org.apache.iotdb.commons.utils.NodeUrlUtils;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.connector.client.IoTDBDataNodeSyncClientManager;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.slf4j.Logger;
@@ -37,6 +39,8 @@ import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
+@TreeModel
+@TableModel
public abstract class IoTDBDataNodeSyncConnector extends IoTDBSslSyncConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBDataNodeSyncConnector.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBDataRegionSyncConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBDataRegionSyncConnector.java
index 99b2e71c731..56816220a81 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBDataRegionSyncConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBDataRegionSyncConnector.java
@@ -44,6 +44,8 @@ import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
@@ -69,6 +71,8 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
+@TreeModel
+@TableModel
public class IoTDBDataRegionSyncConnector extends IoTDBDataNodeSyncConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBDataRegionSyncConnector.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBSchemaRegionConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBSchemaRegionConnector.java
index e7a1a0e6d5b..e235f225ec3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBSchemaRegionConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBSchemaRegionConnector.java
@@ -28,6 +28,8 @@ import org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransfer
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionSnapshotEvent;
import org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionWritePlanEvent;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
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;
@@ -46,6 +48,8 @@ import java.io.IOException;
import java.util.Collections;
import java.util.Objects;
+@TreeModel
+@TableModel
public class IoTDBSchemaRegionConnector extends IoTDBDataNodeSyncConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBSchemaRegionConnector.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/websocket/WebSocketConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/websocket/WebSocketConnector.java
index cf258452346..a9ed5cb46a7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/websocket/WebSocketConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/websocket/WebSocketConnector.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertio
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -39,6 +40,7 @@ import org.slf4j.LoggerFactory;
import java.util.Arrays;
import java.util.Optional;
+@TreeModel
public class WebSocketConnector implements PipeConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketConnector.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/writeback/WriteBackConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/writeback/WriteBackConnector.java
index 84a82bc052f..44e4f06d60f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/writeback/WriteBackConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/writeback/WriteBackConnector.java
@@ -33,6 +33,8 @@ import org.apache.iotdb.db.queryengine.plan.Coordinator;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException;
import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeRuntimeEnvironment;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -48,6 +50,8 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Objects;
+@TreeModel
+@TableModel
public class WriteBackConnector implements PipeConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(WriteBackConnector.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResource.java
index afb34bfff6b..4755b2b962a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResource.java
@@ -23,7 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex;
-import org.apache.iotdb.commons.pipe.datastructure.PersistentResource;
+import org.apache.iotdb.commons.pipe.datastructure.resource.PersistentResource;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.consensus.DataRegionConsensusImpl;
import org.apache.iotdb.db.pipe.event.common.deletion.PipeDeleteDataNodeEvent;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java
index 0a033aff297..d4ce3e7663d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java
@@ -41,6 +41,8 @@ import org.apache.iotdb.db.pipe.metric.PipeDataNodeRemainingEventAndTimeMetrics;
import org.apache.iotdb.db.pipe.metric.PipeDataRegionExtractorMetrics;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -111,6 +113,8 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstan
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant._EXTRACTOR_WATERMARK_INTERVAL_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant._SOURCE_WATERMARK_INTERVAL_KEY;
+@TreeModel
+@TableModel
public class IoTDBDataRegionExtractor extends IoTDBExtractor {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBDataRegionExtractor.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java
index dfaf71a218d..549d338a123 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java
@@ -28,9 +28,9 @@ import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskExtractorRuntimeEnvironment;
-import org.apache.iotdb.commons.pipe.datastructure.PersistentResource;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
+import org.apache.iotdb.commons.pipe.datastructure.resource.PersistentResource;
import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResource;
import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResourceManager;
import org.apache.iotdb.db.pipe.event.common.deletion.PipeDeleteDataNodeEvent;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/IoTDBSchemaRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/IoTDBSchemaRegionExtractor.java
index 7036e3d0d50..208b32f340c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/IoTDBSchemaRegionExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/IoTDBSchemaRegionExtractor.java
@@ -39,6 +39,8 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.AlterTimeSeriesNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeOperateSchemaQueueNode;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.exception.PipeException;
@@ -48,6 +50,8 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Set;
+@TreeModel
+@TableModel
public class IoTDBSchemaRegionExtractor extends IoTDBNonDataRegionExtractor {
public static final PipePlanTreePatternParseVisitor TREE_PATTERN_PARSE_VISITOR =
new PipePlanTreePatternParseVisitor();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java
index 7c2de9ae6d2..15b38dacf72 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java
@@ -46,6 +46,7 @@ import org.apache.iotdb.db.queryengine.transformation.dag.udf.UDFParametersFacto
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.pipe.api.PipeProcessor;
import org.apache.iotdb.pipe.api.access.Row;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.collector.RowCollector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
@@ -104,6 +105,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstan
* AbstractWindowingProcessor} and {@link AbstractOperatorProcessor} can be implemented by user and
* loaded as a normal {@link PipeProcessor}
*/
+@TreeModel
public class AggregateProcessor implements PipeProcessor {
private static final String WINDOWING_PROCESSOR_SUFFIX = "-windowing-processor";
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/operator/processor/StandardStatisticsOperatorProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/operator/processor/StandardStatisticsOperatorProcessor.java
index fe010a0a1a1..2f92687414f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/operator/processor/StandardStatisticsOperatorProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/operator/processor/StandardStatisticsOperatorProcessor.java
@@ -35,6 +35,7 @@ import org.apache.iotdb.db.pipe.processor.aggregate.operator.intermediateresult.
import org.apache.iotdb.db.pipe.processor.aggregate.operator.intermediateresult.sametype.numeric.IntegralPoweredSumOperator;
import org.apache.iotdb.db.pipe.processor.aggregate.operator.intermediateresult.specifictype.doubletype.FractionPoweredSumOperator;
import org.apache.iotdb.db.pipe.processor.aggregate.operator.intermediateresult.specifictype.integertype.CountOperator;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import java.util.Arrays;
import java.util.Collections;
@@ -42,6 +43,7 @@ import java.util.HashSet;
import java.util.Set;
import java.util.function.Supplier;
+@TreeModel
public class StandardStatisticsOperatorProcessor extends AbstractOperatorProcessor {
@Override
public Set<AggregatedResultOperator> getAggregatorOperatorSet() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/window/processor/TumblingWindowingProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/window/processor/TumblingWindowingProcessor.java
index cfaca460623..27828cba443 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/window/processor/TumblingWindowingProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/window/processor/TumblingWindowingProcessor.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.db.pipe.processor.aggregate.window.datastructure.WindowO
import org.apache.iotdb.db.pipe.processor.aggregate.window.datastructure.WindowState;
import org.apache.iotdb.db.utils.DateTimeUtils;
import org.apache.iotdb.db.utils.TimestampPrecisionUtils;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -40,6 +41,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstan
import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_SLIDING_SECONDS_DEFAULT_VALUE;
import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_SLIDING_SECONDS_KEY;
+@TreeModel
public class TumblingWindowingProcessor extends AbstractSimpleTimeWindowingProcessor {
private long slidingBoundaryTime;
private long slidingInterval;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/changing/ChangingValueSamplingProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/changing/ChangingValueSamplingProcessor.java
index 6badb70755b..0ab08ecb403 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/changing/ChangingValueSamplingProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/changing/ChangingValueSamplingProcessor.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.db.pipe.event.common.row.PipeRow;
import org.apache.iotdb.db.pipe.processor.downsampling.DownSamplingProcessor;
import org.apache.iotdb.db.pipe.processor.downsampling.PartialPathLastObjectCache;
import org.apache.iotdb.pipe.api.access.Row;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.collector.RowCollector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -37,6 +38,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicReference;
+@TreeModel
public class ChangingValueSamplingProcessor extends DownSamplingProcessor {
private static final Logger LOGGER =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/sdt/SwingingDoorTrendingSamplingProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/sdt/SwingingDoorTrendingSamplingProcessor.java
index 18a453e2e8d..f7e90f749c8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/sdt/SwingingDoorTrendingSamplingProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/sdt/SwingingDoorTrendingSamplingProcessor.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.db.pipe.event.common.row.PipeRow;
import org.apache.iotdb.db.pipe.processor.downsampling.DownSamplingProcessor;
import org.apache.iotdb.db.pipe.processor.downsampling.PartialPathLastObjectCache;
import org.apache.iotdb.pipe.api.access.Row;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.collector.RowCollector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -37,6 +38,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicReference;
+@TreeModel
public class SwingingDoorTrendingSamplingProcessor extends DownSamplingProcessor {
private static final Logger LOGGER =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/tumbling/TumblingTimeSamplingProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/tumbling/TumblingTimeSamplingProcessor.java
index 665f5781801..a7411aada3e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/tumbling/TumblingTimeSamplingProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/tumbling/TumblingTimeSamplingProcessor.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.db.pipe.processor.downsampling.DownSamplingProcessor;
import org.apache.iotdb.db.pipe.processor.downsampling.PartialPathLastObjectCache;
import org.apache.iotdb.db.utils.TimestampPrecisionUtils;
import org.apache.iotdb.pipe.api.access.Row;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.collector.RowCollector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -40,6 +41,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstan
import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_TUMBLING_TIME_INTERVAL_SECONDS_DEFAULT_VALUE;
import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_TUMBLING_TIME_INTERVAL_SECONDS_KEY;
+@TreeModel
public class TumblingTimeSamplingProcessor extends DownSamplingProcessor {
private static final Logger LOGGER = LoggerFactory.getLogger(TumblingTimeSamplingProcessor.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/pipeconsensus/PipeConsensusProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/pipeconsensus/PipeConsensusProcessor.java
index bbb02084514..c34033aa03c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/pipeconsensus/PipeConsensusProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/pipeconsensus/PipeConsensusProcessor.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.pipe.api.PipeProcessor;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -37,6 +39,8 @@ import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
import java.util.Map;
+@TreeModel
+@TableModel
public class PipeConsensusProcessor implements PipeProcessor {
private static final int DATA_NODE_ID = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/schemachange/RenameDatabaseProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/schemachange/RenameDatabaseProcessor.java
index 1d611a440d1..c4b798dfcb5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/schemachange/RenameDatabaseProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/schemachange/RenameDatabaseProcessor.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
import org.apache.iotdb.db.queryengine.plan.execution.config.TableConfigTaskVisitor;
import org.apache.iotdb.pipe.api.PipeProcessor;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -36,6 +37,7 @@ import static org.apache.iotdb.commons.conf.IoTDBConstant.MAX_DATABASE_NAME_LENG
import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_RENAME_DATABASE_NEW_DB_NAME;
import static org.apache.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR;
+@TableModel
public class RenameDatabaseProcessor implements PipeProcessor {
// Currently this processor is only used for table model.
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/twostage/plugin/TwoStageCountProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/twostage/plugin/TwoStageCountProcessor.java
index 5595fbffd39..d47a270ac0c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/twostage/plugin/TwoStageCountProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/twostage/plugin/TwoStageCountProcessor.java
@@ -41,6 +41,7 @@ import org.apache.iotdb.db.pipe.processor.twostage.exchange.sender.TwoStageAggre
import org.apache.iotdb.db.pipe.processor.twostage.operator.CountOperator;
import org.apache.iotdb.db.pipe.processor.twostage.state.CountState;
import org.apache.iotdb.pipe.api.PipeProcessor;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -73,6 +74,7 @@ import java.util.concurrent.atomic.AtomicReference;
import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_OUTPUT_SERIES_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant._PROCESSOR_OUTPUT_SERIES_KEY;
+@TreeModel
public class TwoStageCountProcessor implements PipeProcessor {
private static final Logger LOGGER = LoggerFactory.getLogger(TwoStageCountProcessor.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
index f50fbfd63bd..4268a7ffc06 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
@@ -149,6 +149,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowModelReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowModelResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowPipePluginReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq;
@@ -959,6 +960,13 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie
() -> client.getPipePluginTable(), resp -> !updateConfigNodeLeader(resp.status));
}
+ @Override
+ public TGetPipePluginTableResp getPipePluginTableExtended(TShowPipePluginReq req)
+ throws TException {
+ return executeRemoteCallWithRetry(
+ () -> client.getPipePluginTableExtended(req), resp -> !updateConfigNodeLeader(resp.status));
+ }
+
@Override
public TGetJarInListResp getPipePluginJar(TGetJarInListReq req) throws TException {
return executeRemoteCallWithRetry(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
index 19315e6d8fc..e4c00286f56 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
@@ -769,7 +769,7 @@ public class TableConfigTaskVisitor extends AstVisitor<IConfigTask, MPPQueryCont
@Override
protected IConfigTask visitShowPipePlugins(ShowPipePlugins node, MPPQueryContext context) {
context.setQueryType(QueryType.READ);
- return new ShowPipePluginsTask();
+ return new ShowPipePluginsTask(node);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TreeConfigTaskVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TreeConfigTaskVisitor.java
index a1ee08d84e4..bd674f4f18d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TreeConfigTaskVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TreeConfigTaskVisitor.java
@@ -397,7 +397,7 @@ public class TreeConfigTaskVisitor extends StatementVisitor<IConfigTask, MPPQuer
@Override
public IConfigTask visitShowPipePlugins(
ShowPipePluginsStatement showPipePluginStatement, MPPQueryContext context) {
- return new ShowPipePluginsTask();
+ return new ShowPipePluginsTask(showPipePluginStatement);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index 3c41ea2ac2e..4fa66c90a42 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -121,6 +121,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowModelReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowModelResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TShowPipePluginReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp;
@@ -229,6 +230,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.CreatePipePl
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.CreatePipeStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.DropPipePluginStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.DropPipeStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.ShowPipePluginsStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.ShowPipesStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.StartPipeStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.StopPipeStatement;
@@ -995,7 +997,8 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
client.dropPipePlugin(
new TDropPipePluginReq()
.setPluginName(dropPipePluginStatement.getPluginName())
- .setIfExistsCondition(dropPipePluginStatement.hasIfExistsCondition()));
+ .setIfExistsCondition(dropPipePluginStatement.hasIfExistsCondition())
+ .setIsTableModel(dropPipePluginStatement.isTableModel()));
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != executionStatus.getCode()) {
LOGGER.warn(
"[{}] Failed to drop pipe plugin {}.",
@@ -1012,11 +1015,14 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
}
@Override
- public SettableFuture<ConfigTaskResult> showPipePlugins() {
+ public SettableFuture<ConfigTaskResult> showPipePlugins(
+ ShowPipePluginsStatement showPipePluginsStatement) {
final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
try (final ConfigNodeClient client =
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
- TGetPipePluginTableResp getPipePluginTableResp = client.getPipePluginTable();
+ TGetPipePluginTableResp getPipePluginTableResp =
+ client.getPipePluginTableExtended(
+ new TShowPipePluginReq().setIsTableModel(showPipePluginsStatement.isTableModel()));
if (getPipePluginTableResp.getStatus().getCode()
!= TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
future.setException(
@@ -1981,53 +1987,63 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
// Construct temporary pipe static meta for validation
final String pipeName = alterPipeStatement.getPipeName();
+ final Map<String, String> extractorAttributes;
+ final Map<String, String> processorAttributes;
+ final Map<String, String> connectorAttributes;
try {
if (!alterPipeStatement.getExtractorAttributes().isEmpty()) {
if (alterPipeStatement.isReplaceAllExtractorAttributes()) {
- PipeDataNodeAgent.plugin().validateExtractor(alterPipeStatement.getExtractorAttributes());
+ extractorAttributes = alterPipeStatement.getExtractorAttributes();
} else {
pipeMetaFromCoordinator
.getStaticMeta()
.getExtractorParameters()
.addOrReplaceEquivalentAttributes(
new PipeParameters(alterPipeStatement.getExtractorAttributes()));
- PipeDataNodeAgent.plugin()
- .validateExtractor(
- pipeMetaFromCoordinator.getStaticMeta().getExtractorParameters().getAttribute());
+ extractorAttributes =
+ pipeMetaFromCoordinator.getStaticMeta().getExtractorParameters().getAttribute();
}
+ } else {
+ extractorAttributes =
+ pipeMetaFromCoordinator.getStaticMeta().getExtractorParameters().getAttribute();
}
if (!alterPipeStatement.getProcessorAttributes().isEmpty()) {
if (alterPipeStatement.isReplaceAllProcessorAttributes()) {
- PipeDataNodeAgent.plugin().validateProcessor(alterPipeStatement.getProcessorAttributes());
+ processorAttributes = alterPipeStatement.getProcessorAttributes();
} else {
pipeMetaFromCoordinator
.getStaticMeta()
.getProcessorParameters()
.addOrReplaceEquivalentAttributes(
new PipeParameters(alterPipeStatement.getProcessorAttributes()));
- PipeDataNodeAgent.plugin()
- .validateProcessor(
- pipeMetaFromCoordinator.getStaticMeta().getProcessorParameters().getAttribute());
+ processorAttributes =
+ pipeMetaFromCoordinator.getStaticMeta().getProcessorParameters().getAttribute();
}
+ } else {
+ processorAttributes =
+ pipeMetaFromCoordinator.getStaticMeta().getProcessorParameters().getAttribute();
}
if (!alterPipeStatement.getConnectorAttributes().isEmpty()) {
if (alterPipeStatement.isReplaceAllConnectorAttributes()) {
- PipeDataNodeAgent.plugin()
- .validateConnector(pipeName, alterPipeStatement.getConnectorAttributes());
+ connectorAttributes = alterPipeStatement.getConnectorAttributes();
} else {
pipeMetaFromCoordinator
.getStaticMeta()
.getConnectorParameters()
.addOrReplaceEquivalentAttributes(
new PipeParameters(alterPipeStatement.getConnectorAttributes()));
- PipeDataNodeAgent.plugin()
- .validateConnector(
- pipeName,
- pipeMetaFromCoordinator.getStaticMeta().getConnectorParameters().getAttribute());
+ connectorAttributes =
+ pipeMetaFromCoordinator.getStaticMeta().getConnectorParameters().getAttribute();
}
+ } else {
+ connectorAttributes =
+ pipeMetaFromCoordinator.getStaticMeta().getConnectorParameters().getAttribute();
}
+
+ PipeDataNodeAgent.plugin()
+ .validate(pipeName, extractorAttributes, processorAttributes, connectorAttributes);
} catch (final Exception e) {
future.setException(
new IoTDBException(e.getMessage(), TSStatusCode.PIPE_ERROR.getStatusCode()));
@@ -2251,9 +2267,11 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
new TopicMeta(topicName, System.currentTimeMillis(), topicAttributes);
try {
PipeDataNodeAgent.plugin()
- .validateExtractor(temporaryTopicMeta.generateExtractorAttributes());
- PipeDataNodeAgent.plugin()
- .validateProcessor(temporaryTopicMeta.generateProcessorAttributes());
+ .validate(
+ "fakePipeName",
+ temporaryTopicMeta.generateExtractorAttributes(),
+ temporaryTopicMeta.generateProcessorAttributes(),
+ temporaryTopicMeta.generateConnectorAttributes("fakeConsumerGroupId"));
} catch (final Exception e) {
future.setException(
new IoTDBException(e.getMessage(), TSStatusCode.CREATE_TOPIC_ERROR.getStatusCode()));
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java
index 179d2455037..9aac3ffaddd 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java
@@ -63,6 +63,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.CreatePipePl
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.CreatePipeStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.DropPipePluginStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.DropPipeStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.ShowPipePluginsStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.ShowPipesStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.StartPipeStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.StopPipeStatement;
@@ -133,7 +134,8 @@ public interface IConfigTaskExecutor {
SettableFuture<ConfigTaskResult> dropPipePlugin(DropPipePluginStatement dropPipePluginStatement);
- SettableFuture<ConfigTaskResult> showPipePlugins();
+ SettableFuture<ConfigTaskResult> showPipePlugins(
+ ShowPipePluginsStatement showPipePluginsStatement);
SettableFuture<ConfigTaskResult> setTTL(SetTTLStatement setTTLStatement, String taskName);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/DropPipePluginTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/DropPipePluginTask.java
index c014809aad2..7cd106e1713 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/DropPipePluginTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/DropPipePluginTask.java
@@ -31,18 +31,19 @@ public class DropPipePluginTask implements IConfigTask {
private final DropPipePluginStatement dropPipePluginStatement;
- public DropPipePluginTask(DropPipePluginStatement dropPipePluginStatement) {
+ public DropPipePluginTask(final DropPipePluginStatement dropPipePluginStatement) {
this.dropPipePluginStatement = dropPipePluginStatement;
}
- public DropPipePluginTask(DropPipePlugin node) {
- dropPipePluginStatement = new DropPipePluginStatement();
- dropPipePluginStatement.setPluginName(node.getPluginName());
- dropPipePluginStatement.setIfExists(node.hasIfExistsCondition());
+ public DropPipePluginTask(final DropPipePlugin node) {
+ this.dropPipePluginStatement = new DropPipePluginStatement();
+ this.dropPipePluginStatement.setPluginName(node.getPluginName());
+ this.dropPipePluginStatement.setIfExists(node.hasIfExistsCondition());
+ this.dropPipePluginStatement.setTableModel(true);
}
@Override
- public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
+ public ListenableFuture<ConfigTaskResult> execute(final IConfigTaskExecutor configTaskExecutor)
throws InterruptedException {
return configTaskExecutor.dropPipePlugin(dropPipePluginStatement);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowPipePluginsTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowPipePluginsTask.java
index fc2b47aa689..3292d838e20 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowPipePluginsTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowPipePluginsTask.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.db.queryengine.common.header.DatasetHeaderFactory;
import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
import org.apache.iotdb.db.queryengine.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.executor.IConfigTaskExecutor;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowPipePlugins;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.ShowPipePluginsStatement;
import org.apache.iotdb.rpc.TSStatusCode;
import com.google.common.util.concurrent.ListenableFuture;
@@ -49,14 +51,26 @@ public class ShowPipePluginsTask implements IConfigTask {
private static final Binary PIPE_JAR_NAME_EMPTY_FIELD = BytesUtils.valueOf("");
+ private final ShowPipePluginsStatement showPipePluginsStatement;
+
+ public ShowPipePluginsTask(final ShowPipePluginsStatement showPipePluginsStatement) {
+ this.showPipePluginsStatement = showPipePluginsStatement;
+ }
+
+ public ShowPipePluginsTask(final ShowPipePlugins node) {
+ this.showPipePluginsStatement = new ShowPipePluginsStatement();
+ this.showPipePluginsStatement.setTableModel(true);
+ }
+
@Override
- public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
+ public ListenableFuture<ConfigTaskResult> execute(final IConfigTaskExecutor configTaskExecutor)
throws InterruptedException {
- return configTaskExecutor.showPipePlugins();
+ return configTaskExecutor.showPipePlugins(showPipePluginsStatement);
}
public static void buildTsBlock(
- List<ByteBuffer> allPipePluginsInformation, SettableFuture<ConfigTaskResult> future) {
+ final List<ByteBuffer> allPipePluginsInformation,
+ final SettableFuture<ConfigTaskResult> future) {
final List<PipePluginMeta> pipePluginMetaList = new ArrayList<>();
if (allPipePluginsInformation != null) {
for (final ByteBuffer pipePluginInformationByteBuffer : allPipePluginsInformation) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/pipe/DropPipePluginStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/pipe/DropPipePluginStatement.java
index c7f4ebfd5df..f5c140c1cf3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/pipe/DropPipePluginStatement.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/pipe/DropPipePluginStatement.java
@@ -37,6 +37,7 @@ public class DropPipePluginStatement extends Statement implements IConfigStateme
private String pluginName;
private boolean ifExistsCondition;
+ private boolean isTableModel;
public DropPipePluginStatement() {
super();
@@ -51,14 +52,22 @@ public class DropPipePluginStatement extends Statement implements IConfigStateme
return ifExistsCondition;
}
- public void setPluginName(String pluginName) {
+ public boolean isTableModel() {
+ return isTableModel;
+ }
+
+ public void setPluginName(final String pluginName) {
this.pluginName = pluginName;
}
- public void setIfExists(boolean ifExistsCondition) {
+ public void setIfExists(final boolean ifExistsCondition) {
this.ifExistsCondition = ifExistsCondition;
}
+ public void setTableModel(final boolean tableModel) {
+ this.isTableModel = tableModel;
+ }
+
@Override
public QueryType getQueryType() {
return QueryType.WRITE;
@@ -70,12 +79,12 @@ public class DropPipePluginStatement extends Statement implements IConfigStateme
}
@Override
- public <R, C> R accept(StatementVisitor<R, C> visitor, C context) {
+ public <R, C> R accept(final StatementVisitor<R, C> visitor, final C context) {
return visitor.visitDropPipePlugin(this, context);
}
@Override
- public TSStatus checkPermissionBeforeProcess(String userName) {
+ public TSStatus checkPermissionBeforeProcess(final String userName) {
if (AuthorityChecker.SUPER_USER.equals(userName)) {
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/pipe/ShowPipePluginsStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/pipe/ShowPipePluginsStatement.java
index 351ab731754..3cf09eb2eb5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/pipe/ShowPipePluginsStatement.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/pipe/ShowPipePluginsStatement.java
@@ -21,7 +21,6 @@ package org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.auth.entity.PrivilegeType;
-import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.queryengine.plan.analyze.QueryType;
import org.apache.iotdb.db.queryengine.plan.statement.IConfigStatement;
@@ -30,18 +29,25 @@ import org.apache.iotdb.db.queryengine.plan.statement.StatementVisitor;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowStatement;
import org.apache.iotdb.rpc.TSStatusCode;
-import java.util.Collections;
-import java.util.List;
-
public class ShowPipePluginsStatement extends ShowStatement implements IConfigStatement {
+ private boolean isTableModel;
+
+ public boolean isTableModel() {
+ return isTableModel;
+ }
+
+ public void setTableModel(final boolean tableModel) {
+ this.isTableModel = tableModel;
+ }
+
public ShowPipePluginsStatement() {
super();
statementType = StatementType.SHOW_PIPEPLUGINS;
}
@Override
- public <R, C> R accept(StatementVisitor<R, C> visitor, C context) {
+ public <R, C> R accept(final StatementVisitor<R, C> visitor, final C context) {
return visitor.visitShowPipePlugins(this, context);
}
@@ -51,12 +57,7 @@ public class ShowPipePluginsStatement extends ShowStatement implements IConfigSt
}
@Override
- public List<PartialPath> getPaths() {
- return Collections.emptyList();
- }
-
- @Override
- public TSStatus checkPermissionBeforeProcess(String userName) {
+ public TSStatus checkPermissionBeforeProcess(final String userName) {
if (AuthorityChecker.SUPER_USER.equals(userName)) {
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
index d173de421e6..668679a8fe6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
@@ -24,7 +24,7 @@ import org.apache.iotdb.commons.consensus.index.ProgressIndexType;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.path.IFullPath;
import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.commons.pipe.datastructure.PersistentResource;
+import org.apache.iotdb.commons.pipe.datastructure.resource.PersistentResource;
import org.apache.iotdb.commons.utils.CommonDateTimeUtils;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.conf.IoTDBConfig;
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/plugin/annotation/PipePluginAnnotationTest.java
similarity index 60%
copy from iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java
copy to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/plugin/annotation/PipePluginAnnotationTest.java
index 8d562f7552c..a1c78e499d6 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/plugin/annotation/PipePluginAnnotationTest.java
@@ -17,16 +17,23 @@
* under the License.
*/
-package org.apache.iotdb.commons.pipe.datastructure;
+package org.apache.iotdb.db.pipe.plugin.annotation;
-import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import org.apache.iotdb.commons.pipe.datastructure.result.Result;
+import org.apache.iotdb.commons.pipe.datastructure.visibility.VisibilityTestUtils;
-public interface PersistentResource {
- ProgressIndex getProgressIndex();
+import org.junit.Test;
- /** Used in PipeHistoricalDataRegionTsFileAndDeletionExtractor to sort PersistentResource */
- long getFileStartTime();
+import static org.junit.Assert.fail;
- /** Used in PipeHistoricalDataRegionTsFileAndDeletionExtractor to sort PersistentResource */
- long getFileEndTime();
+public class PipePluginAnnotationTest {
+
+ @Test
+ public void testPipePluginVisibility() {
+ final Result<Void, String> result =
+ VisibilityTestUtils.testVisibilityCompatibilityEntry("org.apache.iotdb.db");
+ if (result.isErr()) {
+ fail(result.getErr());
+ }
+ }
}
diff --git a/iotdb-core/node-commons/pom.xml b/iotdb-core/node-commons/pom.xml
index 2243f340bfc..8700c4de34b 100644
--- a/iotdb-core/node-commons/pom.xml
+++ b/iotdb-core/node-commons/pom.xml
@@ -167,6 +167,10 @@
<groupId>com.github.luben</groupId>
<artifactId>zstd-jni</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.reflections</groupId>
+ <artifactId>reflections</artifactId>
+ </dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/PipePluginAgent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/PipePluginAgent.java
index 240ab8c44de..352c3be1283 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/PipePluginAgent.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/PipePluginAgent.java
@@ -37,12 +37,12 @@ import org.apache.iotdb.pipe.api.exception.PipeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
public abstract class PipePluginAgent {
@@ -92,7 +92,8 @@ public abstract class PipePluginAgent {
validateConnector(pipeName, connectorAttributes);
}
- public void validateExtractor(Map<String, String> extractorAttributes) throws Exception {
+ protected PipeExtractor validateExtractor(Map<String, String> extractorAttributes)
+ throws Exception {
final PipeParameters extractorParameters = new PipeParameters(extractorAttributes);
final PipeExtractor temporaryExtractor = reflectExtractor(extractorParameters);
try {
@@ -104,9 +105,11 @@ public abstract class PipePluginAgent {
LOGGER.warn("Failed to close temporary extractor: {}", e.getMessage(), e);
}
}
+ return temporaryExtractor;
}
- public void validateProcessor(Map<String, String> processorAttributes) throws Exception {
+ protected PipeProcessor validateProcessor(Map<String, String> processorAttributes)
+ throws Exception {
final PipeParameters processorParameters = new PipeParameters(processorAttributes);
final PipeProcessor temporaryProcessor = reflectProcessor(processorParameters);
try {
@@ -118,10 +121,11 @@ public abstract class PipePluginAgent {
LOGGER.warn("Failed to close temporary processor: {}", e.getMessage(), e);
}
}
+ return temporaryProcessor;
}
- public void validateConnector(String pipeName, Map<String, String> connectorAttributes)
- throws Exception {
+ protected PipeConnector validateConnector(
+ String pipeName, Map<String, String> connectorAttributes) throws Exception {
final PipeParameters connectorParameters = new PipeParameters(connectorAttributes);
final PipeConnector temporaryConnector = reflectConnector(connectorParameters);
try {
@@ -137,6 +141,7 @@ public abstract class PipePluginAgent {
LOGGER.warn("Failed to close temporary connector: {}", e.getMessage(), e);
}
}
+ return temporaryConnector;
}
/**
@@ -150,7 +155,7 @@ public abstract class PipePluginAgent {
*/
public final List<String> getSubProcessorNamesWithSpecifiedParent(
Class<? extends PipeProcessor> parentClass) throws PipeException {
- return Arrays.stream(pipePluginMetaKeeper.getAllPipePluginMeta())
+ return StreamSupport.stream(pipePluginMetaKeeper.getAllPipePluginMeta().spliterator(), false)
.map(pipePluginMeta -> pipePluginMeta.getPluginName().toLowerCase())
.filter(
pluginName -> {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/PlaceholderConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/PlaceholderConnector.java
index 2ad6e893d3c..39c2676120d 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/PlaceholderConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/PlaceholderConnector.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector;
import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -34,6 +36,8 @@ import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
* server module will replace this class with the real implementation when initializing the IoTDB
* pipe connector.
*/
+@TreeModel
+@TableModel
public class PlaceholderConnector implements PipeConnector {
private static final String PLACEHOLDER_ERROR_MSG =
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/donothing/DoNothingConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/donothing/DoNothingConnector.java
index 12132f85cd2..366df72b23e 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/donothing/DoNothingConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/donothing/DoNothingConnector.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.donothing;
import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -27,6 +29,8 @@ 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;
+@TreeModel
+@TableModel
public class DoNothingConnector implements PipeConnector {
@Override
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/airgap/IoTDBAirGapConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/airgap/IoTDBAirGapConnector.java
index 641a0c993a6..67d0461e875 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/airgap/IoTDBAirGapConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/airgap/IoTDBAirGapConnector.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.airgap;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderConnector;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
/**
* This class is a placeholder and should not be initialized. It represents the IoTDB Air Gap
@@ -27,4 +29,6 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderC
* pipe agent in the server module will replace this class with the real implementation when
* initializing the IoTDB Air Gap connector.
*/
+@TreeModel
+@TableModel
public class IoTDBAirGapConnector extends PlaceholderConnector {}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/consensus/PipeConsensusAsyncConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/consensus/PipeConsensusAsyncConnector.java
index fb55bc71a10..583cdf1fecd 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/consensus/PipeConsensusAsyncConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/consensus/PipeConsensusAsyncConnector.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.consensus;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderConnector;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
/**
* This class is a placeholder and should not be initialized. It represents the
@@ -27,4 +29,6 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderC
* imported here. The pipe agent in the server module will replace this class with the real
* implementation when initializing the PipeConsensusAsyncConnector.
*/
+@TreeModel
+@TableModel
public class PipeConsensusAsyncConnector extends PlaceholderConnector {}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBLegacyPipeConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBLegacyPipeConnector.java
index ec5e73f5298..81d7efef375 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBLegacyPipeConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBLegacyPipeConnector.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.thrift;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderConnector;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
/**
* This class is a placeholder and should not be initialized. It represents the IoTDB legacy pipe
@@ -27,4 +28,5 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderC
* imported here. The pipe agent in the server module will replace this class with the real
* implementation when initializing the IoTDB legacy pipe connector.
*/
+@TreeModel
public class IoTDBLegacyPipeConnector extends PlaceholderConnector {}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftAsyncConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftAsyncConnector.java
index 8f3d02d5ada..22008397464 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftAsyncConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftAsyncConnector.java
@@ -19,10 +19,15 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.thrift;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
+
/**
* This class is a placeholder and should not be initialized. It represents the IoTDB Thrift async
* connector. There is a real implementation in the server module but cannot be imported here. The
* pipe agent in the server module will replace this class with the real implementation when
* initializing the IoTDB Thrift async connector.
*/
+@TreeModel
+@TableModel
public class IoTDBThriftAsyncConnector extends IoTDBThriftConnector {}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftConnector.java
index a2786eba616..4fb4178a6c9 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftConnector.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.thrift;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderConnector;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
/**
* This class is a placeholder and should not be initialized. It represents the IoTDB Thrift
@@ -27,4 +29,6 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderC
* pipe agent in the server module will replace this class with the real implementation when
* initializing the IoTDB Thrift connector.
*/
+@TreeModel
+@TableModel
public class IoTDBThriftConnector extends PlaceholderConnector {}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftSslConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftSslConnector.java
index 165d91707d7..a1092f1faab 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftSslConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftSslConnector.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.thrift;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderConnector;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
/**
* This class is a placeholder and should not be initialized. It represents the IoTDB SSL connector.
@@ -27,4 +29,6 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderC
* in the server module will replace this class with the real implementation when initializing the
* IoTDB SSL connector.
*/
+@TreeModel
+@TableModel
public class IoTDBThriftSslConnector extends PlaceholderConnector {}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftSyncConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftSyncConnector.java
index ee80ef63c17..6ef726b871c 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftSyncConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/iotdb/thrift/IoTDBThriftSyncConnector.java
@@ -19,10 +19,15 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.iotdb.thrift;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
+
/**
* This class is a placeholder and should not be initialized. It represents the IoTDB Thrift sync
* connector. There is a real implementation in the server module but cannot be imported here. The
* pipe agent in the server module will replace this class with the real implementation when
* initializing the IoTDB Thrift sync connector.
*/
+@TreeModel
+@TableModel
public class IoTDBThriftSyncConnector extends IoTDBThriftConnector {}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/opcua/OpcUaConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/opcua/OpcUaConnector.java
index a6448d4209c..dc2746491cf 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/opcua/OpcUaConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/opcua/OpcUaConnector.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.opcua;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderConnector;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
/**
* This class is a placeholder and should not be initialized. It represents the OPC UA connector.
@@ -27,4 +29,6 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderC
* in the server module will replace this class with the real implementation when initializing the
* OPC UA connector.
*/
+@TreeModel
+@TableModel
public class OpcUaConnector extends PlaceholderConnector {}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/websocket/WebSocketConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/websocket/WebSocketConnector.java
index a687a9c79a3..0d6e1de3c57 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/websocket/WebSocketConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/websocket/WebSocketConnector.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.websocket;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderConnector;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
/**
* This class is a placeholder and should not be initialized. It represents the Web Socket
@@ -27,4 +28,5 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderC
* pipe agent in the server module will replace this class with the real implementation when
* initializing the Web Socket connector.
*/
+@TreeModel
public class WebSocketConnector extends PlaceholderConnector {}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/writeback/WriteBackConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/writeback/WriteBackConnector.java
index 11ab79200f4..f32471a8856 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/writeback/WriteBackConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/connector/writeback/WriteBackConnector.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.writeback;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderConnector;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
/**
* This class is a placeholder and should not be initialized. It represents the Write Back
@@ -27,4 +29,6 @@ import org.apache.iotdb.commons.pipe.agent.plugin.builtin.connector.PlaceholderC
* pipe agent in the server module will replace this class with the real implementation when
* initializing the Write Back connector.
*/
+@TreeModel
+@TableModel
public class WriteBackConnector extends PlaceholderConnector {}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/extractor/donothing/DoNothingExtractor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/extractor/donothing/DoNothingExtractor.java
index 6c9fac230f7..d8cb4e00f64 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/extractor/donothing/DoNothingExtractor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/extractor/donothing/DoNothingExtractor.java
@@ -20,11 +20,15 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.extractor.donothing;
import org.apache.iotdb.pipe.api.PipeExtractor;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
+@TreeModel
+@TableModel
public class DoNothingExtractor implements PipeExtractor {
@Override
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/extractor/iotdb/IoTDBExtractor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/extractor/iotdb/IoTDBExtractor.java
index 53f048f148d..368f1c2fce4 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/extractor/iotdb/IoTDBExtractor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/extractor/iotdb/IoTDBExtractor.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.extractor.iotdb;
import org.apache.iotdb.pipe.api.PipeExtractor;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -31,6 +33,8 @@ import org.apache.iotdb.pipe.api.event.Event;
* imported here. The pipe agent in the server module will replace this class with the real
* implementation when initializing the extractor.
*/
+@TreeModel
+@TableModel
public class IoTDBExtractor implements PipeExtractor {
private static final String PLACEHOLDER_ERROR_MSG =
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/processor/donothing/DoNothingProcessor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/processor/donothing/DoNothingProcessor.java
index 0a2dc4895bd..1aa8a394119 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/processor/donothing/DoNothingProcessor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/processor/donothing/DoNothingProcessor.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.donothing;
import org.apache.iotdb.pipe.api.PipeProcessor;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -30,6 +32,8 @@ import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
import java.io.IOException;
+@TreeModel
+@TableModel
public class DoNothingProcessor implements PipeProcessor {
@Override
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/processor/throwing/ThrowingExceptionProcessor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/processor/throwing/ThrowingExceptionProcessor.java
index ca4cd6f7dde..2b55340160e 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/processor/throwing/ThrowingExceptionProcessor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/processor/throwing/ThrowingExceptionProcessor.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.throwing;
import org.apache.iotdb.pipe.api.PipeProcessor;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -32,6 +34,8 @@ import java.util.Arrays;
import java.util.Set;
import java.util.stream.Collectors;
+@TreeModel
+@TableModel
public class ThrowingExceptionProcessor implements PipeProcessor {
private boolean throwInCustomize = false;
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/meta/PipePluginMetaKeeper.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/meta/PipePluginMetaKeeper.java
index 52b6edd5744..6f4ee1caa2f 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/meta/PipePluginMetaKeeper.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/meta/PipePluginMetaKeeper.java
@@ -20,12 +20,15 @@
package org.apache.iotdb.commons.pipe.agent.plugin.meta;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
+import org.apache.iotdb.commons.pipe.datastructure.visibility.Visibility;
+import org.apache.iotdb.commons.pipe.datastructure.visibility.VisibilityUtils;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.util.Collections;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
@@ -33,21 +36,23 @@ import java.util.concurrent.ConcurrentHashMap;
public abstract class PipePluginMetaKeeper {
protected final Map<String, PipePluginMeta> pipePluginNameToMetaMap = new ConcurrentHashMap<>();
- protected final Map<String, Class<?>> builtinPipePluginNameToClassMap;
+ protected final Map<String, Class<?>> builtinPipePluginNameToClassMap = new ConcurrentHashMap<>();
+ protected final Map<String, Visibility> pipePluginNameToVisibilityMap = new ConcurrentHashMap<>();
public PipePluginMetaKeeper() {
- builtinPipePluginNameToClassMap = new ConcurrentHashMap<>();
loadBuiltinPlugins();
}
- protected void loadBuiltinPlugins() {
+ private void loadBuiltinPlugins() {
for (final BuiltinPipePlugin builtinPipePlugin : BuiltinPipePlugin.values()) {
- addPipePluginMeta(
- builtinPipePlugin.getPipePluginName(),
- new PipePluginMeta(
- builtinPipePlugin.getPipePluginName(), builtinPipePlugin.getClassName()));
- addBuiltinPluginClass(
- builtinPipePlugin.getPipePluginName(), builtinPipePlugin.getPipePluginClass());
+ final String pipePluginName = builtinPipePlugin.getPipePluginName();
+ final Class<?> pipePluginClass = builtinPipePlugin.getPipePluginClass();
+ final String className = builtinPipePlugin.getClassName();
+
+ addPipePluginMeta(pipePluginName, new PipePluginMeta(pipePluginName, className));
+ addBuiltinPluginClass(pipePluginName, pipePluginClass);
+ addPipePluginVisibility(
+ pipePluginName, VisibilityUtils.calculateFromPluginClass(pipePluginClass));
}
}
@@ -63,8 +68,8 @@ public abstract class PipePluginMetaKeeper {
return pipePluginNameToMetaMap.get(pluginName.toUpperCase());
}
- public PipePluginMeta[] getAllPipePluginMeta() {
- return pipePluginNameToMetaMap.values().toArray(new PipePluginMeta[0]);
+ public Iterable<PipePluginMeta> getAllPipePluginMeta() {
+ return pipePluginNameToMetaMap.values();
}
public boolean containsPipePlugin(String pluginName) {
@@ -88,6 +93,26 @@ public abstract class PipePluginMetaKeeper {
return null;
}
+ public void addPipePluginVisibility(String pluginName, Visibility visibility) {
+ pipePluginNameToVisibilityMap.put(pluginName.toUpperCase(), visibility);
+ }
+
+ public void removePipePluginVisibility(String pluginName) {
+ pipePluginNameToVisibilityMap.remove(pluginName.toUpperCase());
+ }
+
+ public boolean visibleUnder(String pluginName, boolean isTableModel) {
+ final Visibility visibility = pipePluginNameToVisibilityMap.get(pluginName);
+ if (Objects.isNull(visibility)) {
+ return false;
+ }
+ return VisibilityUtils.isCompatible(visibility, isTableModel);
+ }
+
+ public Map<String, Visibility> getPipePluginNameToVisibilityMap() {
+ return Collections.unmodifiableMap(pipePluginNameToVisibilityMap);
+ }
+
protected void processTakeSnapshot(OutputStream outputStream) throws IOException {
ReadWriteIOUtils.write(
(int)
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
index a42b7d5ea7e..e271694791c 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
@@ -468,7 +468,7 @@ public abstract class PipeTaskAgent {
final boolean needToStartPipe = pipeStatusFromCoordinator.get() == PipeStatus.RUNNING;
pipeStatusFromCoordinator.set(PipeStatus.STOPPED);
- pipeMetaKeeper.addPipeMeta(pipeName, pipeMetaFromCoordinator);
+ pipeMetaKeeper.addPipeMeta(pipeMetaFromCoordinator);
// If the pipe status from coordinator is RUNNING, we will start the pipe later.
return needToStartPipe;
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeMeta.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeMeta.java
index 033fe650582..997278010e9 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeMeta.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeMeta.java
@@ -19,10 +19,6 @@
package org.apache.iotdb.commons.pipe.agent.task.meta;
-import org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant;
-import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
-import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
-
import org.apache.tsfile.utils.PublicBAOS;
import java.io.DataOutputStream;
@@ -30,7 +26,6 @@ import java.io.FileInputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.ByteBuffer;
-import java.util.Arrays;
import java.util.Objects;
public class PipeMeta {
@@ -66,41 +61,6 @@ public class PipeMeta {
return temporaryMeta;
}
- public boolean visibleUnder(final boolean isTableModel) {
- final PipeParameters extractorParameters = getStaticMeta().getExtractorParameters();
-
- // visible under all model when 'mode.double-living' is set to true
- final boolean isDoubleLiving =
- extractorParameters.getBooleanOrDefault(
- Arrays.asList(
- PipeExtractorConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
- PipeExtractorConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
- PipeExtractorConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
- if (isDoubleLiving) {
- return true;
- }
-
- final boolean isTreeDialect =
- extractorParameters
- .getStringOrDefault(
- SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TREE_VALUE)
- .equals(SystemConstant.SQL_DIALECT_TREE_VALUE);
- final Boolean _isCaptureTree =
- extractorParameters.getBooleanByKeys(
- PipeExtractorConstant.EXTRACTOR_CAPTURE_TREE_KEY,
- PipeExtractorConstant.SOURCE_CAPTURE_TREE_KEY);
- final boolean isCaptureTree = Objects.nonNull(_isCaptureTree) ? _isCaptureTree : isTreeDialect;
- final Boolean _isCaptureTable =
- extractorParameters.getBooleanByKeys(
- PipeExtractorConstant.EXTRACTOR_CAPTURE_TABLE_KEY,
- PipeExtractorConstant.SOURCE_CAPTURE_TABLE_KEY);
- final boolean isCaptureTable =
- Objects.nonNull(_isCaptureTable) ? _isCaptureTable : !isTreeDialect;
-
- // visible under specific tree or table model <-> actually capture tree or table data
- return isTableModel ? isCaptureTable : isCaptureTree;
- }
-
public ByteBuffer serialize() throws IOException {
final PublicBAOS byteArrayOutputStream = new PublicBAOS();
final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream);
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeMetaKeeper.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeMetaKeeper.java
index 4009288dfc7..828cb69afa9 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeMetaKeeper.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeMetaKeeper.java
@@ -69,8 +69,8 @@ public class PipeMetaKeeper {
///////////////////////////////// PipeMeta /////////////////////////////////
- public void addPipeMeta(String pipeName, PipeMeta pipeMeta) {
- pipeNameToPipeMetaMap.put(pipeName, pipeMeta);
+ public void addPipeMeta(PipeMeta pipeMeta) {
+ pipeNameToPipeMetaMap.put(pipeMeta.getStaticMeta().getPipeName(), pipeMeta);
}
public PipeMeta getPipeMeta(String pipeName) {
@@ -90,7 +90,7 @@ public class PipeMetaKeeper {
if (Objects.isNull(pipeMeta)) {
return false;
}
- return pipeMeta.visibleUnder(isTableModel);
+ return pipeMeta.getStaticMeta().visibleUnder(isTableModel);
}
public Iterable<PipeMeta> getPipeMetaList() {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeStaticMeta.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeStaticMeta.java
index 4111a729255..5178aeb2643 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeStaticMeta.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeStaticMeta.java
@@ -19,10 +19,14 @@
package org.apache.iotdb.commons.pipe.agent.task.meta;
+import org.apache.iotdb.commons.pipe.datastructure.visibility.Visibility;
+import org.apache.iotdb.commons.pipe.datastructure.visibility.VisibilityUtils;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.tsfile.utils.PublicBAOS;
import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
@@ -35,6 +39,8 @@ import java.util.Objects;
public class PipeStaticMeta {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PipeStaticMeta.class);
+
private String pipeName;
private long creationTime;
@@ -223,4 +229,12 @@ public class PipeStaticMeta {
final String topicName, final String consumerGroupId) {
return SUBSCRIPTION_PIPE_PREFIX + topicName + "_" + consumerGroupId;
}
+
+ ///////////////////////////////// Tree & Table Isolation /////////////////////////////////
+
+ public boolean visibleUnder(final boolean isTableModel) {
+ final Visibility visibility =
+ VisibilityUtils.calculateFromExtractorParameters(extractorParameters);
+ return VisibilityUtils.isCompatible(visibility, isTableModel);
+ }
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBAirGapConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBAirGapConnector.java
index a404b202cfd..3df2af38a34 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBAirGapConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBAirGapConnector.java
@@ -24,6 +24,8 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.connector.payload.airgap.AirGapELanguageConstant;
import org.apache.iotdb.commons.pipe.connector.payload.airgap.AirGapOneByteResponse;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.exception.PipeConnectionException;
@@ -57,6 +59,8 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstan
import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_AIR_GAP_HANDSHAKE_TIMEOUT_MS_KEY;
import static org.apache.iotdb.commons.utils.BasicStructureSerDeUtil.LONG_LEN;
+@TreeModel
+@TableModel
public abstract class IoTDBAirGapConnector extends IoTDBConnector {
protected static class AirGapSocket extends Socket {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java
index 8c116d70ce9..3539d25ddfb 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java
@@ -29,6 +29,8 @@ import org.apache.iotdb.commons.pipe.connector.payload.thrift.request.PipeTransf
import org.apache.iotdb.commons.pipe.receiver.PipeReceiverStatusHandler;
import org.apache.iotdb.commons.utils.NodeUrlUtils;
import org.apache.iotdb.pipe.api.PipeConnector;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -115,6 +117,8 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstan
import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_LOAD_TSFILE_STRATEGY_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_RATE_LIMIT_KEY;
+@TreeModel
+@TableModel
public abstract class IoTDBConnector implements PipeConnector {
private static final String PARSE_URL_ERROR_FORMATTER =
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBSslSyncConnector.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBSslSyncConnector.java
index b56230109de..329bbd031b8 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBSslSyncConnector.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBSslSyncConnector.java
@@ -26,6 +26,8 @@ import org.apache.iotdb.commons.pipe.connector.client.IoTDBSyncClient;
import org.apache.iotdb.commons.pipe.connector.client.IoTDBSyncClientManager;
import org.apache.iotdb.commons.pipe.connector.payload.thrift.request.PipeTransferFilePieceReq;
import org.apache.iotdb.commons.pipe.connector.payload.thrift.response.PipeTransferFilePieceResp;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -58,6 +60,8 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstan
import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_LEADER_CACHE_ENABLE_KEY;
+@TreeModel
+@TableModel
public abstract class IoTDBSslSyncConnector extends IoTDBConnector {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBSslSyncConnector.class);
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/resource/PersistentResource.java
similarity index 95%
copy from iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java
copy to iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/resource/PersistentResource.java
index 8d562f7552c..0e3b39a0f71 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/resource/PersistentResource.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.commons.pipe.datastructure;
+package org.apache.iotdb.commons.pipe.datastructure.resource;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/result/Result.java
similarity index 57%
copy from iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java
copy to iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/result/Result.java
index 8d562f7552c..48b76508b22 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/result/Result.java
@@ -17,16 +17,39 @@
* under the License.
*/
-package org.apache.iotdb.commons.pipe.datastructure;
+package org.apache.iotdb.commons.pipe.datastructure.result;
-import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+public class Result<T, E> {
-public interface PersistentResource {
- ProgressIndex getProgressIndex();
+ private final T ok;
+ private final E err;
- /** Used in PipeHistoricalDataRegionTsFileAndDeletionExtractor to sort PersistentResource */
- long getFileStartTime();
+ private Result(final T ok, final E err) {
+ this.ok = ok;
+ this.err = err;
+ }
- /** Used in PipeHistoricalDataRegionTsFileAndDeletionExtractor to sort PersistentResource */
- long getFileEndTime();
+ public static <T, E> Result<T, E> ok(final T value) {
+ return new Result<>(value, null);
+ }
+
+ public static <T, E> Result<T, E> err(final E error) {
+ return new Result<>(null, error);
+ }
+
+ public boolean isOk() {
+ return ok != null;
+ }
+
+ public boolean isErr() {
+ return err != null;
+ }
+
+ public T getOk() {
+ return ok;
+ }
+
+ public E getErr() {
+ return err;
+ }
}
diff --git a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipePlugin.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/Visibility.java
similarity index 85%
copy from iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipePlugin.java
copy to iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/Visibility.java
index ec06f5de31e..94500d06961 100644
--- a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipePlugin.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/Visibility.java
@@ -17,6 +17,11 @@
* under the License.
*/
-package org.apache.iotdb.pipe.api;
+package org.apache.iotdb.commons.pipe.datastructure.visibility;
-public interface PipePlugin extends AutoCloseable {}
+public enum Visibility {
+ NONE,
+ TREE_ONLY,
+ TABLE_ONLY,
+ BOTH,
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityTestUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityTestUtils.java
new file mode 100644
index 00000000000..14e268152e6
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityTestUtils.java
@@ -0,0 +1,135 @@
+/*
+ * 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.datastructure.visibility;
+
+import org.apache.iotdb.commons.pipe.datastructure.result.Result;
+import org.apache.iotdb.pipe.api.PipePlugin;
+
+import org.reflections.Reflections;
+import org.reflections.scanners.SubTypesScanner;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+public class VisibilityTestUtils {
+
+ public static Result<Void, String> testVisibilityCompatibilityEntry(final String prefix) {
+ // Use the Reflections library to scan the classpath
+ final Reflections reflections = new Reflections(prefix, new SubTypesScanner(false));
+ final Set<Class<? extends PipePlugin>> subTypes = reflections.getSubTypesOf(PipePlugin.class);
+
+ // Create root node
+ final TreeNode root = new TreeNode(PipePlugin.class);
+
+ // Build the tree
+ final Map<Class<?>, TreeNode> nodeMap = new HashMap<>();
+ nodeMap.put(PipePlugin.class, root);
+
+ for (final Class<?> subType : subTypes) {
+ final TreeNode node = new TreeNode(subType);
+ nodeMap.put(subType, node);
+
+ if (subType.isInterface()) {
+ // Handle super interfaces of the interface
+ final Class<?>[] superInterfaces = subType.getInterfaces();
+ for (Class<?> superInterface : superInterfaces) {
+ while (!nodeMap.containsKey(superInterface) && superInterface != null) {
+ superInterface = getSuperInterface(superInterface);
+ }
+
+ if (superInterface != null) {
+ nodeMap.get(superInterface).children.put(subType, node);
+ }
+ }
+ } else {
+ // Handle superclass of the class
+ Class<?> superClass = subType.getSuperclass();
+ if (superClass == Object.class) {
+ // If the superclass is Object, check the implemented interfaces
+ final Class<?>[] interfaces = subType.getInterfaces();
+ for (final Class<?> iface : interfaces) {
+ if (nodeMap.containsKey(iface)) {
+ nodeMap.get(iface).children.put(subType, node);
+ }
+ }
+ } else {
+ while (!nodeMap.containsKey(superClass) && superClass != null) {
+ superClass = superClass.getSuperclass();
+ }
+
+ if (superClass != null) {
+ nodeMap.get(superClass).children.put(subType, node);
+ }
+ }
+ }
+ }
+
+ root.visibility = Visibility.BOTH;
+ for (final TreeNode node : root.children.values()) {
+ node.visibility = Visibility.BOTH;
+ }
+
+ // Validate the visibility compatibility of the tree
+ return validateTreeNode(root);
+ }
+
+ private static Result<Void, String> validateTreeNode(final TreeNode node) {
+ for (final TreeNode child : node.children.values()) {
+ if (!VisibilityUtils.isCompatible(node.visibility, child.visibility)) {
+ return Result.err(
+ String.format(
+ "Incompatible visibility detected:\n"
+ + "Parent class: "
+ + node.clazz.getName()
+ + ", Visibility: "
+ + node.visibility
+ + "\n"
+ + "Child class: "
+ + child.clazz.getName()
+ + ", Visibility: "
+ + child.visibility));
+ }
+ final Result<Void, String> childResult = validateTreeNode(child);
+ if (childResult.isErr()) {
+ return childResult;
+ }
+ }
+ return Result.ok(null);
+ }
+
+ // Get the super interface of an interface
+ private static Class<?> getSuperInterface(final Class<?> interfaceClass) {
+ final Class<?>[] superInterfaces = interfaceClass.getInterfaces();
+ // TODO: We assume the first interface
+ return superInterfaces.length > 0 ? superInterfaces[0] : null;
+ }
+
+ private static class TreeNode {
+ Class<?> clazz;
+ Visibility visibility;
+ Map<Class<?>, TreeNode> children = new HashMap<>();
+
+ TreeNode(final Class<?> clazz) {
+ this.clazz = clazz;
+ this.visibility = VisibilityUtils.calculateFromPluginClass(clazz);
+ }
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtils.java
new file mode 100644
index 00000000000..5d44fa1379c
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtils.java
@@ -0,0 +1,133 @@
+/*
+ * 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.datastructure.visibility;
+
+import org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant;
+import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.Objects;
+
+public class VisibilityUtils {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(VisibilityUtils.class);
+
+ private VisibilityUtils() {
+ // forbidding instantiation
+ }
+
+ public static boolean isCompatible(final Visibility visibility, final boolean isTableModel) {
+ if (Objects.equals(Visibility.BOTH, visibility)) {
+ return true;
+ }
+ return isTableModel
+ ? Objects.equals(Visibility.TABLE_ONLY, visibility)
+ : Objects.equals(Visibility.TREE_ONLY, visibility);
+ }
+
+ public static boolean isCompatible(final Visibility base, final Visibility incoming) {
+ if (Objects.equals(base, Visibility.BOTH)) {
+ return true;
+ }
+ if (Objects.equals(base, Visibility.TREE_ONLY)) {
+ return Objects.equals(incoming, Visibility.TREE_ONLY);
+ }
+ if (Objects.equals(base, Visibility.TABLE_ONLY)) {
+ return Objects.equals(incoming, Visibility.TABLE_ONLY);
+ }
+ return false;
+ }
+
+ public static boolean isCompatible(
+ final Visibility pipeVisibility, final Visibility... pluginVisibilities) {
+ return Arrays.stream(pluginVisibilities)
+ .allMatch(pluginVisibility -> isCompatible(pluginVisibility, pipeVisibility));
+ }
+
+ public static Visibility calculateFromPluginClass(final Class<?> pipePluginClass) {
+ final boolean isTreeModelAnnotationPresent =
+ pipePluginClass.isAnnotationPresent(TreeModel.class);
+ final boolean isTableModelAnnotationPresent =
+ pipePluginClass.isAnnotationPresent(TableModel.class);
+ if (!isTreeModelAnnotationPresent && !isTableModelAnnotationPresent) {
+ return Visibility.TREE_ONLY; // default to tree only if missing annotations
+ } else if (isTreeModelAnnotationPresent && !isTableModelAnnotationPresent) {
+ return Visibility.TREE_ONLY;
+ } else if (!isTreeModelAnnotationPresent && isTableModelAnnotationPresent) {
+ return Visibility.TABLE_ONLY;
+ } else {
+ return Visibility.BOTH;
+ }
+ }
+
+ public static Visibility calculateFromExtractorParameters(
+ final PipeParameters extractorParameters) {
+ // visible under all model when 'mode.double-living' is set to true
+ final boolean isDoubleLiving =
+ extractorParameters.getBooleanOrDefault(
+ Arrays.asList(
+ PipeExtractorConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
+ PipeExtractorConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
+ PipeExtractorConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+ if (isDoubleLiving) {
+ return Visibility.BOTH;
+ }
+
+ final boolean isTreeDialect =
+ extractorParameters
+ .getStringOrDefault(
+ SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TREE_VALUE)
+ .equals(SystemConstant.SQL_DIALECT_TREE_VALUE);
+ final Boolean _isCaptureTree =
+ extractorParameters.getBooleanByKeys(
+ PipeExtractorConstant.EXTRACTOR_CAPTURE_TREE_KEY,
+ PipeExtractorConstant.SOURCE_CAPTURE_TREE_KEY);
+ final boolean isCaptureTree = Objects.nonNull(_isCaptureTree) ? _isCaptureTree : isTreeDialect;
+ final Boolean _isCaptureTable =
+ extractorParameters.getBooleanByKeys(
+ PipeExtractorConstant.EXTRACTOR_CAPTURE_TABLE_KEY,
+ PipeExtractorConstant.SOURCE_CAPTURE_TABLE_KEY);
+ final boolean isCaptureTable =
+ Objects.nonNull(_isCaptureTable) ? _isCaptureTable : !isTreeDialect;
+
+ // visible under specific tree or table model <-> actually capture tree or table data
+ if (isCaptureTree && isCaptureTable) {
+ return Visibility.BOTH;
+ }
+ if (isCaptureTree) {
+ return Visibility.TREE_ONLY;
+ }
+ if (isCaptureTable) {
+ return Visibility.TABLE_ONLY;
+ }
+
+ // UNREACHABLE CODE
+ LOGGER.error(
+ "BROKEN INVARIANT: DETECT INVISIBLE EXTRACTOR PARAMETERS {}",
+ extractorParameters.getAttribute());
+ return Visibility.NONE;
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBExtractor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBExtractor.java
index 75825bf18c9..62eb86ca728 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBExtractor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBExtractor.java
@@ -23,6 +23,8 @@ import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant;
import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskExtractorRuntimeEnvironment;
import org.apache.iotdb.pipe.api.PipeExtractor;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -37,6 +39,8 @@ import static org.apache.iotdb.commons.pipe.datastructure.options.PipeInclusionO
import static org.apache.iotdb.commons.pipe.datastructure.options.PipeInclusionOptions.hasAtLeastOneOption;
import static org.apache.iotdb.commons.pipe.datastructure.options.PipeInclusionOptions.optionsAreAllLegal;
+@TreeModel
+@TableModel
public abstract class IoTDBExtractor implements PipeExtractor {
// Record these variables to provide corresponding value to tag key of monitoring metrics
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBNonDataRegionExtractor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBNonDataRegionExtractor.java
index eeac1819f25..f10a42bdc67 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBNonDataRegionExtractor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBNonDataRegionExtractor.java
@@ -31,6 +31,8 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.event.PipeSnapshotEvent;
import org.apache.iotdb.commons.pipe.event.PipeWritePlanEvent;
import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
+import org.apache.iotdb.pipe.api.annotation.TableModel;
+import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
@@ -44,6 +46,8 @@ import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
+@TreeModel
+@TableModel
public abstract class IoTDBNonDataRegionExtractor extends IoTDBExtractor {
protected IoTDBTreePattern treePattern;
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/plugin/annotation/PipePluginAnnotationTest.java
similarity index 59%
rename from iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java
rename to iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/plugin/annotation/PipePluginAnnotationTest.java
index 8d562f7552c..da693b1d4ab 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/PersistentResource.java
+++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/plugin/annotation/PipePluginAnnotationTest.java
@@ -17,16 +17,23 @@
* under the License.
*/
-package org.apache.iotdb.commons.pipe.datastructure;
+package org.apache.iotdb.commons.pipe.plugin.annotation;
-import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import org.apache.iotdb.commons.pipe.datastructure.result.Result;
+import org.apache.iotdb.commons.pipe.datastructure.visibility.VisibilityTestUtils;
-public interface PersistentResource {
- ProgressIndex getProgressIndex();
+import org.junit.Test;
- /** Used in PipeHistoricalDataRegionTsFileAndDeletionExtractor to sort PersistentResource */
- long getFileStartTime();
+import static org.junit.Assert.fail;
- /** Used in PipeHistoricalDataRegionTsFileAndDeletionExtractor to sort PersistentResource */
- long getFileEndTime();
+public class PipePluginAnnotationTest {
+
+ @Test
+ public void testPipePluginVisibility() {
+ final Result<Void, String> result =
+ VisibilityTestUtils.testVisibilityCompatibilityEntry("org.apache.iotdb.commons");
+ if (result.isErr()) {
+ fail(result.getErr());
+ }
+ }
}
diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
index ebe67399552..ee62ed30686 100644
--- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
@@ -706,6 +706,7 @@ struct TCreatePipePluginReq {
struct TDropPipePluginReq {
1: required string pluginName
2: optional bool ifExistsCondition
+ 3: optional bool isTableModel
}
// Get PipePlugin table from config node
@@ -714,6 +715,10 @@ struct TGetPipePluginTableResp {
2: required list<binary> allPipePluginMeta
}
+struct TShowPipePluginReq {
+ 1: optional bool isTableModel
+}
+
// Pipe
struct TShowPipeInfo {
@@ -1521,6 +1526,11 @@ service IConfigNodeRPCService {
*/
TGetPipePluginTableResp getPipePluginTable();
+ /**
+ * Return the pipe plugin table
+ */
+ TGetPipePluginTableResp getPipePluginTableExtended(TShowPipePluginReq req);
+
/**
* Return the pipe plugin jar list of the plugin name list
*/