apache · Caideyipi · Jul 6, 2026
diff --git a/...ava/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeDoubleLivingIT.java b/...ava/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeDoubleLivingIT.java
@@ -71,7 +71,7 @@ public void testDoubleLivingInvalidForwardingParameter() throws Exception {
               "create pipe %s"
                   + " with source ("
                   + "'capture.tree'='false',"
-                  + "'mode.double-living'='true')"
+                  + "'double-living'='true')"
                   + " with sink ("
                   + "'node-urls'='%s')",
               "p1", receiverDataNode.getIpAndPortString()));
@@ -84,7 +84,7 @@ public void testDoubleLivingInvalidForwardingParameter() throws Exception {
               "create pipe %s"
                   + " with source ("
                   + "'capture.table'='false',"
-                  + "'mode.double-living'='true')"
+                  + "'source.mode.double-living'='true')"
                   + " with sink ("
                   + "'node-urls'='%s')",
               "p2", receiverDataNode.getIpAndPortString()));
@@ -286,7 +286,7 @@ public void testDoubleLivingIsolation() throws Exception {
           String.format(
               "create pipe %s"
                   + " with source ("
-                  + "'mode.double-living'='true')"
+                  + "'double-living'='true')"
                   + " with sink ("
                   + "'node-urls'='%s')",
               treePipeName, receiverDataNode.getIpAndPortString()));
@@ -308,7 +308,7 @@ public void testDoubleLivingIsolation() throws Exception {
           String.format(
               "create pipe %s"
                   + " with source ("
-                  + "'mode.double-living'='true')"
+                  + "'source.mode.double-living'='true')"
                   + " with sink ("
                   + "'node-urls'='%s')",
               tablePipeName, receiverDataNode.getIpAndPortString()));
@@ -350,7 +350,7 @@ public void testDoubleLivingCreatesSameNamePipesWithSeparateDialects() throws Ex
       statement.execute(
           String.format(
               "create pipe %s"
-                  + " with source ('mode.double-living'='true')"
+                  + " with source ('extractor.double-living'='true')"
                   + " with sink ('node-urls'='%s')",
               pipeName, receiverDataNode.getIpAndPortString()));
     }

diff --git a/...n/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java b/...n/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java
@@ -20,6 +20,7 @@
 package org.apache.iotdb.confignode.manager.pipe.coordinator.task;
 
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.i18n.PipeMessages;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
 import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
