[RTO/RPO] Topology awareness for query plan

integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java

@@ -20,12 +20,14 @@
 package org.apache.iotdb.db.it.utils;
 
 import org.apache.iotdb.commons.auth.entity.PrivilegeType;
+import org.apache.iotdb.commons.cluster.NodeStatus;
 import org.apache.iotdb.isession.SessionConfig;
 import org.apache.iotdb.isession.SessionDataSet;
 import org.apache.iotdb.it.env.EnvFactory;
 import org.apache.iotdb.it.env.cluster.env.AbstractEnv;
 import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
 import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.iotdb.itbase.env.BaseNodeWrapper;
 import org.apache.iotdb.rpc.IoTDBConnectionException;
 import org.apache.iotdb.rpc.RpcUtils;
 import org.apache.iotdb.rpc.StatementExecutionException;
@@ -43,6 +45,7 @@
 import java.text.DateFormat;
 import java.time.ZoneId;
 import java.time.ZoneOffset;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -1706,6 +1709,11 @@ public static void restartDataNodes() {
  long retryIntervalMS = 1000;
  while (true) {
  try (Connection connection = EnvFactory.getEnv().getConnection()) {
+ final List<BaseNodeWrapper> allDataNodes =
+ new ArrayList<>(EnvFactory.getEnv().getDataNodeWrapperList());
+ EnvFactory.getEnv()
+ .ensureNodeStatus(
+ allDataNodes, Collections.nCopies(allDataNodes.size(), NodeStatus.Running));
  break;
  } catch (Exception e) {
  try {

iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java

@@ -136,6 +136,7 @@ public enum TSStatusCode {
 
  QUERY_EXECUTION_MEMORY_NOT_ENOUGH(719),
  QUERY_TIMEOUT(720),
+ PLAN_FAILED_NETWORK_PARTITION(721),
 
  // Arithmetic
  NUMERIC_VALUE_OUT_OF_RANGE(750),

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java

@@ -32,6 +32,7 @@ public enum CnToDnAsyncRequestType {
  SET_SYSTEM_STATUS,
  SET_CONFIGURATION,
  SUBMIT_TEST_CONNECTION_TASK,
+ SUBMIT_TEST_DN_INTERNAL_CONNECTION_TASK,
  TEST_CONNECTION,
 
  // Region Maintenance

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java

@@ -374,6 +374,11 @@ protected void initActionMapBuilder() {
  (req, client, handler) ->
  client.submitTestConnectionTask(
  (TNodeLocations) req, (SubmitTestConnectionTaskRPCHandler) handler));
+ actionMapBuilder.put(
+ CnToDnAsyncRequestType.SUBMIT_TEST_DN_INTERNAL_CONNECTION_TASK,
+ (req, client, handler) ->
+ client.submitInternalTestConnectionTask(
+ (TNodeLocations) req, (SubmitTestConnectionTaskRPCHandler) handler));
  actionMapBuilder.put(
  CnToDnAsyncRequestType.TEST_CONNECTION,
  (req, client, handler) ->

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeAsyncRequestRPCHandler.java

@@ -174,6 +174,7 @@ public static DataNodeAsyncRequestRPCHandler<?> buildHandler(
  (Map<Integer, TRegionLeaderChangeResp>) responseMap,
  countDownLatch);
  case SUBMIT_TEST_CONNECTION_TASK:
+ case SUBMIT_TEST_DN_INTERNAL_CONNECTION_TASK:
  return new SubmitTestConnectionTaskRPCHandler(
  requestType,
  requestId,

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java

@@ -249,7 +249,8 @@ private List<TTestConnectionResult> badNodeConnectionResult(
  location -> {
  TEndPoint endPoint = location.getInternalEndPoint();
  TServiceProvider serviceProvider =
- new TServiceProvider(endPoint, TServiceType.ConfigNodeInternalService);
+ new TServiceProvider(
+ endPoint, TServiceType.ConfigNodeInternalService, location.getConfigNodeId());
  TTestConnectionResult result =
  new TTestConnectionResult().setServiceProvider(serviceProvider).setSender(sender);
  result.setSuccess(false).setReason(errorMessage);
@@ -261,7 +262,8 @@ private List<TTestConnectionResult> badNodeConnectionResult(
  location -> {
  TEndPoint endPoint = location.getInternalEndPoint();
  TServiceProvider serviceProvider =
- new TServiceProvider(endPoint, TServiceType.DataNodeInternalService);
+ new TServiceProvider(
+ endPoint, TServiceType.DataNodeInternalService, location.getDataNodeId());
  TTestConnectionResult result =
  new TTestConnectionResult().setServiceProvider(serviceProvider).setSender(sender);
  result.setSuccess(false).setReason(errorMessage);
@@ -274,7 +276,8 @@ private List<TTestConnectionResult> badNodeConnectionResult(
  location -> {
  TEndPoint endPoint = location.getMPPDataExchangeEndPoint();
  TServiceProvider serviceProvider =
- new TServiceProvider(endPoint, TServiceType.DataNodeMPPService);
+ new TServiceProvider(
+ endPoint, TServiceType.DataNodeMPPService, location.getDataNodeId());
  TTestConnectionResult result =
  new TTestConnectionResult()
  .setServiceProvider(serviceProvider)
@@ -288,7 +291,8 @@ private List<TTestConnectionResult> badNodeConnectionResult(
  location -> {
  TEndPoint endPoint = location.getClientRpcEndPoint();
  TServiceProvider serviceProvider =
- new TServiceProvider(endPoint, TServiceType.DataNodeExternalService);
+ new TServiceProvider(
+ endPoint, TServiceType.DataNodeExternalService, location.getDataNodeId());
  TTestConnectionResult result =
  new TTestConnectionResult()
  .setServiceProvider(serviceProvider)

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java

@@ -44,6 +44,7 @@
 import org.apache.iotdb.confignode.manager.load.service.EventService;
 import org.apache.iotdb.confignode.manager.load.service.HeartbeatService;
 import org.apache.iotdb.confignode.manager.load.service.StatisticsService;
+import org.apache.iotdb.confignode.manager.load.service.TopologyService;
 import org.apache.iotdb.confignode.manager.partition.RegionGroupStatus;
 import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
 
@@ -72,6 +73,7 @@ public class LoadManager {
  protected HeartbeatService heartbeatService;
  private final StatisticsService statisticsService;
  private final EventService eventService;
+ private final TopologyService topologyService;
 
  public LoadManager(IManager configManager) {
  this.configManager = configManager;
@@ -83,7 +85,11 @@ public LoadManager(IManager configManager) {
  this.loadCache = new LoadCache();
  setHeartbeatService(configManager, loadCache);
  this.statisticsService = new StatisticsService(loadCache);
- this.eventService = new EventService(configManager, loadCache, routeBalancer);
+ this.topologyService = new TopologyService(configManager, loadCache::updateTopology);
+ this.eventService = new EventService(loadCache);
+ this.eventService.register(configManager.getPipeManager().getPipeRuntimeCoordinator());
+ this.eventService.register(routeBalancer);
+ this.eventService.register(topologyService);
  }
 
  protected void setHeartbeatService(IManager configManager, LoadCache loadCache) {
@@ -146,6 +152,7 @@ public void startLoadServices() {
  statisticsService.startLoadStatisticsService();
  eventService.startEventService();
  partitionBalancer.setupPartitionBalancer();
+ topologyService.startTopologyService();
  }
 
  public void stopLoadServices() {
@@ -155,6 +162,7 @@ public void stopLoadServices() {
  loadCache.clearHeartbeatCache();
  partitionBalancer.clearPartitionBalancer();
  routeBalancer.clearRegionPriority();
+ topologyService.stopTopologyService();
  }
 
  public void clearDataPartitionPolicyTable(String database) {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/AbstractLoadCache.java

@@ -57,7 +57,7 @@ protected AbstractLoadCache() {
  CONF.getFailureDetectorPhiThreshold(),
  CONF.getFailureDetectorPhiAcceptablePauseInMs() * 1000_000L,
  CONF.getHeartbeatIntervalInMs() * 200_000L,
-60,
+IFailureDetector.PHI_COLD_START_THRESHOLD,
  new FixedDetector(CONF.getFailureDetectorFixedThresholdInMs() * 1000_000L));
  break;
  case IFailureDetector.FIXED_DETECTOR:

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/IFailureDetector.java

@@ -29,11 +29,14 @@ public interface IFailureDetector {
  String FIXED_DETECTOR = "fixed";
  String PHI_ACCRUAL_DETECTOR = "phi_accrual";
 
+ int PHI_COLD_START_THRESHOLD = 60;
+
  /**
  * Given the heartbeat history, decide whether this endpoint is still available
  *
+ * @param id the unique identifier of the history owner
  * @param history heartbeat history
  * @return false if the endpoint is under failure
  */
- boolean isAvailable(List<AbstractHeartbeatSample> history);
+ boolean isAvailable(Object id, List<AbstractHeartbeatSample> history);
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java

@@ -49,12 +49,15 @@
 import org.apache.iotdb.confignode.manager.load.cache.region.RegionStatistics;
 import org.apache.iotdb.confignode.manager.partition.RegionGroupStatus;
 
+import org.apache.thrift.annotation.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -94,6 +97,8 @@ public class LoadCache {
  private final Map<TConsensusGroupId, ConsensusGroupCache> consensusGroupCacheMap;
  // Map<DataNodeId, confirmedConfigNodes>
  private final Map<Integer, Set<TEndPoint>> confirmedConfigNodeMap;
+ private Map<Integer, Set<Integer>> topologyGraph;
+ private final AtomicBoolean topologyUpdated;
 
  public LoadCache() {
  this.nodeCacheMap = new ConcurrentHashMap<>();
@@ -102,6 +107,8 @@ public LoadCache() {
  this.regionSizeMap = new ConcurrentHashMap<>();
  this.consensusGroupCacheMap = new ConcurrentHashMap<>();
  this.confirmedConfigNodeMap = new ConcurrentHashMap<>();
+ this.topologyGraph = new HashMap<>();
+ this.topologyUpdated = new AtomicBoolean(false);
  }
 
  public void initHeartbeatCache(final IManager configManager) {
@@ -175,6 +182,7 @@ private void initRegionGroupHeartbeatCache(
  regionGroupId,
  new RegionGroupCache(
  database,
+ regionGroupId,
  regionReplicaSet.getDataNodeLocations().stream()
  .map(TDataNodeLocation::getDataNodeId)
  .collect(Collectors.toSet()),
@@ -287,7 +295,8 @@ public void createRegionGroupHeartbeatCache(
  String database, TConsensusGroupId regionGroupId, Set<Integer> dataNodeIds) {
  boolean isStrongConsistency = CONF.isConsensusGroupStrongConsistency(regionGroupId);
  regionGroupCacheMap.put(
- regionGroupId, new RegionGroupCache(database, dataNodeIds, isStrongConsistency));
+ regionGroupId,
+ new RegionGroupCache(database, regionGroupId, dataNodeIds, isStrongConsistency));
  consensusGroupCacheMap.put(regionGroupId, new ConsensusGroupCache());
  }
 
@@ -299,7 +308,7 @@ public void createRegionGroupHeartbeatCache(
  */
  public void createRegionCache(TConsensusGroupId regionGroupId, int dataNodeId) {
  Optional.ofNullable(regionGroupCacheMap.get(regionGroupId))
- .ifPresent(cache -> cache.createRegionCache(dataNodeId));
+ .ifPresent(cache -> cache.createRegionCache(dataNodeId, regionGroupId));
  }
 
  /**
@@ -769,6 +778,22 @@ public void waitForLeaderElection(List<TConsensusGroupId> regionGroupIds) {
  regionGroupIds);
  }
 
+ public void updateTopology(Map<Integer, Set<Integer>> latestTopology) {
+ if (!latestTopology.equals(topologyGraph)) {
+ LOGGER.info("[Topology Service] Cluster topology changed, latest: {}", latestTopology);
+ }
+ topologyGraph = latestTopology;
+ topologyUpdated.set(true);
+ }
+
+ @Nullable
+ public Map<Integer, Set<Integer>> getTopology() {
+ if (topologyUpdated.compareAndSet(true, false)) {
+ return Collections.unmodifiableMap(topologyGraph);
+ }
+ return null;
+ }
+
  public void updateConfirmedConfigNodeEndPoints(
  int dataNodeId, Set<TEndPoint> configNodeEndPoints) {
  confirmedConfigNodeMap.put(dataNodeId, configNodeEndPoints);

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/detector/FixedDetector.java

@@ -41,7 +41,7 @@ public FixedDetector(long heartbeatTimeoutNs) {
  }
 
  @Override
- public boolean isAvailable(List<AbstractHeartbeatSample> history) {
+ public boolean isAvailable(Object id, List<AbstractHeartbeatSample> history) {
  final AbstractHeartbeatSample lastSample =
  history.isEmpty() ? null : history.get(history.size() - 1);
  if (lastSample != null) {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/detector/PhiAccrualDetector.java

@@ -24,13 +24,16 @@
 import org.apache.iotdb.confignode.manager.load.cache.node.NodeHeartbeatSample;
 import org.apache.iotdb.confignode.manager.load.cache.region.RegionHeartbeatSample;
 
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
 import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics;
 import org.apache.tsfile.utils.Preconditions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 /**
  * The Phi Failure Detector, proposed by Hayashibara, Naohiro, et al. "The/spl phi/accrual failure
@@ -49,6 +52,8 @@ public class PhiAccrualDetector implements IFailureDetector {
  private final long minHeartbeatStdNs;
  private final int codeStartSampleCount;
  private final IFailureDetector fallbackDuringColdStart;
+ /* We are using cache here to avoid managing entry life cycles manually */
+ private final Cache<Object, Boolean> availibilityCache;
 
  public PhiAccrualDetector(
  long threshold,
@@ -61,17 +66,23 @@ public PhiAccrualDetector(
  this.minHeartbeatStdNs = minHeartbeatStdNs;
  this.codeStartSampleCount = minimalSampleCount;
  this.fallbackDuringColdStart = fallbackDuringColdStart;
+ this.availibilityCache =
+ CacheBuilder.newBuilder().expireAfterAccess(5, TimeUnit.MINUTES).build();
  }
 
  @Override
- public boolean isAvailable(List<AbstractHeartbeatSample> history) {
+ public boolean isAvailable(Object id, List<AbstractHeartbeatSample> history) {
  if (history.size() < codeStartSampleCount) {
  /* We haven't received enough heartbeat replies.*/
- return fallbackDuringColdStart.isAvailable(history);
+ return fallbackDuringColdStart.isAvailable(id, history);
  }
  final PhiAccrual phiAccrual = create(history);
  final boolean isAvailable = phiAccrual.phi() < (double) this.threshold;
- if (!isAvailable && LOGGER.isDebugEnabled()) {
+
+ final Boolean previousAvailability = availibilityCache.getIfPresent(id);
+ availibilityCache.put(id, isAvailable);
+
+ if (Boolean.TRUE.equals(previousAvailability) && !isAvailable) {
  // log the status change and dump the heartbeat history for analysis use
  final StringBuilder builder = new StringBuilder();
  builder.append("[");
@@ -81,7 +92,7 @@ public boolean isAvailable(List<AbstractHeartbeatSample> history) {
  }
  builder.append(phiAccrual.timeElapsedSinceLastHeartbeat / 1000_000);
  builder.append("]");
- LOGGER.debug(String.format("Node Down, heartbeat history (ms): %s", builder));
+ LOGGER.info(String.format("Node Down, heartbeat history (ms): %s", builder));
  }
 
  return isAvailable;

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/AINodeHeartbeatCache.java

@@ -54,7 +54,7 @@ public void updateCurrentStatistics(boolean forceUpdate) {
 
  if (lastSample != null && NodeStatus.Removing.equals(lastSample.getStatus())) {
  status = NodeStatus.Removing;
- } else if (!failureDetector.isAvailable(heartbeatHistory)) {
+ } else if (!failureDetector.isAvailable(nodeId, heartbeatHistory)) {
  /* Failure detector decides that this AINode is UNKNOWN */
  status = NodeStatus.Unknown;
  } else if (lastSample != null) {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/ConfigNodeHeartbeatCache.java

@@ -66,7 +66,7 @@ public synchronized void updateCurrentStatistics(boolean forceUpdate) {
  if (lastSample == null) {
  /* First heartbeat not received from this ConfigNode, status is UNKNOWN */
  status = NodeStatus.Unknown;
- } else if (!failureDetector.isAvailable(heartbeatHistory)) {
+ } else if (!failureDetector.isAvailable(nodeId, heartbeatHistory)) {
  /* Failure detector decides that this ConfigNode is UNKNOWN */
  status = NodeStatus.Unknown;
  } else {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/DataNodeHeartbeatCache.java

@@ -68,7 +68,7 @@ public synchronized void updateCurrentStatistics(boolean forceUpdate) {
  if (lastSample == null) {
  /* First heartbeat not received from this DataNode, status is UNKNOWN */
  status = NodeStatus.Unknown;
- } else if (!failureDetector.isAvailable(heartbeatHistory)) {
+ } else if (!failureDetector.isAvailable(nodeId, heartbeatHistory)) {
  /* Failure detector decides that this DataNode is UNKNOWN */
  status = NodeStatus.Unknown;
  } else {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/region/RegionCache.java

@@ -19,10 +19,13 @@
 
 package org.apache.iotdb.confignode.manager.load.cache.region;
 
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
 import org.apache.iotdb.commons.cluster.RegionStatus;
 import org.apache.iotdb.confignode.manager.load.cache.AbstractHeartbeatSample;
 import org.apache.iotdb.confignode.manager.load.cache.AbstractLoadCache;
 
+import org.apache.tsfile.utils.Pair;
+
 import java.util.Collections;
 import java.util.List;
 
@@ -31,9 +34,11 @@
  * statistics of the Region based on the latest RegionHeartbeatSample.
  */
 public class RegionCache extends AbstractLoadCache {
+ private final Pair<Integer, TConsensusGroupId> id;
 
- public RegionCache() {
+ public RegionCache(int dataNodeId, TConsensusGroupId gid) {
  super();
+ this.id = new Pair<>(dataNodeId, gid);
  this.currentStatistics.set(RegionStatistics.generateDefaultRegionStatistics());
  }
 
@@ -50,7 +55,7 @@ public synchronized void updateCurrentStatistics(boolean forceUpdate) {
  if (lastSample == null) {
  /* First heartbeat not received from this region, status is UNKNOWN */
  status = RegionStatus.Unknown;
- } else if (!failureDetector.isAvailable(history)) {
+ } else if (!failureDetector.isAvailable(id, history)) {
  /* Failure detector decides that this region is UNKNOWN */
  status = RegionStatus.Unknown;
  } else {