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 <rong@apache.org>

Author: VGalaxies

example/pipe-count-point-processor/src/main/java/org/apache/iotdb/CountPointProcessor.java

@@ -23,6 +23,7 @@
 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 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);

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 {}

iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/annotation/TableModel.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.pipe.api.annotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * 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
+ */
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface TableModel {}

iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/annotation/TreeModel.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.pipe.api.annotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * 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
+ */
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface TreeModel {}

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/plugin/PipePluginTableResp.java

@@ -21,22 +21,31 @@
 
 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;
 
 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 TGetPipePluginTableResp convertToThriftResponse() throws IOException {
     }
     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);
+  }
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java

@@ -97,7 +97,7 @@ public PipeTableResp filter(final Boolean whereClause, final String pipeName) {
   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;
   }
 

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java

@@ -215,6 +215,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.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 TGetPipePluginTableResp getPipePluginTable() {
         : 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();

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java

@@ -138,6 +138,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.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 @@ TDataPartitionTableResp getOrCreateDataPartition(
   /** Show pipe plugins. */
   TGetPipePluginTableResp getPipePluginTable();
 
+  /** Show pipe plugins. */
+  TGetPipePluginTableResp getPipePluginTableExtended(TShowPipePluginReq req);
+
   /** Get pipe plugin jar. */
   TGetJarInListResp getPipePluginJar(TGetJarInListReq req);
 

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.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.util.HashMap;
 import java.util.Objects;
 
+@TreeModel
+@TableModel
 public class IoTDBConfigRegionAirGapConnector extends IoTDBAirGapConnector {
 
   private static final Logger LOGGER =

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.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.List;
 import java.util.Objects;
 
+@TreeModel
+@TableModel
 public class IoTDBConfigRegionConnector extends IoTDBSslSyncConnector {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBConfigRegionConnector.class);