Load: Fixed multiple bugs (#17413)

Caideyipi · web-flow · commit 2b7b6cb9063f · 2026-04-02T09:37:39.000+08:00
* fix

* load

* Update ClusterConfigTaskExecutor.java

* fix
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java
@@ -565,59 +565,52 @@ public void testSpecialPartialInsert() throws Exception {
 
   @Test
   public void testTransferMods() {
-    try {
-      TestUtils.executeNonQueries(
-          senderEnv,
-          Arrays.asList(
-              "create database root.sg_nonAligned",
-              "create TIMESERIES root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s0 with datatype=boolean, encoding=RLE,compressor=snappy",
-              "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s1 with datatype=int32, encoding=PLAIN,compressor=LZ4",
-              "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s2 with datatype=int64,encoding=gorilla,compressor=uncompressed",
-              "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s3 with datatype=float,encoding=chimp,compressor=gzip",
-              "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s4 with datatype=double,encoding=ts_2diff,compressor=zstd",
-              "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s5 with datatype=text,encoding=dictionary,compressor=lzma2",
-              "insert into root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`(time,s1, s2,s3,s4,s0,s5) values(1,1,10,5.39,5.51234,true,''),(11,null,20,5.39,15.51234,false,'第2条 device_nonAligned'),(21,3,null,5.39,25.51234,true,'第3条device_nonAligned'),(31,4,40,null,35.51234,true,'第4条device_nonAligned'),(41,5,50,5.39,null,false,'第5条device_nonAligned'),(51,6,60,5.39,55.51234,null,'第6条device_nonAligned'),(61,7,70,5.39,65.51234,false,null),(71,8,80,5.39,75.51234,false,'第8条device_nonAligned'),(81,9,90,5.39,85.51234,false,'第9条device_nonAligned'),(91,10,100,5.39,95.51234,false,'第10条device_nonAligned'),(101,11,110,5.39,105.51234,false,'第11条device_nonAligned'),(111,12,120,5.39,115.51234,false,'第12条device_nonAligned'),(121,13,130,5.39,125.51234,false,'第13条device_nonAligned'),(131,14,140,5.39,135.51234,false,'第14条device_nonAligned'),(141,15,150,5.39,145.51234,false,'第15条'),(151,16,160,5.39,155.51234,false,'第16条'),(161,17,170,5.39,165.51234,false,'第17条'),(171,18,180,5.39,175.51234,false,'第18条'),(181,19,190,5.39,185.51234,false,'第19条'),(191,20,200,5.39,195.51234,false,'第20条'),(201,21,210,5.39,null,false,'第21条')",
-              "flush",
-              "delete timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.s0",
-              String.format(
-                  "create pipe test with source ('source.realtime.mode'='stream','inclusion'='data','path'='root.sg_nonAligned.非对齐序列带有encoding和压缩方式.**','source.realtime.enable'='true','mods.enable'='true') with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s')",
-                  receiverEnv.getDataNodeWrapperList().get(0).getIpAndPortString())));
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList(
+            "create database root.sg_nonAligned",
+            "create TIMESERIES root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s0 with datatype=boolean, encoding=RLE,compressor=snappy",
+            "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s1 with datatype=int32, encoding=PLAIN,compressor=LZ4",
+            "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s2 with datatype=int64,encoding=gorilla,compressor=uncompressed",
+            "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s3 with datatype=float,encoding=chimp,compressor=gzip",
+            "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s4 with datatype=double,encoding=ts_2diff,compressor=zstd",
+            "create timeseries root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`.s5 with datatype=text,encoding=dictionary,compressor=lzma2",
+            "insert into root.sg_nonAligned.`非对齐序列带有encoding和压缩方式`(time,s1, s2,s3,s4,s0,s5) values(1,1,10,5.39,5.51234,true,''),(11,null,20,5.39,15.51234,false,'第2条 device_nonAligned'),(21,3,null,5.39,25.51234,true,'第3条device_nonAligned'),(31,4,40,null,35.51234,true,'第4条device_nonAligned'),(41,5,50,5.39,null,false,'第5条device_nonAligned'),(51,6,60,5.39,55.51234,null,'第6条device_nonAligned'),(61,7,70,5.39,65.51234,false,null),(71,8,80,5.39,75.51234,false,'第8条device_nonAligned'),(81,9,90,5.39,85.51234,false,'第9条device_nonAligned'),(91,10,100,5.39,95.51234,false,'第10条device_nonAligned'),(101,11,110,5.39,105.51234,false,'第11条device_nonAligned'),(111,12,120,5.39,115.51234,false,'第12条device_nonAligned'),(121,13,130,5.39,125.51234,false,'第13条device_nonAligned'),(131,14,140,5.39,135.51234,false,'第14条device_nonAligned'),(141,15,150,5.39,145.51234,false,'第15条'),(151,16,160,5.39,155.51234,false,'第16条'),(161,17,170,5.39,165.51234,false,'第17条'),(171,18,180,5.39,175.51234,false,'第18条'),(181,19,190,5.39,185.51234,false,'第19条'),(191,20,200,5.39,195.51234,false,'第20条'),(201,21,210,5.39,null,false,'第21条')",
+            "flush",
+            "delete timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.s0",
+            String.format(
+                "create pipe test with source ('source.realtime.mode'='stream','inclusion'='data','path'='root.sg_nonAligned.非对齐序列带有encoding和压缩方式.**','source.realtime.enable'='true','mods.enable'='true') with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s')",
+                receiverEnv.getDataNodeWrapperList().get(0).getIpAndPortString())));
 
-      TestUtils.assertDataEventuallyOnEnv(
-          receiverEnv,
-          "count timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.*",
-          "count(timeseries),",
-          Collections.singleton("5,"));
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "count timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.*",
+        "count(timeseries),",
+        Collections.singleton("5,"));
 
-      TestUtils.executeNonQueries(
-          senderEnv, Arrays.asList("drop pipe test_history", "drop pipe test_realtime"));
+    TestUtils.executeNonQueries(
+        senderEnv, Arrays.asList("drop pipe test_history", "drop pipe test_realtime"));
 
-      TestUtils.executeNonQuery(receiverEnv, "drop database root.**");
+    TestUtils.executeNonQuery(receiverEnv, "drop database root.**");
 
-      TestUtils.executeNonQueries(
-          senderEnv,
-          Arrays.asList(
-              "delete timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.s1",
-              String.format(
-                  "create pipe test with source ('source.realtime.mode'='stream','inclusion'='data','path'='root.sg_nonAligned.非对齐序列带有encoding和压缩方式.**','source.realtime.enable'='true','mods.enable'='true') with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s')",
-                  receiverEnv.getDataNodeWrapperList().get(0).getIpAndPortString())));
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList(
+            "delete timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.s1",
+            String.format(
+                "create pipe test with source ('source.realtime.mode'='stream','inclusion'='data','path'='root.sg_nonAligned.非对齐序列带有encoding和压缩方式.**','source.realtime.enable'='true','mods.enable'='true') with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s')",
+                receiverEnv.getDataNodeWrapperList().get(0).getIpAndPortString())));
 
-      TestUtils.assertDataEventuallyOnEnv(
-          receiverEnv,
-          "count timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.*",
-          "count(timeseries),",
-          Collections.singleton("4,"),
-          15);
-      TestUtils.assertDataAlwaysOnEnv(
-          receiverEnv,
-          "count timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.*",
-          "count(timeseries),",
-          Collections.singleton("4,"));
-    } finally {
-      TestUtils.executeNonQueries(
-          senderEnv,
-          Arrays.asList(
-              "drop pipe test_history", "drop pipe test_realtime", "drop database root.**"));
-    }
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "count timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.*",
+        "count(timeseries),",
+        Collections.singleton("4,"),
+        15);
+    TestUtils.assertDataAlwaysOnEnv(
+        receiverEnv,
+        "count timeseries root.sg_nonAligned.非对齐序列带有encoding和压缩方式.*",
+        "count(timeseries),",
+        Collections.singleton("4,"));
   }
 }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java
