Fix pipe receiver type conversion load path (#17849) (#17873)

Backport receiver-side tree-model changes to dev/1.3.

The 2.x table-model load analyzer changes and table/dual ITs are not applicable to dev/1.3.

(cherry picked from commit 90055d5)

Author: Caideyipi

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

@@ -501,24 +501,31 @@ static Map<String, String> buildLoadTsFileAttributesForAsync(
         dataBaseName,
         LoadTsFileStatement.getDatabaseLevelByTreeDatabase(dataBaseName),
         shouldConvertDataTypeOnTypeMismatch,
-        validateTsFile,
+        validateTsFile || shouldConvertDataTypeOnTypeMismatch,
         null,
         shouldMarkAsPipeRequest);
   }
 
   private TSStatus loadTsFileSync(final String dataBaseName, final String fileAbsolutePath)
       throws FileNotFoundException {
     return executeStatementAndClassifyExceptions(
-        buildLoadTsFileStatementForSync(dataBaseName, fileAbsolutePath, validateTsFile.get()));
+        buildLoadTsFileStatementForSync(
+            dataBaseName,
+            fileAbsolutePath,
+            validateTsFile.get(),
+            shouldConvertDataTypeOnTypeMismatch));
   }
 
   static LoadTsFileStatement buildLoadTsFileStatementForSync(
-      final String dataBaseName, final String fileAbsolutePath, final boolean validateTsFile)
+      final String dataBaseName,
+      final String fileAbsolutePath,
+      final boolean validateTsFile,
+      final boolean shouldConvertDataTypeOnTypeMismatch)
       throws FileNotFoundException {
     final LoadTsFileStatement statement = LoadTsFileStatement.createUnchecked(fileAbsolutePath);
     statement.setDeleteAfterLoad(true);
-    statement.setConvertOnTypeMismatch(true);
-    statement.setVerifySchema(validateTsFile);
+    statement.setConvertOnTypeMismatch(shouldConvertDataTypeOnTypeMismatch);
+    statement.setVerifySchema(validateTsFile || shouldConvertDataTypeOnTypeMismatch);
     statement.setAutoCreateDatabase(
         IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled());
     statement.setDatabase(dataBaseName);
@@ -769,9 +776,23 @@ private TSStatus executeStatementWithRetryOnDataTypeMismatch(final Statement sta
           TSStatusCode.PIPE_TRANSFER_EXECUTE_STATEMENT_ERROR, "Execute null statement.");
     }
 
