[RTO/RPO] Topology awareness for query plan by SzyWilliam · Pull Request #15014 · apache/iotdb

SzyWilliam · 2025-03-05T02:27:36Z

Introduction

This Patch implements two functionalities:

ConfigNode Leader now is aware of intra-DataNode network connection topology.
Query / Execution will utilize topology knowledge for better planning.

Part1. Topology awareness

ConfigNode is aware of the intra-datanode topology:

DataNode will receive latest topology in the next heartbeat iff. the topology is changed.

Kill a DataNode manually, ConfigNode failure detector will detect this problem:

DataNode will be notified with this problem:

Part2. How it affects query plan and execution

In each FragmentInstance, the TRegionReplicaSet will be rewrite, only reachable DataNodeLocations will survive.

Empty TRegionReplicaSet in query: fail immediately with ReplicaSetUnreachableException. It will be caught and return the corresponding TSStatus
Empty TRegionReplicaSet in write: won't be scheduled and added to the failed list during schedule phase.

If we cannot select a TRegionReplicaSet as RootFI in query: fail immediately with RootFIPlacementException. It will be caught and return the corresponding TSStatus

Part3. TODOs

Let failure detector print log only during status change, instead of each failure. This todo will be implemented together with next review addressing patch.
Set ReplicaSetUnreachableException & RootFIPlacementException TSStatus to retry in session connection.
Maybe we can immediately trigger topologyProbing when registerDataNode, removeDataNode, restartDataNode.

OneSizeFitsQuorum

Please fix this problem

2025-03-11 14:52:12,134 [AsyncDataNodeHeartbeatServiceClientPool-selector-99] ERROR o.a.t.a.TAsyncClientManager$SelectThread:117 - Ignoring uncaught exception in SelectThread 
java.util.ConcurrentModificationException: null
        at java.base/java.util.LinkedList$ListItr.checkForComodification(LinkedList.java:970)
        at java.base/java.util.LinkedList$ListItr.next(LinkedList.java:892)
        at java.base/java.util.Collections$UnmodifiableCollection$1.next(Collections.java:1054)
        at org.apache.iotdb.confignode.manager.load.cache.detector.PhiAccrualDetector.create(PhiAccrualDetector.java:94)
        at org.apache.iotdb.confignode.manager.load.cache.detector.PhiAccrualDetector.isAvailable(PhiAccrualDetector.java:72)
        at org.apache.iotdb.confignode.manager.load.cache.node.ConfigNodeHeartbeatCache.updateCurrentStatistics(ConfigNodeHeartbeatCache.java:70)
        at org.apache.iotdb.confignode.manager.load.cache.LoadCache.lambda$updateNodeStatistics$21(LoadCache.java:380)
        at java.base/java.util.concurrent.ConcurrentHashMap$ValuesView.forEach(ConcurrentHashMap.java:4780)
        at org.apache.iotdb.confignode.manager.load.cache.LoadCache.updateNodeStatistics(LoadCache.java:380)
        at org.apache.iotdb.confignode.manager.load.LoadManager.forceUpdateNodeCache(LoadManager.java:275)
        at org.apache.iotdb.confignode.client.async.handlers.heartbeat.DataNodeHeartbeatHandler.onError(DataNodeHeartbeatHandler.java:166)
        at org.apache.thrift.async.TAsyncMethodCall.onError(TAsyncMethodCall.java:216)
        at org.apache.thrift.async.TAsyncMethodCall.transition(TAsyncMethodCall.java:210)
        at org.apache.thrift.async.TAsyncClientManager$SelectThread.transitionMethods(TAsyncClientManager.java:143)
        at org.apache.thrift.async.TAsyncClientManager$SelectThread.run(TAsyncClientManager.java:113)

Please optimize log printing issues

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

...fignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java

...nfignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/TopologyService.java

.../org/apache/iotdb/db/queryengine/plan/planner/exceptions/ReplicaSetUnreachableException.java

