IoTConsensusV2: Transfer table deletion without any parse or filter (apache#14988)

* transfer table deletion for iotv2

* minor fix

Author: Pengzna

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java

@@ -57,6 +57,8 @@ public class PipeEventCollector implements EventCollector {
 
   private final boolean skipParseTsFile;
 
+  private final boolean isUsedForConsensusPipe;
+
   private final AtomicInteger collectInvocationCount = new AtomicInteger(0);
   private boolean hasNoGeneratedEvent = true;
   private boolean isFailedToIncreaseReferenceCount = false;
@@ -66,12 +68,14 @@ public PipeEventCollector(
       final long creationTime,
       final int regionId,
       final boolean forceTabletFormat,
-      final boolean skipParseTsFile) {
+      final boolean skipParseTsFile,
+      final boolean isUsedInConsensusPipe) {
     this.pendingQueue = pendingQueue;
     this.creationTime = creationTime;
     this.regionId = regionId;
     this.forceTabletFormat = forceTabletFormat;
     this.skipParseTsFile = skipParseTsFile;
+    this.isUsedForConsensusPipe = isUsedInConsensusPipe;
   }
 
   @Override
@@ -153,6 +157,13 @@ private void collectParsedRawTableEvent(final PipeRawTabletInsertionEvent parsed
   }
 
   private void parseAndCollectEvent(final PipeDeleteDataNodeEvent deleteDataEvent) {
+    // For IoTConsensusV2, there is no need to parse. So we can directly transfer deleteDataEvent
+    if (isUsedForConsensusPipe) {
+      hasNoGeneratedEvent = false;
+      collectEvent(deleteDataEvent);
+      return;
+    }
+
     // Only used by events containing delete data node, no need to bind progress index here since
     // delete data event does not have progress index currently
     (deleteDataEvent.getDeleteDataNode() instanceof DeleteDataNode

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskProcessorStage.java

@@ -23,6 +23,7 @@
 import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
 import org.apache.iotdb.commons.pipe.agent.task.connection.EventSupplier;
 import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
+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.agent.task.stage.PipeTaskStage;
 import org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant;
@@ -99,13 +100,15 @@ public PipeTaskProcessorStage(
     // removed, the new subtask will have the same pipeName and regionId as the
     // old one, so we need creationTime to make their hash code different in the map.
     final String taskId = pipeName + "_" + regionId + "_" + creationTime;
+    final boolean isUsedForConsensusPipe = pipeName.contains(PipeStaticMeta.CONSENSUS_PIPE_PREFIX);
     final PipeEventCollector pipeConnectorOutputEventCollector =
         new PipeEventCollector(
             pipeConnectorOutputPendingQueue,
             creationTime,
             regionId,
             forceTabletFormat,
-            skipParseTsFile);
+            skipParseTsFile,
+            isUsedForConsensusPipe);
     this.pipeProcessorSubtask =
         new PipeProcessorSubtask(
             taskId,