Pipe: Fixed the bug that the split historical pipe's enable-send-tsfile-limit cannot be configured (#17598)

Author: Caideyipi

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeAutoSplitIT.java

@@ -140,9 +140,21 @@ public void testSingleEnv() throws Exception {
             "insert into root.test.device(time, field) values(0,1),(1,2)",
             "delete from root.test.device.* where time == 0",
             String.format(
-                "create pipe a2b with source ('inclusion'='all') with sink ('node-urls'='%s')",
+                "create pipe a2b with source ('inclusion'='all') with sink "
+                    + "('node-urls'='%s', 'enable-send-tsfile-limit'='false')",
                 receiverDataNode.getIpAndPortString())));
 
+    try (final SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+      final List<TShowPipeInfo> showPipeResult =
+          client.showPipe(new TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)).pipeInfoList;
+      showPipeResult.removeIf(i -> i.getId().startsWith("__consensus"));
+      Assert.assertTrue(
+          showPipeResult.stream()
+              .filter(i -> Objects.equals(i.id, "a2b_history"))
+              .anyMatch(i -> i.pipeConnector.contains("enable-send-tsfile-limit=false")));
+    }
+
     TestUtils.assertDataEventuallyOnEnv(
         receiverEnv,
         "select * from root.test.device",

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

@@ -2260,6 +2260,19 @@ public SettableFuture<ConfigTaskResult> createPipe(
         }
 
         // 2. Send request to create the historical data synchronization pipeline
+        final Map<String, String> historySinkAttributes =
+            sinkPipeParameters.hasAnyAttributes(
+                    PipeSinkConstant.SINK_ENABLE_SEND_TSFILE_LIMIT,
+                    PipeSinkConstant.CONNECTOR_ENABLE_SEND_TSFILE_LIMIT)
+                ? createPipeStatement.getSinkAttributes()
+                : sinkPipeParameters
+                    .addOrReplaceEquivalentAttributesWithClone(
+                        new PipeParameters(
+                            Collections.singletonMap(
+                                PipeSinkConstant.SINK_ENABLE_SEND_TSFILE_LIMIT,
+                                Boolean.TRUE.toString())))
+                    .getAttribute();
+
         final TCreatePipeReq historyReq =
             new TCreatePipeReq()
                 // Append suffix to the pipeline name for historical data
@@ -2292,14 +2305,7 @@ public SettableFuture<ConfigTaskResult> createPipe(
                                     PipeSourceConstant.EXTRACTOR_EXCLUSION_DEFAULT_VALUE)))
                         .getAttribute())
                 .setProcessorAttributes(createPipeStatement.getProcessorAttributes())
-                .setConnectorAttributes(
-                    sinkPipeParameters
-                        .addOrReplaceEquivalentAttributesWithClone(
-                            new PipeParameters(
-                                Collections.singletonMap(
-                                    PipeSinkConstant.SINK_ENABLE_SEND_TSFILE_LIMIT,
-                                    Boolean.TRUE.toString())))
-                        .getAttribute());
+                .setConnectorAttributes(historySinkAttributes);
 
         final TSStatus historyTsStatus = configNodeClient.createPipe(historyReq);
         // If creation fails, immediately return with exception