This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new b3b2e95f52e Fix LOAD replica set order comparison (#18551)
b3b2e95f52e is described below
commit b3b2e95f52e60f2d64027b7e91f3d434ed2c928d
Author: Zhenyu Luo <[email protected]>
AuthorDate: Mon Aug 31 17:55:08 2026 +0800
Fix LOAD replica set order comparison (#18551)
---
.../plan/scheduler/load/LoadTsFileScheduler.java | 27 +++++++++++++-
.../scheduler/load/LoadTsFileSchedulerTest.java | 43 ++++++++++++++++++++++
2 files changed, 69 insertions(+), 1 deletion(-)
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
index da143c6b8cb..a116f3ac1d8 100644
---
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
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.queryengine.plan.scheduler.load;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
@@ -355,6 +356,29 @@ public class LoadTsFileScheduler implements IScheduler {
return true;
}
+ static boolean isSameRegionReplicaSet(TRegionReplicaSet original,
TRegionReplicaSet current) {
+ if (!Objects.equals(original.getRegionId(), current.getRegionId())) {
+ return false;
+ }
+
+ final Map<TDataNodeLocation, Integer> locationCounts = new HashMap<>();
+ for (TDataNodeLocation location : original.getDataNodeLocations()) {
+ locationCounts.merge(location, 1, Integer::sum);
+ }
+ for (TDataNodeLocation location : current.getDataNodeLocations()) {
+ final Integer count = locationCounts.get(location);
+ if (count == null) {
+ return false;
+ }
+ if (count == 1) {
+ locationCounts.remove(location);
+ } else {
+ locationCounts.put(location, count - 1);
+ }
+ }
+ return locationCounts.isEmpty();
+ }
+
private boolean dispatchOnePieceNode(
LoadTsFilePieceNode pieceNode, TRegionReplicaSet replicaSet) {
allReplicaSets.add(replicaSet);
@@ -851,7 +875,8 @@ public class LoadTsFileScheduler implements IScheduler {
final TRegionReplicaSet replicaSet =
replicaSets.get(chunkPartitionIndexes[i]);
final TConsensusGroupId regionId = replicaSet.getRegionId();
if (regionId2ReplicaSetAndNode.containsKey(regionId)
- &&
!Objects.equals(regionId2ReplicaSetAndNode.get(regionId).getLeft(),
replicaSet)) {
+ && !isSameRegionReplicaSet(
+ regionId2ReplicaSetAndNode.get(regionId).getLeft(),
replicaSet)) {
// Detected region replica set changed (maybe due to region
migration), throw an exception
throw new RegionReplicaSetChangedException(
regionId2ReplicaSetAndNode.get(regionId).getLeft(), replicaSet);
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java
index 1b738f6496d..4605e6cf814 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java
@@ -19,6 +19,11 @@
package org.apache.iotdb.db.queryengine.plan.scheduler.load;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.PlanFragmentId;
@@ -41,6 +46,7 @@ import java.io.File;
import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
+import java.util.Arrays;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
@@ -160,4 +166,41 @@ public class LoadTsFileSchedulerTest {
Assert.assertEquals(0L, getMemoryUsageMethod.invoke(memoryBlock));
Assert.assertEquals(0L, dataSizeField.getLong(dataManager));
}
+
+ @Test
+ public void testRegionReplicaSetComparison() {
+ final TDataNodeLocation dataNode1 = createDataNodeLocation(1, 10731);
+ final TDataNodeLocation dataNode3 = createDataNodeLocation(3, 10733);
+ final TDataNodeLocation dataNode5 = createDataNodeLocation(5, 10735);
+ final TConsensusGroupId regionId = new
TConsensusGroupId(TConsensusGroupType.DataRegion, 1);
+ final TRegionReplicaSet original =
+ new TRegionReplicaSet(regionId, Arrays.asList(dataNode5, dataNode3,
dataNode1));
+
+ Assert.assertTrue(
+ LoadTsFileScheduler.isSameRegionReplicaSet(
+ original,
+ new TRegionReplicaSet(regionId, Arrays.asList(dataNode3,
dataNode5, dataNode1))));
+ Assert.assertFalse(
+ LoadTsFileScheduler.isSameRegionReplicaSet(
+ original,
+ new TRegionReplicaSet(
+ regionId, Arrays.asList(dataNode3, dataNode5,
createDataNodeLocation(7, 10737)))));
+ Assert.assertFalse(
+ LoadTsFileScheduler.isSameRegionReplicaSet(
+ original,
+ new TRegionReplicaSet(
+ regionId, Arrays.asList(dataNode3, dataNode5,
createDataNodeLocation(1, 11731)))));
+ Assert.assertFalse(
+ LoadTsFileScheduler.isSameRegionReplicaSet(
+ original,
+ new TRegionReplicaSet(
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 2),
+ Arrays.asList(dataNode3, dataNode5, dataNode1))));
+ }
+
+ private static TDataNodeLocation createDataNodeLocation(int dataNodeId, int
internalPort) {
+ return new TDataNodeLocation()
+ .setDataNodeId(dataNodeId)
+ .setInternalEndPoint(new TEndPoint("127.0.0.1", internalPort));
+ }
}
