You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ju...@apache.org on 2026/06/30 09:36:21 UTC
(iotdb) 01/01: Fix deletion visibility in region migration IT (#18065)
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch cp-b35a575-dev-1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 1cdc025268c4fdbbaa2b87ab966e56d0f02098f0
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Tue Jun 30 16:41:51 2026 +0800
Fix deletion visibility in region migration IT (#18065)
---
...TDBRegionMigrateWithDeletionMultiDataDirIT.java | 178 +++++++++++++++++++++
.../daily/iotv1/RegionMigrateFileAssertions.java | 169 +++++++++++++++++++
2 files changed, 347 insertions(+)
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv1/IoTDBRegionMigrateWithDeletionMultiDataDirIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv1/IoTDBRegionMigrateWithDeletionMultiDataDirIT.java
new file mode 100644
index 00000000000..06cb350a91e
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv1/IoTDBRegionMigrateWithDeletionMultiDataDirIT.java
@@ -0,0 +1,178 @@
+/*
+ * 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.confignode.it.regionmigration.pass.daily.iotv1;
+
+import org.apache.iotdb.consensus.ConsensusFactory;
+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.ClusterIT;
+
+import org.apache.tsfile.utils.Pair;
+import org.awaitility.Awaitility;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework.getAllDataNodes;
+import static org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework.getDataRegionMapWithLeader;
+import static org.apache.iotdb.confignode.it.regionmigration.pass.daily.iotv1.RegionMigrateFileAssertions.MULTI_DATA_DIRS;
+import static org.apache.iotdb.confignode.it.regionmigration.pass.daily.iotv1.RegionMigrateFileAssertions.awaitModsVisibleOnReplicas;
+import static org.apache.iotdb.confignode.it.regionmigration.pass.daily.iotv1.RegionMigrateFileAssertions.awaitRegionReplicas;
+import static org.apache.iotdb.confignode.it.regionmigration.pass.daily.iotv1.RegionMigrateFileAssertions.awaitTsFileResourceVisibleOnReplicas;
+import static org.apache.iotdb.confignode.it.regionmigration.pass.daily.iotv1.RegionMigrateFileAssertions.awaitTsFileVisibleOnReplicas;
+import static org.apache.iotdb.confignode.it.regionmigration.pass.daily.iotv1.RegionMigrateFileAssertions.getReplicaDataNodeIds;
+
+/**
+ * Tree-model coverage for IoTConsensus region migration over multiple data dirs: a deletion (mods)
+ * must survive the snapshot transfer to the migrated peer. With several data dirs the snapshot
+ * fragments of one TsFile can be received into different folders, so the receiver groups companion
+ * files and the loader relinks them into one data dir; if that breaks, the migrated replica loses
+ * the deletion.
+ */
+@RunWith(IoTDBTestRunner.class)
+@Category({ClusterIT.class})
+public class IoTDBRegionMigrateWithDeletionMultiDataDirIT {
+
+ @Before
+ public void setUp() throws Exception {
+ EnvFactory.getEnv()
+ .getConfig()
+ .getCommonConfig()
+ .setDataReplicationFactor(2)
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS);
+ EnvFactory.getEnv().getConfig().getDataNodeConfig().setDnDataDirs(MULTI_DATA_DIRS);
+ EnvFactory.getEnv().initClusterEnvironment(1, 3);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ @Test
+ public void testRegionMigratePreservesDeletionWithMultiDataDirs() throws Exception {
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("CREATE DATABASE root.db");
+ statement.execute(
+ "INSERT INTO root.db.d1(timestamp, s1) VALUES (100, 100), (200, 200), (300, 300)");
+ statement.execute("FLUSH");
+
+ Map<Integer, Pair<Integer, Set<Integer>>> dataRegionMapWithLeader =
+ getDataRegionMapWithLeader(statement);
+ int dataRegionIdForTest =
+ dataRegionMapWithLeader
+ .keySet()
+ .stream()
+ .max(Integer::compareTo)
+ .orElseThrow(() -> new AssertionError("No DataRegion found"));
+ Set<Integer> initialReplicaDataNodeIds =
+ getReplicaDataNodeIds(statement, dataRegionIdForTest);
+
+ awaitTsFileVisibleOnReplicas("root.db", dataRegionIdForTest, initialReplicaDataNodeIds);
+ awaitTsFileResourceVisibleOnReplicas(
+ statement, "root.db", dataRegionIdForTest, initialReplicaDataNodeIds);
+
+ statement.execute("DELETE FROM root.db.d1.s1 WHERE time <= 200");
+ statement.execute("FLUSH");
+ awaitModsVisibleOnReplicas("root.db", dataRegionIdForTest, initialReplicaDataNodeIds);
+ assertDeletionVisibleOnAllReplicas(dataRegionIdForTest, 1);
+
+ Pair<Integer, Set<Integer>> leaderAndNodes = dataRegionMapWithLeader.get(dataRegionIdForTest);
+ Set<Integer> allDataNodes = getAllDataNodes(statement);
+ int leaderId = leaderAndNodes.getLeft();
+ int followerId =
+ leaderAndNodes
+ .getRight()
+ .stream()
+ .filter(id -> id != leaderId)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("No follower DataNode found"));
+ int destDataNodeId =
+ allDataNodes
+ .stream()
+ .filter(id -> id != leaderId && id != followerId)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("No destination DataNode found"));
+
+ statement.execute(
+ String.format(
+ "migrate region %d from %d to %d", dataRegionIdForTest, leaderId, destDataNodeId));
+
+ Set<Integer> migratedReplicaDataNodeIds = new HashSet<>();
+ migratedReplicaDataNodeIds.add(followerId);
+ migratedReplicaDataNodeIds.add(destDataNodeId);
+ awaitRegionReplicas(statement, dataRegionIdForTest, migratedReplicaDataNodeIds);
+ awaitModsVisibleOnReplicas(
+ "root.db", dataRegionIdForTest, Collections.singleton(destDataNodeId));
+
+ assertDeletionVisibleOnAllReplicas(dataRegionIdForTest, 1);
+ }
+ }
+
+ private void assertDeletionVisibleOnAllReplicas(int dataRegionId, int expectedCount)
+ throws Exception {
+ Set<Integer> replicaDataNodeIds;
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ replicaDataNodeIds = getReplicaDataNodeIds(statement, dataRegionId);
+ }
+ for (int dataNodeId : replicaDataNodeIds) {
+ DataNodeWrapper dataNodeWrapper =
+ EnvFactory.getEnv()
+ .dataNodeIdToWrapper(dataNodeId)
+ .orElseThrow(() -> new AssertionError("DataNode " + dataNodeId + " not found"));
+ Awaitility.await()
+ .atMost(2, TimeUnit.MINUTES)
+ .pollDelay(500, TimeUnit.MILLISECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertDeletionVisibleOnReplica(dataNodeWrapper, expectedCount));
+ }
+ }
+
+ private void assertDeletionVisibleOnReplica(DataNodeWrapper dataNodeWrapper, int expectedCount)
+ throws Exception {
+ try (Connection connection = EnvFactory.getEnv().getConnection(dataNodeWrapper);
+ Statement dataNodeStatement = connection.createStatement()) {
+ try (ResultSet countResultSet =
+ dataNodeStatement.executeQuery("SELECT COUNT(s1) FROM root.db.d1")) {
+ Assert.assertTrue(countResultSet.next());
+ Assert.assertEquals(expectedCount, countResultSet.getLong(1));
+ }
+ try (ResultSet deletedRangeResultSet =
+ dataNodeStatement.executeQuery("SELECT s1 FROM root.db.d1 WHERE time <= 200")) {
+ Assert.assertFalse(deletedRangeResultSet.next());
+ }
+ }
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv1/RegionMigrateFileAssertions.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv1/RegionMigrateFileAssertions.java
new file mode 100644
index 00000000000..b5b92afee2d
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv1/RegionMigrateFileAssertions.java
@@ -0,0 +1,169 @@
+/*
+ * 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.confignode.it.regionmigration.pass.daily.iotv1;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+
+import org.awaitility.Awaitility;
+import org.junit.Assert;
+
+import java.io.File;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+final class RegionMigrateFileAssertions {
+
+ static final String MULTI_DATA_DIRS =
+ "data/datanode/data/disk0,data/datanode/data/disk1,data/datanode/data/disk2";
+
+ private static final String SEQUENCE_FOLDER = "sequence";
+ private static final String TSFILE_SUFFIX = ".tsfile";
+ private static final String TSFILE_RESOURCE_SUFFIX = ".tsfile.resource";
+ private static final String MODS_SUFFIX = ".mods2";
+
+ private RegionMigrateFileAssertions() {}
+
+ static void awaitTsFileVisibleOnReplicas(
+ String database, int dataRegionId, Set<Integer> dataNodeIds) {
+ awaitFileVisibleOnReplicas(database, dataRegionId, dataNodeIds, TSFILE_SUFFIX);
+ }
+
+ static void awaitTsFileResourceVisibleOnReplicas(
+ String database, int dataRegionId, Set<Integer> dataNodeIds) {
+ awaitFileVisibleOnReplicas(database, dataRegionId, dataNodeIds, TSFILE_RESOURCE_SUFFIX);
+ }
+
+ static void awaitTsFileResourceVisibleOnReplicas(
+ Statement statement, String database, int dataRegionId, Set<Integer> dataNodeIds) {
+ awaitFileVisibleOnReplicas(
+ statement, database, dataRegionId, dataNodeIds, TSFILE_RESOURCE_SUFFIX);
+ }
+
+ static void awaitModsVisibleOnReplicas(
+ String database, int dataRegionId, Set<Integer> dataNodeIds) {
+ awaitFileVisibleOnReplicas(database, dataRegionId, dataNodeIds, MODS_SUFFIX);
+ }
+
+ static void awaitRegionReplicas(
+ Statement statement, int dataRegionId, Set<Integer> expectedReplicaDataNodeIds) {
+ Awaitility.await()
+ .atMost(10, TimeUnit.MINUTES)
+ .pollDelay(1, TimeUnit.SECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assert.assertEquals(
+ expectedReplicaDataNodeIds, getReplicaDataNodeIds(statement, dataRegionId)));
+ }
+
+ static Set<Integer> getReplicaDataNodeIds(Statement statement, int dataRegionId)
+ throws Exception {
+ Set<Integer> replicaDataNodeIds = new HashSet<>();
+ try (ResultSet showRegions = statement.executeQuery("SHOW REGIONS")) {
+ while (showRegions.next()) {
+ if ("DataRegion".equals(showRegions.getString("Type"))
+ && showRegions.getInt("RegionId") == dataRegionId) {
+ replicaDataNodeIds.add(showRegions.getInt("DataNodeId"));
+ }
+ }
+ }
+ Assert.assertFalse(replicaDataNodeIds.isEmpty());
+ return replicaDataNodeIds;
+ }
+
+ private static void awaitFileVisibleOnReplicas(
+ String database, int dataRegionId, Set<Integer> dataNodeIds, String suffix) {
+ awaitFileVisibleOnReplicas(null, database, dataRegionId, dataNodeIds, suffix);
+ }
+
+ private static void awaitFileVisibleOnReplicas(
+ Statement flushStatement,
+ String database,
+ int dataRegionId,
+ Set<Integer> dataNodeIds,
+ String suffix) {
+ for (int dataNodeId : dataNodeIds) {
+ DataNodeWrapper dataNodeWrapper =
+ EnvFactory.getEnv()
+ .dataNodeIdToWrapper(dataNodeId)
+ .orElseThrow(() -> new AssertionError("DataNode " + dataNodeId + " not found"));
+ Awaitility.await()
+ .atMost(2, TimeUnit.MINUTES)
+ .pollDelay(500, TimeUnit.MILLISECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ if (flushStatement != null) {
+ flushStatement.execute("FLUSH");
+ }
+ Assert.assertTrue(
+ String.format(
+ "Expected file with suffix %s for database %s region %d on DataNode %d",
+ suffix, database, dataRegionId, dataNodeId),
+ containsSequenceFileWithSuffix(
+ dataNodeWrapper, database, dataRegionId, suffix));
+ });
+ }
+ }
+
+ private static boolean containsSequenceFileWithSuffix(
+ DataNodeWrapper dataNodeWrapper, String database, int dataRegionId, String suffix) {
+ for (String dataDir : MULTI_DATA_DIRS.split(",")) {
+ File regionDir =
+ new File(
+ dataNodeWrapper.getNodePath(),
+ dataDir
+ + File.separator
+ + SEQUENCE_FOLDER
+ + File.separator
+ + database
+ + File.separator
+ + dataRegionId);
+ if (containsFileWithSuffix(regionDir, suffix)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean containsFileWithSuffix(File file, String suffix) {
+ if (!file.exists()) {
+ return false;
+ }
+ if (file.isFile()) {
+ // IoTConsensus followers can create an open zero-byte TsFile before a later FLUSH closes it.
+ return file.getName().endsWith(suffix) && (TSFILE_SUFFIX.equals(suffix) || file.length() > 0);
+ }
+ File[] children = file.listFiles();
+ if (children == null) {
+ return false;
+ }
+ for (File child : children) {
+ if (containsFileWithSuffix(child, suffix)) {
+ return true;
+ }
+ }
+ return false;
+ }
+}