@@ -387,10 +387,10 @@ private void verifyTableDataTypeAndGenerateTagColumnMapper(
             && (realColumn == null || !fileColumn.getType().equals(realColumn.getType()))) {
           LOGGER.debug(
               "Data type mismatch for column {} in table {}, type in TsFile: {}, type in IoTDB: {}",
-              realColumn.getName(),
+              fileColumn.getName(),
               realSchema.getTableName(),
               fileColumn.getType(),
-              realColumn.getType());
+              Objects.nonNull(realColumn) ? realColumn.getType() : null);
         }
       }
     }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -3514,6 +3514,7 @@ public SettableFuture<ConfigTaskResult> removeDataNode(
         future.setException(
             new IOException(
                 "The DataNode to be removed is not in the cluster, or the input format is incorrect."));
+        return future;
       }
 
       LOGGER.info("Starting to remove DataNode with nodeIds: {}", nodeIds);
@@ -3583,6 +3584,7 @@ public SettableFuture<ConfigTaskResult> removeConfigNode(
         future.setException(
             new IOException(
                 "The ConfigNode to be removed is not in the cluster, or the input format is incorrect."));
+        return future;
       }
 
       TConfigNodeLocation configNodeLocation = removeConfigNodeLocations.get(0);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadTsFileNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadTsFileNode.java
@@ -152,6 +152,8 @@ private List<WritePlanNode> splitByPartitionForTableModel(
                 statement.isDeleteAfterLoad(),
                 statement.getWritePointCount(i),
                 needDecode4TimeColumn));
