WIP: Introduce Consistency Check And Repair

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java

@@ -73,4 +73,18 @@ public ConfigNodeConfig setLeaderDistributionPolicy(String policy) {
     properties.setProperty("leader_distribution_policy", policy);
     return this;
   }
+
+  @Override
+  public ConfigNodeConfig setConsistencyCheckSchedulerInitialDelayInMs(long initialDelayInMs) {
+    properties.setProperty(
+        "consistency_check_scheduler_initial_delay_in_ms", String.valueOf(initialDelayInMs));
+    return this;
+  }
+
+  @Override
+  public ConfigNodeConfig setConsistencyCheckSchedulerIntervalInMs(long intervalInMs) {
+    properties.setProperty(
+        "consistency_check_scheduler_interval_in_ms", String.valueOf(intervalInMs));
+    return this;
+  }
 }

integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteConfigNodeConfig.java

@@ -43,4 +43,14 @@ public ConfigNodeConfig setMetricPrometheusReporterPassword(String password) {
   public ConfigNodeConfig setLeaderDistributionPolicy(String policy) {
     return this;
   }
+
+  @Override
+  public ConfigNodeConfig setConsistencyCheckSchedulerInitialDelayInMs(long initialDelayInMs) {
+    return this;
+  }
+
+  @Override
+  public ConfigNodeConfig setConsistencyCheckSchedulerIntervalInMs(long intervalInMs) {
+    return this;
+  }
 }

integration-test/src/main/java/org/apache/iotdb/itbase/env/ConfigNodeConfig.java

@@ -31,4 +31,8 @@ public interface ConfigNodeConfig {
   ConfigNodeConfig setMetricPrometheusReporterPassword(String password);
 
   ConfigNodeConfig setLeaderDistributionPolicy(String policy);
+
+  ConfigNodeConfig setConsistencyCheckSchedulerInitialDelayInMs(long initialDelayInMs);
+
+  ConfigNodeConfig setConsistencyCheckSchedulerIntervalInMs(long intervalInMs);
 }

integration-test/src/test/java/org/apache/iotdb/db/it/iotconsensusv2/batch/IoTDBIoTConsensusV2Batch3C3DBasicIT.java

@@ -44,6 +44,49 @@ public void testReplicaConsistencyAfterLeaderStop() throws Exception {
     super.testReplicaConsistencyAfterLeaderStop();
   }
 
