From 84721e4269e6927dbe5686a618fa9ad29beddbfc Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 9 Jun 2026 11:20:37 +0800 Subject: [PATCH] Fix pipe receiver type conversion load path (#17849) 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 90055d55b6e1166580824e5cbed77a7c253ef514) --- .../thrift/IoTDBDataNodeReceiver.java | 33 +++++++++++--- .../thrift/IoTDBDataNodeReceiverTest.java | 43 ++++++++++++++++--- 2 files changed, 65 insertions(+), 11 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java index 1ec4d6d53b7ac..8a5c4ad060fa6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java @@ -501,7 +501,7 @@ static Map buildLoadTsFileAttributesForAsync( dataBaseName, LoadTsFileStatement.getDatabaseLevelByTreeDatabase(dataBaseName), shouldConvertDataTypeOnTypeMismatch, - validateTsFile, + validateTsFile || shouldConvertDataTypeOnTypeMismatch, null, shouldMarkAsPipeRequest); } @@ -509,16 +509,23 @@ static Map buildLoadTsFileAttributesForAsync( 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 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()) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java index f41c44763f997..1e279a18febfb 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java +++ b/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 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); + } + } }