+      } else {
+        throw new IllegalStateException("LoadTsFile statement is null during table model split.");
       }
     }
     return res;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
@@ -286,7 +286,10 @@ public void start() {
         final StringBuilder failedTsFiles =
             new StringBuilder(
                 !tsFileNodeList.isEmpty()
-                    ? tsFileNodeList.get(0).getTsFileResource().getTsFilePath()
+                    ? tsFileNodeList
+                        .get(failedTsFileNodeIndexes.get(0))
+                        .getTsFileResource()
+                        .getTsFilePath()
                     : "");
         final ListIterator<Integer> iterator = failedTsFileNodeIndexes.listIterator(1);
         while (iterator.hasNext()) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java
@@ -224,7 +224,7 @@ public void writeToDataRegion(DataRegion dataRegion, LoadTsFilePieceNode pieceNo
       }
     }
 
-    final Optional<CleanupTask> cleanupTask = Optional.of(uuid2CleanupTask.get(uuid));
+    final Optional<CleanupTask> cleanupTask = Optional.ofNullable(uuid2CleanupTask.get(uuid));
     cleanupTask.ifPresent(CleanupTask::markLoadTaskRunning);
     try {
       final AtomicReference<Exception> exception = new AtomicReference<>();
@@ -304,7 +304,7 @@ public boolean loadAll(
       return false;
     }
 
-    final Optional<CleanupTask> cleanupTask = Optional.of(uuid2CleanupTask.get(uuid));
+    final Optional<CleanupTask> cleanupTask = Optional.ofNullable(uuid2CleanupTask.get(uuid));
     cleanupTask.ifPresent(CleanupTask::markLoadTaskRunning);
     try {
       uuid2WriterManager.get(uuid).loadAll(isGeneratedByPipe, timePartitionProgressIndexMap);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPendingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPendingQueue.java
@@ -70,11 +70,11 @@ public synchronized boolean isFilePendingOrLoading(final String file) {
     return loadingFileSet.contains(file) || pendingFileSet.contains(file);
   }
 
-  public int size() {
+  public synchronized int size() {
     return pendingFileQueue.size() + loadingFileSet.size();
   }
 
-  public boolean isEmpty() {
+  public synchronized boolean isEmpty() {
     return pendingFileQueue.isEmpty() && loadingFileSet.isEmpty();
   }
 
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/config/LoadTsFileConfigurator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/config/LoadTsFileConfigurator.java
@@ -83,14 +83,14 @@ public static void validateSynonymParameters(final Map<String, String> parameter
 
   public static void validateDatabaseLevelParam(final String databaseLevel) {
     try {
-      int level = Integer.parseInt(databaseLevel);
+      final int level = Integer.parseInt(databaseLevel);
       if (level < DATABASE_LEVEL_MIN_VALUE) {
         throw new SemanticException(
             String.format(
                 "Given database level %d is less than the minimum value %d, please input a valid database level.",
                 level, DATABASE_LEVEL_MIN_VALUE));
       }
-    } catch (Exception e) {
+    } catch (final NumberFormatException e) {
       throw new SemanticException(
           String.format(
               "Given database level %s is not a valid integer, please input a valid database level.",
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManager.java
@@ -156,7 +156,7 @@ synchronized void forceResize(LoadTsFileMemoryBlock memoryBlock, long newSizeInB
     try {
       forceAllocateFromQuery(bytesNeeded);
       if (LOGGER.isDebugEnabled()) {
-        LOGGER.info(
+        LOGGER.debug(
             "Load: Force resized LoadTsFileMemoryBlock with memory from query engine, size added: {}, new size: {}",
             bytesNeeded,
             newSizeInBytes);
@@ -180,7 +180,6 @@ public synchronized LoadTsFileDataCacheMemoryBlock allocateDataCacheMemoryBlock(
       final long actuallyAllocateMemoryInBytes =
           tryAllocateFromQuery(MEMORY_TOTAL_SIZE_FROM_QUERY_IN_BYTES >> 2);
       dataCacheMemoryBlock = new LoadTsFileDataCacheMemoryBlock(actuallyAllocateMemoryInBytes);
-      usedMemorySizeInBytes.addAndGet(actuallyAllocateMemoryInBytes);
       LOGGER.info(
           "Create Data Cache Memory Block {}, allocate memory {}",
           dataCacheMemoryBlock,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/metrics/LoadTsFileCostMetricsSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/metrics/LoadTsFileCostMetricsSet.java
@@ -159,7 +159,7 @@ public void unbindFrom(AbstractMetricService metricService) {
                     stage));
 
     metricService.remove(
-        MetricType.RATE,
+        MetricType.COUNTER,
         Metric.LOAD_DISK_IO.toString(),
         Tag.NAME.toString(),
         String.valueOf(IoTDBDescriptor.getInstance().getConfig().getDataNodeId()));

Original file line number	Diff line number	Diff line change
`@@ -387,10 +387,10 @@ private void verifyTableDataTypeAndGenerateTagColumnMapper(`
`387`	`387`	`&& (realColumn == null \|\| !fileColumn.getType().equals(realColumn.getType()))) {`
`388`	`388`	`LOGGER.debug(`
`389`	`389`	`"Data type mismatch for column {} in table {}, type in TsFile: {}, type in IoTDB: {}",`
`390`		`- realColumn.getName(),`
	`390`	`+ fileColumn.getName(),`
`391`	`391`	`realSchema.getTableName(),`
`392`	`392`	`fileColumn.getType(),`
`393`		`- realColumn.getType());`
	`393`	`+ Objects.nonNull(realColumn) ? realColumn.getType() : null);`
`394`	`394`	`}`
`395`	`395`	`}`
`396`	`396`	`}`
Original file line number	Diff line number	Diff line change
`@@ -3514,6 +3514,7 @@ public SettableFuture<ConfigTaskResult> removeDataNode(`
`3514`	`3514`	`future.setException(`
`3515`	`3515`	`new IOException(`
`3516`	`3516`	`"The DataNode to be removed is not in the cluster, or the input format is incorrect."));`
	`3517`	`+ return future;`
`3517`	`3518`	`}`
`3518`	`3519`
`3519`	`3520`	`LOGGER.info("Starting to remove DataNode with nodeIds: {}", nodeIds);`
`@@ -3583,6 +3584,7 @@ public SettableFuture<ConfigTaskResult> removeConfigNode(`
`3583`	`3584`	`future.setException(`
`3584`	`3585`	`new IOException(`
`3585`	`3586`	`"The ConfigNode to be removed is not in the cluster, or the input format is incorrect."));`
	`3587`	`+ return future;`
`3586`	`3588`	`}`
`3587`	`3589`
`3588`	`3590`	`TConfigNodeLocation configNodeLocation = removeConfigNodeLocations.get(0);`
Original file line number	Diff line number	Diff line change
`@@ -152,6 +152,8 @@ private List<WritePlanNode> splitByPartitionForTableModel(`
`152`	`152`	`statement.isDeleteAfterLoad(),`
`153`	`153`	`statement.getWritePointCount(i),`
`154`	`154`	`needDecode4TimeColumn));`
	`155`	`+ } else {`
	`156`	`+ throw new IllegalStateException("LoadTsFile statement is null during table model split.");`
`155`	`157`	`}`
`156`	`158`	`}`
`157`	`159`	`return res;`
Original file line number	Diff line number	Diff line change
`@@ -224,7 +224,7 @@ public void writeToDataRegion(DataRegion dataRegion, LoadTsFilePieceNode pieceNo`
`224`	`224`	`}`
`225`	`225`	`}`
`226`	`226`
`227`		`- final Optional<CleanupTask> cleanupTask = Optional.of(uuid2CleanupTask.get(uuid));`
	`227`	`+ final Optional<CleanupTask> cleanupTask = Optional.ofNullable(uuid2CleanupTask.get(uuid));`
`228`	`228`	`cleanupTask.ifPresent(CleanupTask::markLoadTaskRunning);`
`229`	`229`	`try {`
`230`	`230`	`final AtomicReference<Exception> exception = new AtomicReference<>();`
`@@ -304,7 +304,7 @@ public boolean loadAll(`
`304`	`304`	`return false;`
`305`	`305`	`}`
`306`	`306`
`307`		`- final Optional<CleanupTask> cleanupTask = Optional.of(uuid2CleanupTask.get(uuid));`
	`307`	`+ final Optional<CleanupTask> cleanupTask = Optional.ofNullable(uuid2CleanupTask.get(uuid));`
`308`	`308`	`cleanupTask.ifPresent(CleanupTask::markLoadTaskRunning);`
`309`	`309`	`try {`
`310`	`310`	`uuid2WriterManager.get(uuid).loadAll(isGeneratedByPipe, timePartitionProgressIndexMap);`
Original file line number	Diff line number	Diff line change
`@@ -70,11 +70,11 @@ public synchronized boolean isFilePendingOrLoading(final String file) {`
`70`	`70`	`return loadingFileSet.contains(file) \|\| pendingFileSet.contains(file);`
`71`	`71`	`}`
`72`	`72`
`73`		`- public int size() {`
	`73`	`+ public synchronized int size() {`
`74`	`74`	`return pendingFileQueue.size() + loadingFileSet.size();`
`75`	`75`	`}`
`76`	`76`
`77`		`- public boolean isEmpty() {`
	`77`	`+ public synchronized boolean isEmpty() {`
`78`	`78`	`return pendingFileQueue.isEmpty() && loadingFileSet.isEmpty();`
`79`	`79`	`}`
`80`	`80`