+  @Override
+  @Test
+  public void testFollowerCanReadConsistentDataAfterCatchUp() throws Exception {
+    super.testFollowerCanReadConsistentDataAfterCatchUp();
+  }
+
+  @Override
+  @Test
+  public void testReplicaConsistencyAfterDeleteAndLeaderStop() throws Exception {
+    super.testReplicaConsistencyAfterDeleteAndLeaderStop();
+  }
+
+  @Override
+  @Test
+  public void testReplicaConsistencyRepairAfterFollowerLosesSealedTsFile() throws Exception {
+    super.testReplicaConsistencyRepairAfterFollowerLosesSealedTsFile();
+  }
+
+  @Override
+  @Test
+  public void testBackgroundConsistencyCheckOnlyRunsOnColdPartitions() throws Exception {
+    super.testBackgroundConsistencyCheckOnlyRunsOnColdPartitions();
+  }
+
+  @Override
+  @Test
+  public void testBackgroundConsistencyCheckRebuildsLogicalSnapshotAfterFollowerRestart()
+      throws Exception {
+    super.testBackgroundConsistencyCheckRebuildsLogicalSnapshotAfterFollowerRestart();
+  }
+
+  @Override
+  @Test
+  public void testBackgroundConsistencyCheckWaitsForSyncLagToClear() throws Exception {
+    super.testBackgroundConsistencyCheckWaitsForSyncLagToClear();
+  }
+
+  @Override
+  @Test
+  public void testRepairProgressSurvivesConfigNodeLeaderRestart() throws Exception {
+    super.testRepairProgressSurvivesConfigNodeLeaderRestart();
+  }
+
   @Override
   @Test
   public void test3C3DWriteFlushAndQuery() throws Exception {

integration-test/src/test/java/org/apache/iotdb/db/it/iotconsensusv2/stream/IoTDBIoTConsensusV2Stream3C3DBasicIT.java

@@ -44,6 +44,49 @@ public void testReplicaConsistencyAfterLeaderStop() throws Exception {
     super.testReplicaConsistencyAfterLeaderStop();
   }
 
+  @Override
+  @Test
+  public void testFollowerCanReadConsistentDataAfterCatchUp() throws Exception {
+    super.testFollowerCanReadConsistentDataAfterCatchUp();
+  }
+
+  @Override
+  @Test
+  public void testReplicaConsistencyAfterDeleteAndLeaderStop() throws Exception {
+    super.testReplicaConsistencyAfterDeleteAndLeaderStop();
+  }
+
+  @Override
+  @Test
+  public void testReplicaConsistencyRepairAfterFollowerLosesSealedTsFile() throws Exception {
+    super.testReplicaConsistencyRepairAfterFollowerLosesSealedTsFile();
+  }
+
+  @Override
+  @Test
+  public void testBackgroundConsistencyCheckOnlyRunsOnColdPartitions() throws Exception {
+    super.testBackgroundConsistencyCheckOnlyRunsOnColdPartitions();
+  }
+
+  @Override
+  @Test
+  public void testBackgroundConsistencyCheckRebuildsLogicalSnapshotAfterFollowerRestart()
+      throws Exception {
+    super.testBackgroundConsistencyCheckRebuildsLogicalSnapshotAfterFollowerRestart();
+  }
+
+  @Override
+  @Test
+  public void testBackgroundConsistencyCheckWaitsForSyncLagToClear() throws Exception {
+    super.testBackgroundConsistencyCheckWaitsForSyncLagToClear();
+  }
+
+  @Override
+  @Test
+  public void testRepairProgressSurvivesConfigNodeLeaderRestart() throws Exception {
+    super.testRepairProgressSurvivesConfigNodeLeaderRestart();
+  }
+
   @Override
   @Test
   public void test3C3DWriteFlushAndQuery() throws Exception {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/CnToDnSyncRequestType.java

@@ -36,6 +36,13 @@ public enum CnToDnSyncRequestType {
   REMOVE_REGION_PEER,
   DELETE_OLD_REGION_PEER,
   RESET_PEER_LIST,
+  GET_CONSISTENCY_ELIGIBILITY,
+  GET_SNAPSHOT_SUBTREE,
+  ESTIMATE_LEAF_DIFF,
+  DECODE_LEAF_DIFF,
+  STREAM_LOGICAL_REPAIR,
+  APPLY_LOGICAL_REPAIR_BATCH,
+  FINISH_LOGICAL_REPAIR_SESSION,
 
   // PartitionCache
   INVALIDATE_PARTITION_CACHE,

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java

@@ -28,17 +28,24 @@
 import org.apache.iotdb.commons.client.exception.ClientManagerException;
 import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
 import org.apache.iotdb.commons.exception.UncheckedStartupException;
+import org.apache.iotdb.mpp.rpc.thrift.TApplyLogicalRepairBatchReq;
 import org.apache.iotdb.mpp.rpc.thrift.TCleanDataNodeCacheReq;
 import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq;
 import org.apache.iotdb.mpp.rpc.thrift.TCreatePeerReq;
 import org.apache.iotdb.mpp.rpc.thrift.TCreateSchemaRegionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TDecodeLeafDiffReq;
+import org.apache.iotdb.mpp.rpc.thrift.TEstimateLeafDiffReq;
+import org.apache.iotdb.mpp.rpc.thrift.TFinishLogicalRepairSessionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TGetConsistencyEligibilityReq;
+import org.apache.iotdb.mpp.rpc.thrift.TGetSnapshotSubtreeReq;
 import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq;
 import org.apache.iotdb.mpp.rpc.thrift.TInvalidatePermissionCacheReq;
 import org.apache.iotdb.mpp.rpc.thrift.TKillQueryInstanceReq;
 import org.apache.iotdb.mpp.rpc.thrift.TMaintainPeerReq;
 import org.apache.iotdb.mpp.rpc.thrift.TRegionLeaderChangeReq;
 import org.apache.iotdb.mpp.rpc.thrift.TRegionLeaderChangeResp;
 import org.apache.iotdb.mpp.rpc.thrift.TResetPeerListReq;
+import org.apache.iotdb.mpp.rpc.thrift.TStreamLogicalRepairReq;
 import org.apache.iotdb.mpp.rpc.thrift.TUpdateTableReq;
 import org.apache.iotdb.mpp.rpc.thrift.TUpdateTemplateReq;
 import org.apache.iotdb.rpc.TSStatusCode;
@@ -98,6 +105,27 @@ private void buildActionMap() {
     actionMapBuilder.put(
         CnToDnSyncRequestType.DELETE_REGION,
         (req, client) -> client.deleteRegion((TConsensusGroupId) req));
+    actionMapBuilder.put(
+        CnToDnSyncRequestType.GET_CONSISTENCY_ELIGIBILITY,
+        (req, client) -> client.getConsistencyEligibility((TGetConsistencyEligibilityReq) req));
+    actionMapBuilder.put(
+        CnToDnSyncRequestType.GET_SNAPSHOT_SUBTREE,
+        (req, client) -> client.getSnapshotSubtree((TGetSnapshotSubtreeReq) req));
+    actionMapBuilder.put(
+        CnToDnSyncRequestType.ESTIMATE_LEAF_DIFF,
+        (req, client) -> client.estimateLeafDiff((TEstimateLeafDiffReq) req));
+    actionMapBuilder.put(
+        CnToDnSyncRequestType.DECODE_LEAF_DIFF,
+        (req, client) -> client.decodeLeafDiff((TDecodeLeafDiffReq) req));
+    actionMapBuilder.put(
+        CnToDnSyncRequestType.STREAM_LOGICAL_REPAIR,
+        (req, client) -> client.streamLogicalRepair((TStreamLogicalRepairReq) req));
+    actionMapBuilder.put(
+        CnToDnSyncRequestType.APPLY_LOGICAL_REPAIR_BATCH,
+        (req, client) -> client.applyLogicalRepairBatch((TApplyLogicalRepairBatchReq) req));
+    actionMapBuilder.put(
+        CnToDnSyncRequestType.FINISH_LOGICAL_REPAIR_SESSION,
+        (req, client) -> client.finishLogicalRepairSession((TFinishLogicalRepairSessionReq) req));
     actionMapBuilder.put(
         CnToDnSyncRequestType.INVALIDATE_PERMISSION_CACHE,
         (req, client) -> client.invalidatePermissionCache((TInvalidatePermissionCacheReq) req));

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java

@@ -317,6 +317,9 @@ public class ConfigNodeConfig {
   /** The getOrCreatePartitionTable interface will log new created Partition if set true. */
   private boolean isEnablePrintingNewlyCreatedPartition = false;
 
+  private long consistencyCheckSchedulerInitialDelayInMs = 60_000L;
+  private long consistencyCheckSchedulerIntervalInMs = 900_000L;
+
   private long forceWalPeriodForConfigNodeSimpleInMs = 100;
 
   public ConfigNodeConfig() {
@@ -1174,6 +1177,23 @@ public void setEnablePrintingNewlyCreatedPartition(boolean enablePrintingNewlyCr
     isEnablePrintingNewlyCreatedPartition = enablePrintingNewlyCreatedPartition;
   }
 
+  public long getConsistencyCheckSchedulerInitialDelayInMs() {
+    return consistencyCheckSchedulerInitialDelayInMs;
+  }
+
+  public void setConsistencyCheckSchedulerInitialDelayInMs(
+      long consistencyCheckSchedulerInitialDelayInMs) {
+    this.consistencyCheckSchedulerInitialDelayInMs = consistencyCheckSchedulerInitialDelayInMs;
+  }
+
+  public long getConsistencyCheckSchedulerIntervalInMs() {
+    return consistencyCheckSchedulerIntervalInMs;
+  }
+
+  public void setConsistencyCheckSchedulerIntervalInMs(long consistencyCheckSchedulerIntervalInMs) {
+    this.consistencyCheckSchedulerIntervalInMs = consistencyCheckSchedulerIntervalInMs;
+  }
+
   public long getForceWalPeriodForConfigNodeSimpleInMs() {
     return forceWalPeriodForConfigNodeSimpleInMs;
   }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java

@@ -348,6 +348,18 @@ private void loadProperties(TrimProperties properties) throws BadNodeUrlExceptio
                 "enable_auto_leader_balance_for_iot_consensus",
                 String.valueOf(conf.isEnableAutoLeaderBalanceForIoTConsensus()))));
 
+    conf.setConsistencyCheckSchedulerInitialDelayInMs(
+        Long.parseLong(
+            properties.getProperty(
+                "consistency_check_scheduler_initial_delay_in_ms",
+                String.valueOf(conf.getConsistencyCheckSchedulerInitialDelayInMs()))));
+
+    conf.setConsistencyCheckSchedulerIntervalInMs(
+        Long.parseLong(
+            properties.getProperty(
+                "consistency_check_scheduler_interval_in_ms",
+                String.valueOf(conf.getConsistencyCheckSchedulerIntervalInMs()))));
+
     String routePriorityPolicy =
         properties.getProperty("route_priority_policy", conf.getRoutePriorityPolicy());
     if (IPriorityBalancer.GREEDY_POLICY.equals(routePriorityPolicy)