(iotdb) branch master updated: Subscription: let subscription module fully manage the parsing process of the insert node event (#15012)

[rong]

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 ce47273ba7f Subscription: let subscription module fully manage the 
parsing process of the insert node event (#15012)
ce47273ba7f is described below

commit ce47273ba7fcfb4ad44afe3c3a175f5cfe39daf0
Author: VGalaxies <vgalax...@apache.org>
AuthorDate: Fri Mar 7 17:01:57 2025 +0800

    Subscription: let subscription module fully manage the parsing process of 
the insert node event (#15012)
---
 .../db/pipe/agent/task/connection/PipeEventCollector.java  | 14 +++++++++-----
 .../db/pipe/agent/task/stage/PipeTaskProcessorStage.java   |  4 ++--
 2 files changed, 11 insertions(+), 7 deletions(-)

diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
index af9ed00b4ec..61ecdc2f228 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
@@ -55,7 +55,7 @@ public class PipeEventCollector implements EventCollector {
 
   private final boolean forceTabletFormat;
 
-  private final boolean skipParseTsFile;
+  private final boolean skipParsing;
 
   private final boolean isUsedForConsensusPipe;
 
@@ -68,13 +68,13 @@ public class PipeEventCollector implements EventCollector {
       final long creationTime,
       final int regionId,
       final boolean forceTabletFormat,
-      final boolean skipParseTsFile,
+      final boolean skipParsing,
       final boolean isUsedInConsensusPipe) {
     this.pendingQueue = pendingQueue;
     this.creationTime = creationTime;
     this.regionId = regionId;
     this.forceTabletFormat = forceTabletFormat;
-    this.skipParseTsFile = skipParseTsFile;
+    this.skipParsing = skipParsing;
     this.isUsedForConsensusPipe = isUsedInConsensusPipe;
   }
 
@@ -100,7 +100,11 @@ public class PipeEventCollector implements EventCollector {
   }
 
   private void parseAndCollectEvent(final PipeInsertNodeTabletInsertionEvent 
sourceEvent) {
-    // TODO: let subscription module fully manage the parsing process of the 
insert node event
+    if (skipParsing) {
+      collectEvent(sourceEvent);
+      return;
+    }
+
     if (sourceEvent.shouldParseTimeOrPattern()) {
       for (final PipeRawTabletInsertionEvent parsedEvent :
           sourceEvent.toRawTabletInsertionEvents()) {
@@ -126,7 +130,7 @@ public class PipeEventCollector implements EventCollector {
       return;
     }
 
-    if (skipParseTsFile) {
+    if (skipParsing) {
       collectEvent(sourceEvent);
       return;
     }
diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskProcessorStage.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskProcessorStage.java
index 4242dbcb8b3..3be40341f03 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskProcessorStage.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskProcessorStage.java
@@ -70,7 +70,7 @@ public class PipeTaskProcessorStage extends PipeTaskStage {
       final PipeProcessorSubtaskExecutor executor,
       final PipeTaskMeta pipeTaskMeta,
       final boolean forceTabletFormat,
-      final boolean skipParseTsFile) {
+      final boolean skipParsing) {
     final PipeProcessorRuntimeConfiguration runtimeConfiguration =
         new PipeTaskRuntimeConfiguration(
             new PipeTaskProcessorRuntimeEnvironment(
@@ -107,7 +107,7 @@ public class PipeTaskProcessorStage extends PipeTaskStage {
             creationTime,
             regionId,
             forceTabletFormat,
-            skipParseTsFile,
+            skipParsing,
             isUsedForConsensusPipe);
     this.pipeProcessorSubtask =
         new PipeProcessorSubtask(