This is an automated email from the ASF dual-hosted git repository.
jackietien 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 b1b0c0192ab Root FragmentInstance placement error may occur when
querying table model aggregation with null data partition
b1b0c0192ab is described below
commit b1b0c0192abfdefead506712896f4bc78cb7418e
Author: shuwenwei <[email protected]>
AuthorDate: Thu Jul 17 09:57:57 2025 +0800
Root FragmentInstance placement error may occur when querying table model
aggregation with null data partition
---
...ableAggregationQueryWithNetworkPartitionIT.java | 186 +++++++++++++++++++++
.../iotdb/db/queryengine/plan/ClusterTopology.java | 16 +-
.../distribute/TableDistributedPlanGenerator.java | 4 +-
3 files changed, 201 insertions(+), 5 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationQueryWithNetworkPartitionIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationQueryWithNetworkPartitionIT.java
new file mode 100644
index 00000000000..8678d8d6630
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationQueryWithNetworkPartitionIT.java
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.relational.it.query.recent;
+
+import org.apache.iotdb.commons.cluster.NodeStatus;
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.isession.ITableSession;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.TableClusterIT;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.iotdb.db.it.utils.TestUtils.prepareTableData;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({TableClusterIT.class})
+public class IoTDBTableAggregationQueryWithNetworkPartitionIT {
+
+ private static final String testConsensusProtocolClass =
ConsensusFactory.RATIS_CONSENSUS;
+ private static final String IoTConsensusProtocolClass =
ConsensusFactory.IOT_CONSENSUS;
+ private static final int testReplicationFactor = 3;
+ private static final long testTimePartitionInterval = 604800000;
+ private static final int testDataRegionGroupPerDatabase = 4;
+ protected static final String DATABASE_NAME = "test";
+ protected static final String[] createSqls =
+ new String[] {
+ "CREATE DATABASE " + DATABASE_NAME,
+ "USE " + DATABASE_NAME,
+ "CREATE TABLE table1(device STRING TAG, s1 INT32 FIELD)",
+ "INSERT INTO table1 (time, device, s1) VALUES (1, 'd1', '1')",
+ "INSERT INTO table1 (time, device, s1) VALUES (-1, 'd2', '1')",
+ };
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv()
+ .getConfig()
+ .getCommonConfig()
+ .setConfigNodeConsensusProtocolClass(testConsensusProtocolClass)
+ .setSchemaRegionConsensusProtocolClass(testConsensusProtocolClass)
+ .setDataRegionConsensusProtocolClass(IoTConsensusProtocolClass)
+ .setSchemaReplicationFactor(testReplicationFactor)
+ .setDataReplicationFactor(testReplicationFactor)
+ .setTimePartitionInterval(testTimePartitionInterval)
+
.setDefaultDataRegionGroupNumPerDatabase(testDataRegionGroupPerDatabase);
+ EnvFactory.getEnv().initClusterEnvironment(1, 3);
+ prepareTableData(createSqls);
+ }
+
+ @AfterClass
+ public static void tearDownAfterClass() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ ensureAllDataNodeRunning();
+ }
+
+ @Test
+ public void test1() throws IoTDBConnectionException,
StatementExecutionException, SQLException {
+ try (ITableSession session =
+ EnvFactory.getEnv().getTableSessionConnectionWithDB(DATABASE_NAME)) {
+ SessionDataSet sessionDataSet =
+ session.executeQueryStatement(
+ "select count(s1) from table1 where device = 'd1' and time < -1
group by device");
+ Assert.assertFalse(sessionDataSet.hasNext());
+ }
+ EnvFactory.getEnv().shutdownAllDataNodes();
+ List<DataNodeWrapper> dataNodeWrapperList =
EnvFactory.getEnv().getDataNodeWrapperList();
+ for (DataNodeWrapper dataNodeWrapper : dataNodeWrapperList) {
+ EnvFactory.getEnv()
+ .ensureNodeStatus(
+ Collections.singletonList(dataNodeWrapper),
+ Collections.singletonList(NodeStatus.Unknown));
+ }
+
+ List<String> otherNodes = new ArrayList<>();
+ for (int i = 1; i < dataNodeWrapperList.size(); i++) {
+ EnvFactory.getEnv().startDataNode(i);
+ EnvFactory.getEnv()
+ .ensureNodeStatus(
+ Collections.singletonList(dataNodeWrapperList.get(i)),
+ Collections.singletonList(NodeStatus.Running));
+ DataNodeWrapper dataNodeWrapper = dataNodeWrapperList.get(i);
+ otherNodes.add(dataNodeWrapper.getIpAndPortString());
+ }
+
+ try (ITableSession session =
EnvFactory.getEnv().getTableSessionConnection(otherNodes)) {
+ session.executeNonQueryStatement("use " + DATABASE_NAME);
+ SessionDataSet sessionDataSet =
+ session.executeQueryStatement(
+ "select count(s1) from table1 where device = 'd1' and time < -1
group by device");
+ Assert.assertFalse(sessionDataSet.hasNext());
+ }
+ }
+
+ @Test
+ public void test2() throws IoTDBConnectionException,
StatementExecutionException, SQLException {
+ try (ITableSession session =
+ EnvFactory.getEnv().getTableSessionConnectionWithDB(DATABASE_NAME)) {
+ SessionDataSet sessionDataSet =
+ session.executeQueryStatement(
+ "select device, count(s1) from table1 where (device = 'd1' or
device = 'd2') and time < -1 group by device");
+ Assert.assertFalse(sessionDataSet.hasNext());
+ }
+ EnvFactory.getEnv().shutdownAllDataNodes();
+ List<DataNodeWrapper> dataNodeWrapperList =
EnvFactory.getEnv().getDataNodeWrapperList();
+ for (DataNodeWrapper dataNodeWrapper : dataNodeWrapperList) {
+ EnvFactory.getEnv()
+ .ensureNodeStatus(
+ Collections.singletonList(dataNodeWrapper),
+ Collections.singletonList(NodeStatus.Unknown));
+ }
+
+ List<String> otherNodes = new ArrayList<>();
+ for (int i = 1; i < dataNodeWrapperList.size(); i++) {
+ EnvFactory.getEnv().startDataNode(i);
+ EnvFactory.getEnv()
+ .ensureNodeStatus(
+ Collections.singletonList(dataNodeWrapperList.get(i)),
+ Collections.singletonList(NodeStatus.Running));
+ DataNodeWrapper dataNodeWrapper = dataNodeWrapperList.get(i);
+ otherNodes.add(dataNodeWrapper.getIpAndPortString());
+ }
+
+ try (ITableSession session =
EnvFactory.getEnv().getTableSessionConnection(otherNodes)) {
+ session.executeNonQueryStatement("use " + DATABASE_NAME);
+ SessionDataSet sessionDataSet =
+ session.executeQueryStatement(
+ "select device, count(s1) from table1 where (device = 'd1' or
device = 'd2') and time <= -1 group by device");
+ int count = 0;
+ while (sessionDataSet.hasNext()) {
+ sessionDataSet.next();
+ count++;
+ }
+ Assert.assertEquals(1, count);
+ }
+ }
+
+ private void ensureAllDataNodeRunning() {
+ for (DataNodeWrapper dataNodeWrapper :
EnvFactory.getEnv().getDataNodeWrapperList()) {
+ if (dataNodeWrapper.isAlive()) {
+ continue;
+ }
+ dataNodeWrapper.start();
+ EnvFactory.getEnv()
+ .ensureNodeStatus(
+ Collections.singletonList(dataNodeWrapper),
+ Collections.singletonList(NodeStatus.Running));
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ClusterTopology.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ClusterTopology.java
index 57128834963..7a277ce0165 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ClusterTopology.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/ClusterTopology.java
@@ -22,7 +22,9 @@ package org.apache.iotdb.db.queryengine.plan;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.commons.partition.DataPartition;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import
org.apache.iotdb.db.queryengine.plan.planner.exceptions.ReplicaSetUnreachableException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -71,12 +73,18 @@ public class ClusterTopology {
}
final List<TRegionReplicaSet> allSets =
input.stream().map(Map.Entry::getKey).collect(Collectors.toList());
- final List<TRegionReplicaSet> candidates = getReachableCandidates(allSets);
+ final List<TRegionReplicaSet> candidates =
+ getReachableCandidates(
+
allSets.stream().filter(TRegionReplicaSet::isSetRegionId).collect(Collectors.toList()));
final Map<TConsensusGroupId, TRegionReplicaSet> newMap = new HashMap<>();
candidates.forEach(set -> newMap.put(set.getRegionId(), set));
final Map<TRegionReplicaSet, T> candidateMap = new HashMap<>();
for (final Map.Entry<TRegionReplicaSet, T> entry : input) {
final TConsensusGroupId gid = entry.getKey().getRegionId();
+ if (gid == null) {
+ candidateMap.put(DataPartition.NOT_ASSIGNED, entry.getValue());
+ continue;
+ }
final TRegionReplicaSet replicaSet = newMap.get(gid);
if (replicaSet != null) {
candidateMap.put(replicaSet, entry.getValue());
@@ -89,9 +97,11 @@ public class ClusterTopology {
if (!isPartitioned.get() || all == null || all.isEmpty()) {
return all;
}
- if (all.stream().anyMatch(set -> set.getDataNodeLocationsSize() == 0)) {
+ for (TRegionReplicaSet replicaSet : all) {
// some TRegionReplicaSet is unreachable since all DataNodes are down
- return Collections.emptyList();
+ if (replicaSet.getDataNodeLocationsSize() == 0) {
+ throw new ReplicaSetUnreachableException(replicaSet);
+ }
}
final Map<Integer, Set<Integer>> topologyMapCurrent =
Collections.unmodifiableMap(this.topologyMap.get());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
index 08bf122de5c..f7b0c432e34 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
@@ -1038,11 +1038,11 @@ public class TableDistributedPlanGenerator
node instanceof AggregationTreeDeviceViewScanNode
? ((AggregationTreeDeviceViewScanNode) node).getTreeDBName()
: node.getQualifiedObjectName().getDatabaseName();
- if (dbName == null) {
+ DataPartition dataPartition = analysis.getDataPartitionInfo();
+ if (dbName == null || dataPartition == null) {
node.setRegionReplicaSet(NOT_ASSIGNED);
return Collections.singletonList(node);
}
- DataPartition dataPartition = analysis.getDataPartitionInfo();
boolean needSplit = false;
List<List<TRegionReplicaSet>> regionReplicaSetsList = new ArrayList<>();
if (dataPartition != null) {