@@ -48,7 +49,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -121,16 +121,15 @@ public TSStatus createPipe(TCreatePipeReq req) {
   }
 
   private boolean isDoubleLivingPipe(final TCreatePipeReq req) {
-    return new PipeParameters(req.getExtractorAttributes())
-        .getBooleanOrDefault(
-            Arrays.asList(
-                PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
-                PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
-            PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+    return PipeSourceConstant.isDoubleLiving(new PipeParameters(req.getExtractorAttributes()));
   }
 
   private TSStatus createDoubleLivingPipe(final TCreatePipeReq req) {
     final PipeParameters sourceParameters = new PipeParameters(req.getExtractorAttributes());
+    final TSStatus validationStatus = validateDoubleLivingPipeParameters(sourceParameters);
+    if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != validationStatus.getCode()) {
+      return validationStatus;
+    }
     final String currentDialect =
         sourceParameters.getStringOrDefault(
             SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TREE_VALUE);
@@ -180,11 +179,24 @@ private TSStatus createDoubleLivingPipe(final TCreatePipeReq req) {
     return secondStatus;
   }
 
+  private TSStatus validateDoubleLivingPipeParameters(final PipeParameters sourceParameters) {
+    final Boolean isForwardingPipeRequests =
+        sourceParameters.getBooleanByKeys(
+            PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
+            PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY);
+    return Boolean.TRUE.equals(isForwardingPipeRequests)
+        ? RpcUtils.getStatus(
+            TSStatusCode.PIPE_ERROR,
+            PipeMessages
+                .EXCEPTION_FORWARDING_PIPE_REQUESTS_CAN_NOT_SPECIFIED_TRUE_DOUBLE_LIVING_ENABLED_B000E8A1)
+        : RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+  }
+
   private TCreatePipeReq cloneCreatePipeRequestWithDialect(
       final TCreatePipeReq req, final PipeParameters sourceParameters, final String sqlDialect) {
     final Map<String, String> sourceAttributes = new HashMap<>(sourceParameters.getAttribute());
-    sourceAttributes.remove(PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY);
-    sourceAttributes.remove(PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY);
+    PipeSourceConstant.removeDoubleLivingAttributes(sourceAttributes);
+    PipeSourceConstant.disableForwardingPipeRequests(sourceAttributes);
     sourceAttributes.put(SystemConstant.SQL_DIALECT_KEY, sqlDialect);
     sourceAttributes.put(
         SystemConstant.PIPE_VISIBILITY_KEY, SystemConstant.PIPE_VISIBILITY_STRICT_VALUE);

diff --git a/...db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java b/...db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java
@@ -395,20 +395,13 @@ public void customize(
 
     skipIfNoPrivileges = getSkipIfNoPrivileges(parameters);
 
-    final boolean isDoubleLiving =
-        parameters.getBooleanOrDefault(
-            Arrays.asList(
-                PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
-                PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
-            PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+    final boolean isDoubleLiving = PipeSourceConstant.isDoubleLiving(parameters);
     if (isDoubleLiving) {
       isForwardingPipeRequests = false;
     } else {
       isForwardingPipeRequests =
           parameters.getBooleanOrDefault(
-              Arrays.asList(
-                  PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
-                  PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY),
+              PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS,
               PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE);
     }
 

diff --git a/...ava/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java b/...ava/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
@@ -262,20 +262,13 @@ public void customize(
             ? TimePartitionUtils.getTimePartitionId(realtimeDataExtractionEndTime)
             : TimePartitionUtils.getTimePartitionId(realtimeDataExtractionEndTime) - 1;
 
-    final boolean isDoubleLiving =
-        parameters.getBooleanOrDefault(
-            Arrays.asList(
-                PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
-                PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
-            PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+    final boolean isDoubleLiving = PipeSourceConstant.isDoubleLiving(parameters);
     if (isDoubleLiving) {
       isForwardingPipeRequests = false;
     } else {
       isForwardingPipeRequests =
           parameters.getBooleanOrDefault(
-              Arrays.asList(
-                  PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
-                  PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY),
+              PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS,
               PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE);
     }
 

diff --git a/...apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/...apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -60,6 +60,7 @@
 import org.apache.iotdb.commons.exception.SemanticException;
 import org.apache.iotdb.commons.executable.ExecutableManager;
 import org.apache.iotdb.commons.executable.ExecutableResource;
+import org.apache.iotdb.commons.i18n.PipeMessages;
 import org.apache.iotdb.commons.path.MeasurementPath;
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.path.PathPatternTree;
@@ -347,6 +348,7 @@
 import org.apache.iotdb.db.trigger.service.TriggerClassLoader;
 import org.apache.iotdb.pipe.api.PipePlugin;
 import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException;
 import org.apache.iotdb.pipe.api.exception.PipePasswordCheckException;
 import org.apache.iotdb.rpc.RpcUtils;
 import org.apache.iotdb.rpc.StatementExecutionException;
@@ -2278,6 +2280,7 @@ public SettableFuture<ConfigTaskResult> createPipe(
     // Validate pipe plugin before creation
     try {
       if (isDoubleLivingPipe(sourcePipeParameters)) {
+        validateDoubleLivingPipeParameters(sourcePipeParameters);
         validatePipePlugin(
             pipeName,
             cloneSourceParametersWithDialect(
@@ -2342,18 +2345,26 @@ private void validatePipePlugin(
   }
 
   private boolean isDoubleLivingPipe(final PipeParameters sourcePipeParameters) {
-    return sourcePipeParameters.getBooleanOrDefault(
-        Arrays.asList(
-            PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
-            PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
-        PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+    return PipeSourceConstant.isDoubleLiving(sourcePipeParameters);
+  }
+
+  private void validateDoubleLivingPipeParameters(final PipeParameters sourcePipeParameters) {
+    final Boolean isForwardingPipeRequests =
+        sourcePipeParameters.getBooleanByKeys(
+            PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
+            PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY);
+    if (Boolean.TRUE.equals(isForwardingPipeRequests)) {
+      throw new PipeParameterNotValidException(
+          PipeMessages
+              .EXCEPTION_FORWARDING_PIPE_REQUESTS_CAN_NOT_SPECIFIED_TRUE_DOUBLE_LIVING_ENABLED_B000E8A1);
+    }
   }
 
   private PipeParameters cloneSourceParametersWithDialect(
       final PipeParameters sourcePipeParameters, final String sqlDialect) {
     final Map<String, String> sourceAttributes = new HashMap<>(sourcePipeParameters.getAttribute());
-    sourceAttributes.remove(PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY);
-    sourceAttributes.remove(PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY);
+    PipeSourceConstant.removeDoubleLivingAttributes(sourceAttributes);
+    PipeSourceConstant.disableForwardingPipeRequests(sourceAttributes);
     sourceAttributes.put(SystemConstant.SQL_DIALECT_KEY, sqlDialect);
     sourceAttributes.put(
         SystemConstant.PIPE_VISIBILITY_KEY, SystemConstant.PIPE_VISIBILITY_STRICT_VALUE);

diff --git a/...mmons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java b/...mmons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java
@@ -20,10 +20,13 @@
 package org.apache.iotdb.commons.pipe.config.constant;
 
 import org.apache.iotdb.commons.i18n.PipeMessages;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
 
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 public class PipeSourceConstant {
@@ -87,6 +90,10 @@ public class PipeSourceConstant {
   public static final String SOURCE_FORWARDING_PIPE_REQUESTS_KEY =
       "source.forwarding-pipe-requests";
   public static final boolean EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE = true;
+  public static final List<String> FORWARDING_PIPE_REQUESTS_KEYS =
+      Collections.unmodifiableList(
+          Arrays.asList(
+              EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY, SOURCE_FORWARDING_PIPE_REQUESTS_KEY));
 
   public static final String EXTRACTOR_HISTORY_ENABLE_KEY = "extractor.history.enable";
   public static final String SOURCE_HISTORY_ENABLE_KEY = "source.history.enable";
@@ -137,7 +144,16 @@ public class PipeSourceConstant {
   public static final boolean EXTRACTOR_MODE_SNAPSHOT_DEFAULT_VALUE = false;
   public static final String EXTRACTOR_MODE_DOUBLE_LIVING_KEY = "extractor.mode.double-living";
   public static final String SOURCE_MODE_DOUBLE_LIVING_KEY = "source.mode.double-living";
+  public static final String EXTRACTOR_DOUBLE_LIVING_KEY = "extractor.double-living";
+  public static final String SOURCE_DOUBLE_LIVING_KEY = "source.double-living";
   public static final boolean EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE = false;
+  public static final List<String> DOUBLE_LIVING_KEYS =
+      Collections.unmodifiableList(
+          Arrays.asList(
+              EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
+              SOURCE_MODE_DOUBLE_LIVING_KEY,
+              EXTRACTOR_DOUBLE_LIVING_KEY,
+              SOURCE_DOUBLE_LIVING_KEY));
 
   public static final String EXTRACTOR_START_TIME_KEY = "extractor.start-time";
   public static final String SOURCE_START_TIME_KEY = "source.start-time";
@@ -195,6 +211,31 @@ public class PipeSourceConstant {
   public static final String EXTRACTOR_CONSENSUS_RECEIVER_DATANODE_ID_KEY =
       "extractor.consensus.receiver-dn-id";
 
+  public static boolean isDoubleLiving(final PipeParameters sourceParameters) {
+    return sourceParameters.getBooleanOrDefault(
+        DOUBLE_LIVING_KEYS, EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+  }
+
+  public static void removeDoubleLivingAttributes(final Map<String, String> sourceAttributes) {
+    removeEquivalentAttributes(sourceAttributes, DOUBLE_LIVING_KEYS);
+  }
+
+  public static void disableForwardingPipeRequests(final Map<String, String> sourceAttributes) {
+    removeEquivalentAttributes(sourceAttributes, FORWARDING_PIPE_REQUESTS_KEYS);
+    sourceAttributes.put(SOURCE_FORWARDING_PIPE_REQUESTS_KEY, Boolean.FALSE.toString());
+  }
+
+  private static void removeEquivalentAttributes(
+      final Map<String, String> sourceAttributes, final List<String> equivalentKeys) {
+    final Set<String> reducedKeys =
+        equivalentKeys.stream()
+            .map(PipeParameters.KeyReducer::reduce)
+            .collect(java.util.stream.Collectors.toSet());
+    sourceAttributes
+        .keySet()
+        .removeIf(key -> reducedKeys.contains(PipeParameters.KeyReducer.reduce(key)));
+  }
+
   private PipeSourceConstant() {
     throw new IllegalStateException(PipeMessages.UTILITY_CLASS);
   }

diff --git a/...src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtils.java b/...src/main/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtils.java
@@ -27,16 +27,10 @@
 import org.apache.iotdb.rpc.subscription.config.TopicConfig;
 
 import java.util.Arrays;
-import java.util.List;
 import java.util.Objects;
 
 public class VisibilityUtils {
 
-  private static final List<String> DOUBLE_LIVING_KEYS =
-      Arrays.asList(
-          PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
-          PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY);
-
   private VisibilityUtils() {
     // forbidding instantiation
   }
@@ -91,8 +85,7 @@ public static Visibility calculateFromExtractorParameters(
       return calculateStrictlyFromExtractorParameters(extractorParameters);
     }
 
-    if (extractorParameters.getBooleanOrDefault(
-        DOUBLE_LIVING_KEYS, PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE)) {
+    if (PipeSourceConstant.isDoubleLiving(extractorParameters)) {
       return Visibility.BOTH;
     }
 

diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBSource.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBSource.java
@@ -115,12 +115,7 @@ public void validate(final PipeParameterValidator validator) throws Exception {
   }
 
   private void validateDoubleLiving(final PipeParameters parameters) {
-    final boolean isDoubleLiving =
-        parameters.getBooleanOrDefault(
-            Arrays.asList(
-                PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
-                PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
-            PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+    final boolean isDoubleLiving = PipeSourceConstant.isDoubleLiving(parameters);
     if (!isDoubleLiving) {
       return;
     }
@@ -149,20 +144,13 @@ public void customize(
       pipeTaskMeta = ((PipeTaskSourceRuntimeEnvironment) environment).getPipeTaskMeta();
     }
 
-    final boolean isDoubleLiving =
-        parameters.getBooleanOrDefault(
-            Arrays.asList(
-                PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
-                PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
-            PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+    final boolean isDoubleLiving = PipeSourceConstant.isDoubleLiving(parameters);
     if (isDoubleLiving) {
       isForwardingPipeRequests = false;
     } else {
       isForwardingPipeRequests =
           parameters.getBooleanOrDefault(
-              Arrays.asList(
-                  PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
-                  PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY),
+              PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS,
               PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE);
     }
 

diff --git a/...test/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtilsTest.java b/...test/java/org/apache/iotdb/commons/pipe/datastructure/visibility/VisibilityUtilsTest.java
@@ -70,6 +70,10 @@ public void testLegacyVisibilityKeepsDoubleLivingAndCaptureAttributes() {
     doubleLivingAttributes.put(PipeSourceConstant.EXTRACTOR_CAPTURE_TABLE_KEY, "false");
     assertVisibility(Visibility.BOTH, doubleLivingAttributes, true, true);
 
+    final Map<String, String> directDoubleLivingAttributes = new HashMap<>();
+    directDoubleLivingAttributes.put("double-living", "true");
+    assertVisibility(Visibility.BOTH, directDoubleLivingAttributes, true, true);
+
     final Map<String, String> captureBothAttributes = new HashMap<>();
     captureBothAttributes.put(PipeSourceConstant.EXTRACTOR_CAPTURE_TREE_KEY, "true");
     captureBothAttributes.put(PipeSourceConstant.EXTRACTOR_CAPTURE_TABLE_KEY, "true");
@@ -81,6 +85,26 @@ public void testLegacyVisibilityKeepsDoubleLivingAndCaptureAttributes() {
     assertVisibility(Visibility.NONE, captureNoneAttributes, false, false);
   }
 
+  @Test
+  public void testDoubleLivingSugarCanBeNormalizedForStrictPipes() {
+    final Map<String, String> attributes = new HashMap<>();
+    attributes.put("double-living", "true");
+    attributes.put("mode.double-living", "true");
+    attributes.put(PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY, "true");
+    attributes.put("forwarding-pipe-requests", "true");
+    attributes.put(PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY, "true");
+
+    PipeSourceConstant.removeDoubleLivingAttributes(attributes);
+    PipeSourceConstant.disableForwardingPipeRequests(attributes);
+
+    final PipeParameters parameters = new PipeParameters(attributes);
+    Assert.assertFalse(PipeSourceConstant.isDoubleLiving(parameters));
+    Assert.assertFalse(
+        parameters.getBooleanOrDefault(
+            PipeSourceConstant.FORWARDING_PIPE_REQUESTS_KEYS,
+            PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE));
+  }
+
   private void assertVisibility(
       final Visibility expectedVisibility,
       final Map<String, String> attributes,