Fixed the concurrency issue of region migrate and load (#16796)

* rq

* gra

* fix

* fix

* coverage

* fix

* fix

* fix

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java

@@ -28,6 +28,7 @@
 import org.apache.iotdb.commons.memory.IMemoryBlock;
 import org.apache.iotdb.commons.memory.MemoryBlockType;
 import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
+import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.consensus.ConsensusFactory;
 import org.apache.iotdb.consensus.IConsensus;
 import org.apache.iotdb.consensus.config.ConsensusConfig;
@@ -65,6 +66,11 @@ private DataRegionConsensusImpl() {
     // do nothing
   }
 
+  @TestOnly
+  public static void setInstance(final IConsensus instance) {
+    DataRegionConsensusImplHolder.INSTANCE = instance;
+  }
+
   public static IConsensus getInstance() {
     return DataRegionConsensusImplHolder.INSTANCE;
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java

@@ -93,7 +93,7 @@ public boolean takeSnapshot(File snapshotDir) {
       logger.error(
           "Exception occurs when taking snapshot for {}-{} in {}",
           region.getDatabaseName(),
-          region.getDataRegionId(),
+          region.getDataRegionIdString(),
           snapshotDir,
           e);
       return false;
@@ -109,7 +109,7 @@ public boolean takeSnapshot(File snapshotDir, String snapshotTmpId, String snaps
       logger.error(
           "Exception occurs when taking snapshot for {}-{} in {}",
           region.getDatabaseName(),
-          region.getDataRegionId(),
+          region.getDataRegionIdString(),
           snapshotDir,
           e);
       return false;
@@ -127,7 +127,7 @@ public void loadSnapshot(File latestSnapshotRootDir) {
         new SnapshotLoader(
                 latestSnapshotRootDir.getAbsolutePath(),
                 region.getDatabaseName(),
-                region.getDataRegionId())
+                region.getDataRegionIdString())
             .loadSnapshotForStateMachine();
     if (newRegion == null) {
       logger.error("Fail to load snapshot from {}", latestSnapshotRootDir);
@@ -136,7 +136,8 @@ public void loadSnapshot(File latestSnapshotRootDir) {
     this.region = newRegion;
     try {
       StorageEngine.getInstance()
-          .setDataRegion(new DataRegionId(Integer.parseInt(region.getDataRegionId())), region);
+          .setDataRegion(
+              new DataRegionId(Integer.parseInt(region.getDataRegionIdString())), region);
       ChunkCache.getInstance().clear();
       TimeSeriesMetadataCache.getInstance().clear();
       BloomFilterCache.getInstance().clear();
@@ -185,13 +186,13 @@ public List<File> getSnapshotFiles(File latestSnapshotRootDir) {
       return new SnapshotLoader(
               latestSnapshotRootDir.getAbsolutePath(),
               region.getDatabaseName(),
-              region.getDataRegionId())
+              region.getDataRegionIdString())
           .getSnapshotFileInfo();
     } catch (IOException e) {
       logger.error(
           "Meets error when getting snapshot files for {}-{}",
           region.getDatabaseName(),
-          region.getDataRegionId(),
+          region.getDataRegionIdString(),
           e);
       return null;
     }
@@ -276,7 +277,7 @@ public File getSnapshotRoot() {
               + File.separator
               + region.getDatabaseName()
               + "-"
-              + region.getDataRegionId();
+              + region.getDataRegionIdString();
       return new File(snapshotDir).getCanonicalFile();
     } catch (IOException | NullPointerException e) {
       logger.warn("{}: cannot get the canonical file of {} due to {}", this, snapshotDir, e);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java

@@ -779,7 +779,7 @@ private IQueryDataSource getUnfinishedQueryDataSource() {
     if (initQueryDataSourceRetryCount % 10 == 0) {
       LOGGER.warn(
           "Failed to acquire the read lock of DataRegion-{} for {} times",
-          dataRegion == null ? "UNKNOWN" : dataRegion.getDataRegionId(),
+          dataRegion == null ? "UNKNOWN" : dataRegion.getDataRegionIdString(),
           initQueryDataSourceRetryCount);
     }
     return UNFINISHED_QUERY_DATA_SOURCE;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecution.java

@@ -152,7 +152,7 @@ private boolean fillFragmentInstanceStatistics(
       // We don't need to output the region having ExplainAnalyzeOperator only.
       return false;
     }
-    statistics.setDataRegion(context.getDataRegion().getDataRegionId());
+    statistics.setDataRegion(context.getDataRegion().getDataRegionIdString());
     statistics.setIp(CONFIG.getInternalAddress() + ":" + CONFIG.getInternalPort());
     statistics.setStartTimeInMS(context.getStartTime());
     statistics.setEndTimeInMS(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java

@@ -531,7 +531,7 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException {
                           MemTableFlushTask.recordFlushPointsMetricInternal(
                               node.getWritePointCount(),
                               databaseName,
-                              dataRegion.getDataRegionId());
+                              dataRegion.getDataRegionIdString());
 
                           MetricService.getInstance()
                               .count(
@@ -543,7 +543,7 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException {
                                   Tag.DATABASE.toString(),
                                   databaseName,
                                   Tag.REGION.toString(),
-                                  dataRegion.getDataRegionId(),
+                                  dataRegion.getDataRegionIdString(),
                                   Tag.TYPE.toString(),
                                   Metric.LOAD_POINT_COUNT.toString());
                           MetricService.getInstance()
@@ -556,7 +556,7 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException {
                                   Tag.DATABASE.toString(),
                                   databaseName,
                                   Tag.REGION.toString(),
-                                  dataRegion.getDataRegionId(),
+                                  dataRegion.getDataRegionIdString(),
                                   Tag.TYPE.toString(),
                                   Metric.LOAD_POINT_COUNT.toString());
                         }));

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/WritingMetrics.java

@@ -520,7 +520,8 @@ public void unbindDataRegionMetrics() {
   }
 
   public void createDataRegionMemoryCostMetrics(DataRegion dataRegion) {
-    DataRegionId dataRegionId = new DataRegionId(Integer.parseInt(dataRegion.getDataRegionId()));
+    DataRegionId dataRegionId =
+        new DataRegionId(Integer.parseInt(dataRegion.getDataRegionIdString()));
     MetricService.getInstance()
         .createAutoGauge(
             Metric.DATA_REGION_MEM_COST.toString(),

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java

@@ -529,7 +529,7 @@ public void syncCloseProcessorsInDatabase(String databaseName) {
   public void syncCloseProcessorsInRegion(List<String> dataRegionIds) {
     List<Future<Void>> tasks = new ArrayList<>();
     for (DataRegion dataRegion : dataRegionMap.values()) {
-      if (dataRegion != null && dataRegionIds.contains(dataRegion.getDataRegionId())) {
+      if (dataRegion != null && dataRegionIds.contains(dataRegion.getDataRegionIdString())) {
         tasks.add(
             cachedThreadPool.submit(
                 () -> {
@@ -785,7 +785,9 @@ public void deleteDataRegion(DataRegionId regionId) {
             // delete wal
             WALManager.getInstance()
                 .deleteWALNode(
-                    region.getDatabaseName() + FILE_NAME_SEPARATOR + region.getDataRegionId());
+                    region.getDatabaseName()
+                        + FILE_NAME_SEPARATOR
+                        + region.getDataRegionIdString());
             // delete snapshot
             for (String dataDir : CONFIG.getLocalDataDirs()) {
               File regionSnapshotDir =
@@ -805,7 +807,7 @@ public void deleteDataRegion(DataRegionId regionId) {
             // delete region information in wal and may delete wal
             WALManager.getInstance()
                 .deleteRegionAndMayDeleteWALNode(
-                    region.getDatabaseName(), region.getDataRegionId());
+                    region.getDatabaseName(), region.getDataRegionIdString());
             break;
           case ConsensusFactory.RATIS_CONSENSUS:
           default:
@@ -814,14 +816,15 @@ public void deleteDataRegion(DataRegionId regionId) {
         WRITING_METRICS.removeDataRegionMemoryCostMetrics(regionId);
         WRITING_METRICS.removeFlushingMemTableStatusMetrics(regionId);
         WRITING_METRICS.removeActiveMemtableCounterMetrics(regionId);
-        FileMetrics.getInstance().deleteRegion(region.getDatabaseName(), region.getDataRegionId());
+        FileMetrics.getInstance()
+            .deleteRegion(region.getDatabaseName(), region.getDataRegionIdString());
         CompressionRatio.getInstance().removeDataRegionRatio(String.valueOf(regionId.getId()));
         LOGGER.info("Removed data region {}", regionId);
       } catch (Exception e) {
         LOGGER.error(
             "Error occurs when deleting data region {}-{}",
             region.getDatabaseName(),
-            region.getDataRegionId(),
+            region.getDataRegionIdString(),
             e);
       } finally {
         deletingDataRegionMap.remove(regionId);