...n/java/org/apache/iotdb/db/queryengine/plan/planner/exceptions/RootFIPlacementException.java

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ClusterTopology.java

SzyWilliam · 2025-03-14T06:03:18Z

When a new DataNode registers, ConfigNode will trigger topology probing immediately

2. DataNode will learn the latest topology from ConfigNode via heartbeats

3. Manually kill a DataNode, topology change will be detected:

Restart this DataNode again, the topology will soon be updated.

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolFactory.java

...rc/main/java/org/apache/iotdb/confignode/manager/load/cache/detector/PhiAccrualDetector.java

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

...n/java/org/apache/iotdb/db/queryengine/plan/planner/exceptions/RootFIPlacementException.java

.../org/apache/iotdb/db/queryengine/plan/planner/exceptions/ReplicaSetUnreachableException.java

...rc/main/java/org/apache/iotdb/confignode/manager/load/cache/detector/PhiAccrualDetector.java

...nfignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/TopologyService.java

OneSizeFitsQuorum

Awesome work!

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

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

...nfignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/TopologyService.java

OneSizeFitsQuorum · 2025-03-19T04:31:34Z

...nfignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/TopologyService.java

+
+  public synchronized void stopTopologyService() {
+    shouldRun.set(false);
+    topologyThread.shutdown();


Maybe we don't need to shutdown the thread pool, but make sure that the thread completes and waits for 60 seconds before being recycled.

Careful design is required to ensure that this single-threaded thread pool only has a single task when switching leaders

...nfignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/TopologyService.java

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ClusterTopology.java

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolFactory.java

JackieTien97 · 2025-03-19T10:23:50Z

...org/apache/iotdb/db/queryengine/plan/planner/distribution/SimpleFragmentParallelPlanner.java


  // Record FragmentInstances dispatched to same DataNode
  private final Map<TDataNodeLocation, List<FragmentInstance>> dataNodeFIMap;
+  private final ClusterTopology topology = ClusterTopology.getInstance();


better not using singleton, passing it in constructor will be better for UT code

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/partition/StorageExecutor.java

...main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java

...n/java/org/apache/iotdb/db/queryengine/plan/planner/exceptions/RootFIPlacementException.java

.../org/apache/iotdb/db/queryengine/plan/planner/exceptions/ReplicaSetUnreachableException.java

JackieTien97 · 2025-03-19T10:38:24Z

.../iotdb/db/queryengine/plan/relational/planner/distribute/TableModelQueryFragmentPlanner.java


  // Record FragmentInstances dispatched to same DataNode
  private final Map<TDataNodeLocation, List<FragmentInstance>> dataNodeFIMap = new HashMap<>();
+  private final ClusterTopology topology = ClusterTopology.getInstance();


same as above

JackieTien97 · 2025-03-19T10:38:47Z

...e/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java

  private final SymbolAllocator symbolAllocator;
  private final Map<PlanNodeId, OrderingScheme> nodeOrderingMap = new HashMap<>();
  private final DataNodeLocationSupplierFactory.DataNodeLocationSupplier dataNodeLocationSupplier;
+  private final ClusterTopology topology = ClusterTopology.getInstance();


avoid using singleton

JackieTien97 · 2025-03-19T11:54:58Z

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ClusterTopology.java

+    final List<TRegionReplicaSet> allSets =
+        input.stream().map(Map.Entry::getKey).collect(Collectors.toList());
+    final List<TRegionReplicaSet> candidates = getReachableCandidates(allSets);


why we need to firstly convert ti to List

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ClusterTopology.java

OneSizeFitsQuorum

LGTM, only minor issues

OneSizeFitsQuorum · 2025-03-22T06:57:22Z

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

    long retryIntervalMS = 1000;
    while (true) {
      try (Connection connection = EnvFactory.getEnv().getConnection()) {
+        final List<BaseNodeWrapper> allDataNodes =


Remove this in next PR

OneSizeFitsQuorum · 2025-03-22T08:14:11Z

...nfignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/TopologyService.java

+  private void logAsymmetricPartition(final Map<Integer, Set<Integer>> partitioned) {
+    for (final int fromId : partitioned.keySet()) {
+      for (final int toId : partitioned.get(fromId)) {
+        if (partitioned.get(toId) == null || !partitioned.get(toId).contains(fromId)) {


It seems that if one node generates a symmetric network partition, all other nodes log an asymmetric network partition.

ConfigNode-leader should theoretically be able to identify whether symmetric or asymmetric partitions are generated. I think we can use latestTopology to detect whether symmetric or asymmetric network partitions are present. Because the key contains the full number of nodes, and the partitioned only contains the successful parts, it cannot be used as from for the failed nodes

OneSizeFitsQuorum · 2025-03-22T08:44:14Z

...n/java/org/apache/iotdb/db/queryengine/plan/planner/exceptions/RootFIPlacementException.java

+public class RootFIPlacementException extends IoTDBRuntimeException {
+  public RootFIPlacementException(Collection<TRegionReplicaSet> replicaSets) {
+    super(
+        "root FragmentInstance placement error: " + replicaSets.toString(),


OneSizeFitsQuorum · 2025-03-22T08:46:12Z

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ErrorHandlingUtils.java

    } else if (t instanceof QueryInBatchStatementException) {
      return RpcUtils.getStatus(
          TSStatusCode.QUERY_NOT_ALLOWED, INFO_NOT_ALLOWED_IN_BATCH_ERROR + rootCause.getMessage());
+    } else if (t instanceof RootFIPlacementException) {


merge them in one if

temp save for CI

6ea3c15

SzyWilliam force-pushed the topology_awareness_v2 branch from 854382e to 6ea3c15 Compare March 5, 2025 16:50

This was referenced Mar 6, 2025

[RTO/RPO] Topology awareness #14996

Closed

[RTO/RPO] Make ConfigNode aware of DataNode topology #14915

Closed

save topology service

49a0c15

SzyWilliam changed the title ~~[RTO/RPO] topology awareness v2~~ [RTO/RPO] Topology awareness for query plan Mar 10, 2025

SzyWilliam added 4 commits March 10, 2025 14:55

fix ut and it

a9e1384

fix ut and it

e16688d

fix ut and it

2b622b1

fix ut and it

d2f0451

OneSizeFitsQuorum reviewed Mar 12, 2025

View reviewed changes

address review issues

ebfbdf8

SzyWilliam added 8 commits March 14, 2025 14:12

address review issues

4e418fb

fix conflicts

b9d5cc5

fix conflicts

cb3ad7e

spotless

055e696

solve more conflicts and do regression test

84f5d33

spotless

7d065f7

fixing the ut and it

e844034

fixing the ut and it

f3832c2

OneSizeFitsQuorum reviewed Mar 18, 2025

View reviewed changes

address review issues

d91ae73

OneSizeFitsQuorum reviewed Mar 19, 2025

View reviewed changes

JackieTien97 requested changes Mar 19, 2025

View reviewed changes

address review issues

4bb12fc

JackieTien97 approved these changes Mar 21, 2025

View reviewed changes

SzyWilliam added 2 commits March 22, 2025 13:23

address review issues

810753c

address review issues

044e9c7

OneSizeFitsQuorum approved these changes Mar 22, 2025

View reviewed changes

OneSizeFitsQuorum merged commit af03c16 into apache:master Mar 22, 2025
57 of 59 checks passed

SzyWilliam mentioned this pull request Mar 23, 2025

[RTO/RPO] Chore: Fix remaining issues in PR 15014 #15171

Merged

Comments

Conversation

SzyWilliam commented Mar 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Introduction

Part1. Topology awareness

Part2. How it affects query plan and execution

Part3. TODOs

Uh oh!

OneSizeFitsQuorum left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

SzyWilliam commented Mar 14, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

OneSizeFitsQuorum left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

OneSizeFitsQuorum left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

SzyWilliam commented Mar 5, 2025 •

edited

Loading