Pipe: Fixed the bug that the schema may be sent twice when split-enabled & database may be null for non-first schema pipes (apache#16586)

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/IoTDBSchemaRegionSource.java

@@ -115,9 +115,14 @@ public void start() throws Exception {
         == 1) {
       SchemaRegionConsensusImpl.getInstance()
           .write(schemaRegionId, new PipeOperateSchemaQueueNode(new PlanNodeId(""), true));
-      database = SchemaEngine.getInstance().getSchemaRegion(schemaRegionId).getDatabaseFullPath();
+    } else if (!PipeDataNodeAgent.runtime().schemaListener(schemaRegionId).isOpened()) {
+      // This may be being concurrently opened, we should not continue to start or else the snapshot
+      // may not be listened
+      PipeDataNodeAgent.runtime().decreaseAndGetSchemaListenerReferenceCount(schemaRegionId);
+      return;
     }
 
+    database = SchemaEngine.getInstance().getSchemaRegion(schemaRegionId).getDatabaseFullPath();
     super.start();
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java

@@ -2105,10 +2105,10 @@ public SettableFuture<ConfigTaskResult> createPipe(
 
     // Syntactic sugar: if full-sync mode is detected (i.e. not snapshot mode, or both realtime
     // and history are true), the pipe is split into history-only and realtime–only modes.
-    final PipeParameters extractorPipeParameters =
+    final PipeParameters sourcePipeParameters =
         new PipeParameters(createPipeStatement.getExtractorAttributes());
     if (PipeConfig.getInstance().getPipeAutoSplitFullEnabled()
-        && PipeDataNodeAgent.task().isFullSync(extractorPipeParameters)) {
+        && PipeDataNodeAgent.task().isFullSync(sourcePipeParameters)) {
       try (final ConfigNodeClient configNodeClient =
           CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
         // 1. Send request to create the real-time data synchronization pipeline
@@ -2120,7 +2120,7 @@ public SettableFuture<ConfigTaskResult> createPipe(
                 .setIfNotExistsCondition(true)
                 // Use extractor parameters for real-time data
                 .setExtractorAttributes(
-                    extractorPipeParameters
+                    sourcePipeParameters
                         .addOrReplaceEquivalentAttributesWithClone(
                             new PipeParameters(
                                 ImmutableMap.of(
@@ -2145,16 +2145,23 @@ public SettableFuture<ConfigTaskResult> createPipe(
                 // Append suffix to the pipeline name for historical data
                 .setPipeName(createPipeStatement.getPipeName() + "_history")
                 .setIfNotExistsCondition(createPipeStatement.hasIfNotExistsCondition())
-                // Use extractor parameters for historical data
+                // Use source parameters for historical data
                 .setExtractorAttributes(
-                    extractorPipeParameters
+                    sourcePipeParameters
                         .addOrReplaceEquivalentAttributesWithClone(
                             new PipeParameters(
                                 ImmutableMap.of(
                                     PipeSourceConstant.EXTRACTOR_REALTIME_ENABLE_KEY,
                                     Boolean.toString(false),
                                     PipeSourceConstant.EXTRACTOR_HISTORY_ENABLE_KEY,
-                                    Boolean.toString(true))))
+                                    Boolean.toString(true),
+                                    // We force the historical pipe to transfer data only
+                                    // Thus we can transfer schema only once
+                                    // And may drop the historical pipe on successfully transferred
+                                    PipeSourceConstant.SOURCE_INCLUSION_KEY,
+                                    PipeSourceConstant.EXTRACTOR_INCLUSION_DEFAULT_VALUE,
+                                    PipeSourceConstant.SOURCE_EXCLUSION_KEY,
+                                    PipeSourceConstant.EXTRACTOR_EXCLUSION_DEFAULT_VALUE)))
                         .getAttribute())
                 .setProcessorAttributes(createPipeStatement.getProcessorAttributes())
                 .setConnectorAttributes(createPipeStatement.getConnectorAttributes());

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/queue/listening/AbstractPipeListeningQueue.java

@@ -76,10 +76,12 @@ protected synchronized void tryListen(final List<PipeSnapshotEvent> events) {
     }
   }
 
-  public synchronized Pair<Long, List<PipeSnapshotEvent>> findAvailableSnapshots() {
-    if (queueTailIndex2SnapshotsCache.getLeft()
-        < queue.getTailIndex()
-            - PipeConfig.getInstance().getPipeListeningQueueTransferSnapshotThreshold()) {
+  public synchronized Pair<Long, List<PipeSnapshotEvent>> findAvailableSnapshots(
+      final boolean mayClear) {
+    if (mayClear
+        && queueTailIndex2SnapshotsCache.getLeft()
+            < queue.getTailIndex()
+                - PipeConfig.getInstance().getPipeListeningQueueTransferSnapshotThreshold()) {
       clearSnapshots();
     }
     return queueTailIndex2SnapshotsCache;

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSource.java

@@ -112,10 +112,10 @@ public void start() throws Exception {
   private long getNextIndexAfterSnapshot() {
     long nextIndex;
     if (needTransferSnapshot()) {
-      nextIndex = findSnapshot();
+      nextIndex = findSnapshot(true);
       if (nextIndex == Long.MIN_VALUE) {
         triggerSnapshot();
-        nextIndex = findSnapshot();
+        nextIndex = findSnapshot(false);
         if (nextIndex == Long.MIN_VALUE) {
           throw new PipeException("Cannot get the newest snapshot after triggering one.");
         }
@@ -128,9 +128,9 @@ private long getNextIndexAfterSnapshot() {
     return nextIndex;
   }
 
-  private long findSnapshot() {
+  private long findSnapshot(final boolean mayClear) {
     final Pair<Long, List<PipeSnapshotEvent>> queueTailIndex2Snapshots =
-        getListeningQueue().findAvailableSnapshots();
+        getListeningQueue().findAvailableSnapshots(mayClear);
     final long nextIndex =
         Objects.nonNull(queueTailIndex2Snapshots.getLeft())
                 && queueTailIndex2Snapshots.getLeft() != Long.MIN_VALUE