Pipe: Fixed the remaining correctness bug for PR 17661 && the replication configuration of IoTDBPipeReceiverAutoCreateDisabledIT (#17678)

* Update IoTDBPipeReceiverAutoCreateDisabledIT.java

* Fix

* Update IoTDBPipeReceiverAutoCreateDisabledIT.java

* Update IoTDBDataNodeReceiver.java

* Update IoTDBDataNodeReceiver.java

* Fix

Author: Caideyipi

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

@@ -57,7 +57,17 @@ public void setUp() {
   @Override
   protected void setupConfig() {
     super.setupConfig();
-    receiverEnv.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(false);
+    senderEnv
+        .getConfig()
+        .getCommonConfig()
+        .setDataReplicationFactor(1)
+        .setSchemaReplicationFactor(1);
+    receiverEnv
+        .getConfig()
+        .getCommonConfig()
+        .setAutoCreateSchemaEnabled(false)
+        .setDataReplicationFactor(1)
+        .setSchemaReplicationFactor(1);
   }
 
   @Test
@@ -69,10 +79,12 @@ public void testReceiverAutoCreateSchemaDisabledWithSpecialTimeSeries() throws E
 
     final String createPipeSql =
         String.format(
-            "create pipe test with source ('inclusion'='all','source.realtime.mode'='stream','source.realtime.enable'='true') "
+            "create pipe test with source ('inclusion'='all') "
                 + "with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s');",
             receiverEnv.getDataNodeWrapper(0).getIpAndPortString());
     final String createDatabaseSql = "create database root.test.sg;";
+    final String createSecondDatabaseSql =
+        "create database root.test.ABCDEFGHIJKLMNOPQRSTUVWXYZ_abcdefghijklmnopqrstuvwxyz1;";
     final String createFirstTimeSeriesSql =
         "create timeseries root.test.sg.`1~!@#$%^&*()_+=:'\"/|[]{}`.`~!@#$%^&*()_+=:'\"/|[]{}` float;";
     final String insertFirstSql =
@@ -95,6 +107,7 @@ public void testReceiverAutoCreateSchemaDisabledWithSpecialTimeSeries() throws E
       statement.execute(createFirstTimeSeriesSql);
       statement.execute(insertFirstSql);
       final QueryResult firstQueryResult = queryForResult(statement, firstSelectSql);
+      statement.execute(createSecondDatabaseSql);
       statement.execute(createSecondTimeSeriesSql);
       statement.execute(insertSecondSql);
       final QueryResult secondQueryResult = queryForResult(statement, secondSelectSql);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java

@@ -871,9 +871,11 @@ private TSStatus executeStatementWithPermissionCheckAndRetryOnDataTypeMismatch(
     // Judge which model the statement belongs to
     final boolean isTableModelStatement;
     final String databaseName;
-    if (statement instanceof LoadTsFileStatement
-        && ((LoadTsFileStatement) statement).getDatabase() != null) {
-      isTableModelStatement = true;
+    if (statement instanceof LoadTsFileStatement) {
+      // Pipe receiver always constructs a tree-model LoadTsFileStatement. Its database field is
+      // only an explicit database hint for table data or pipe-generated tree-model loads, so it
+      // must not be used to route execution into the table-model pipeline.
+      isTableModelStatement = false;
       databaseName = ((LoadTsFileStatement) statement).getDatabase();
     } else if (statement instanceof InsertBaseStatement
         && ((InsertBaseStatement) statement).isWriteToTable()) {
@@ -925,18 +927,34 @@ private TSStatus executeStatementWithPermissionCheckAndRetryOnDataTypeMismatch(
 
     // Try to convert data type if the statement is a tree model statement
     // and the status code is not success
-    return shouldConvertDataTypeOnTypeMismatch
-            && ((statement instanceof InsertBaseStatement
-                    && ((InsertBaseStatement) statement).hasFailedMeasurements())
-                || (status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()
-                    && status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()))
-        ? (isTableModelStatement
-            ? statement
-                .accept(
-                    tableStatementDataTypeConvertExecutionVisitor, new Pair<>(status, databaseName))
-                .orElse(status)
-            : statement.accept(treeStatementDataTypeConvertExecutionVisitor, status).orElse(status))
-        : status;
+    if (!shouldConvertDataTypeOnTypeMismatch
+        || !((statement instanceof InsertBaseStatement
+                && ((InsertBaseStatement) statement).hasFailedMeasurements())
+            || (status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()
+                && status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()))) {
+      return status;
+    }
+
+    if (statement instanceof LoadTsFileStatement
+        && shouldUseTableModelVisitorForLoadStatement((LoadTsFileStatement) statement)) {
+      return statement
+          .accept(tableStatementDataTypeConvertExecutionVisitor, new Pair<>(status, databaseName))
+          .orElse(status);
+    }
+
+    return isTableModelStatement
+        ? statement
+            .accept(tableStatementDataTypeConvertExecutionVisitor, new Pair<>(status, databaseName))
+            .orElse(status)
+        : statement.accept(treeStatementDataTypeConvertExecutionVisitor, status).orElse(status);
+  }
+
+  private boolean shouldUseTableModelVisitorForLoadStatement(
+      final LoadTsFileStatement loadTsFileStatement) {
+    final List<Boolean> isTableModel = loadTsFileStatement.getIsTableModel();
+    return Objects.nonNull(isTableModel)
+        && !isTableModel.isEmpty()
+        && isTableModel.stream().allMatch(Boolean.TRUE::equals);
   }
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java

@@ -341,7 +341,7 @@ private void doTransfer(
               tsFile.length(),
               pipeTsFileInsertionEvent.isTableModelEvent()
                   ? pipeTsFileInsertionEvent.getTableModelDatabaseName()
-                  : null))) {
+                  : pipeTsFileInsertionEvent.getTreeModelDatabaseName()))) {
         receiverStatusHandler.handle(
             new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode())
                 .setMessage(errorMessage),
@@ -362,7 +362,7 @@ private void doTransfer(
               tsFile.length(),
               pipeTsFileInsertionEvent.isTableModelEvent()
                   ? pipeTsFileInsertionEvent.getTableModelDatabaseName()
-                  : null))) {
+                  : pipeTsFileInsertionEvent.getTreeModelDatabaseName()))) {
         receiverStatusHandler.handle(
             new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode())
                 .setMessage(errorMessage),

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java

@@ -423,7 +423,7 @@ private boolean transferWithoutCheck(final TsFileInsertionEvent tsFileInsertionE
                   && clientManager.supportModsIfIsDataNodeReceiver(),
               pipeTsFileInsertionEvent.isTableModelEvent()
                   ? pipeTsFileInsertionEvent.getTableModelDatabaseName()
-                  : null);
+                  : pipeTsFileInsertionEvent.getTreeModelDatabaseName());
 
       transfer(pipeTransferTsFileHandler);
       return true;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java

@@ -499,7 +499,7 @@ private void doTransferWrapper(final PipeTsFileInsertionEvent pipeTsFileInsertio
           pipeTsFileInsertionEvent.isWithMod() ? pipeTsFileInsertionEvent.getModFile() : null,
           pipeTsFileInsertionEvent.isTableModelEvent()
               ? pipeTsFileInsertionEvent.getTableModelDatabaseName()
-              : null);
+              : pipeTsFileInsertionEvent.getTreeModelDatabaseName());
     } finally {
       pipeTsFileInsertionEvent.decreaseReferenceCount(
           IoTDBDataRegionSyncSink.class.getName(), false);