+    // Execute insert statements through the conversion wrapper first to avoid writing a partial
+    // row/tablet before the type mismatch is converted.
+    if (shouldConvertDataTypeOnTypeMismatch && statement instanceof InsertBaseStatement) {
+      final Optional<TSStatus> convertedStatus =
+          statement.accept(
+              statementDataTypeConvertExecutionVisitor,
+              new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+      if (convertedStatus.isPresent()) {
+        return convertedStatus.get();
+      }
+    }
+
     final TSStatus status = executeStatement(statement);
 
-    // Try to convert the data type and the status code is not success
+    // Try to convert data type if the status code is not success. Insert statements normally return
+    // above after the first converted execution. The retry path is kept for load and fallback
+    // cases.
     return shouldConvertDataTypeOnTypeMismatch
             && ((statement instanceof InsertBaseStatement
                     && ((InsertBaseStatement) statement).hasFailedMeasurements())

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java

@@ -39,7 +39,7 @@ public void testLoadTsFileSyncStatementUsesTreeDatabaseLevelFromDatabaseName() t
     try {
       final LoadTsFileStatement statement =
           IoTDBDataNodeReceiver.buildLoadTsFileStatementForSync(
-              "root.test.sg_0", tsFile.toString(), true);
+              "root.test.sg_0", tsFile.toString(), true, true);
 
       Assert.assertEquals("root.test.sg_0", statement.getDatabase());
       Assert.assertEquals(2, statement.getDatabaseLevel());
@@ -54,16 +54,17 @@ public void testLoadTsFileAsyncAttributesUseTreeDatabaseLevelFromDatabaseName()
     try {
       final Map<String, String> attributes =
           IoTDBDataNodeReceiver.buildLoadTsFileAttributesForAsync(
-              "root.test.sg_0", true, true, true);
+              "root.test.sg_0", true, false, true);
 
       Assert.assertEquals(
           "root.test.sg_0", attributes.get(LoadTsFileConfigurator.DATABASE_NAME_KEY));
       Assert.assertEquals("2", attributes.get(LoadTsFileConfigurator.DATABASE_LEVEL_KEY));
 
       final LoadTsFileStatement statement = LoadTsFileStatement.createUnchecked(tsFile.toString());
-      ActiveLoadPathHelper.applyAttributesToStatement(attributes, statement, true);
+      ActiveLoadPathHelper.applyAttributesToStatement(attributes, statement, false);
       Assert.assertEquals("root.test.sg_0", statement.getDatabase());
       Assert.assertEquals(2, statement.getDatabaseLevel());
+      Assert.assertTrue(statement.isVerifySchema());
     } finally {
       Files.deleteIfExists(tsFile);
     }
@@ -75,7 +76,8 @@ public void testLoadTsFileSyncStatementKeepsDefaultDatabaseLevelWhenDatabaseName
     final Path tsFile = Files.createTempFile("pipe-load-default-database-level", ".tsfile");
     try {
       final LoadTsFileStatement statement =
-          IoTDBDataNodeReceiver.buildLoadTsFileStatementForSync(null, tsFile.toString(), true);
+          IoTDBDataNodeReceiver.buildLoadTsFileStatementForSync(
+              null, tsFile.toString(), true, true);
 
       Assert.assertNull(statement.getDatabase());
       Assert.assertEquals(
@@ -92,7 +94,7 @@ public void testRepeatedStatementExceptionLogIsReduced() throws Exception {
     try {
       final LoadTsFileStatement statement =
           IoTDBDataNodeReceiver.buildLoadTsFileStatementForSync(
-              "root.test.sg_0", tsFile.toString(), true);
+              "root.test.sg_0", tsFile.toString(), true, true);
       final long receiverId = System.nanoTime();
       final Exception exception = new RuntimeException("repeated receiver exception " + receiverId);
 
@@ -107,4 +109,35 @@ public void testRepeatedStatementExceptionLogIsReduced() throws Exception {
       Files.deleteIfExists(tsFile);
     }
   }
+
+  @Test
+  public void testLoadTsFileSyncStatementVerifiesSchemaWhenConvertingType() throws Exception {
+    final Path tsFile = Files.createTempFile("pipe-load-convert-verify-schema", ".tsfile");
+    try {
+      final LoadTsFileStatement statement =
+          IoTDBDataNodeReceiver.buildLoadTsFileStatementForSync(
+              "root.test.sg_0", tsFile.toString(), false, true);
+
+      Assert.assertTrue(statement.isConvertOnTypeMismatch());
+      Assert.assertTrue(statement.isVerifySchema());
+    } finally {
+      Files.deleteIfExists(tsFile);
+    }
+  }
+
+  @Test
+  public void testLoadTsFileSyncStatementCanSkipVerifySchemaWhenNotConvertingType()
+      throws Exception {
+    final Path tsFile = Files.createTempFile("pipe-load-no-convert-no-verify-schema", ".tsfile");
+    try {
+      final LoadTsFileStatement statement =
+          IoTDBDataNodeReceiver.buildLoadTsFileStatementForSync(
+              "root.test.sg_0", tsFile.toString(), false, false);
+
+      Assert.assertFalse(statement.isConvertOnTypeMismatch());
+      Assert.assertFalse(statement.isVerifySchema());
+    } finally {
+      Files.deleteIfExists(tsFile);
+    }
+  }
 }