From 224d85fb2b3f4ad5865dbe888cf279ac3804eb84 Mon Sep 17 00:00:00 2001 From: 761417898 <761417898@qq.com> Date: Mon, 9 Nov 2026 15:16:25 +0800 Subject: [PATCH 1/2] Fix snapshots of deleted TsFile resources --- .../IoTDBRemoveDataNodeAfterDeleteIT.java | 153 ++++++++++++++++++ .../iotdb/db/i18n/StorageEngineMessages.java | 2 +- .../iotdb/db/i18n/StorageEngineMessages.java | 2 +- .../dataregion/snapshot/SnapshotTaker.java | 13 +- .../dataregion/tsfile/TsFileResource.java | 4 + .../dataregion/DataRegionTest.java | 39 +++++ .../snapshot/IoTDBSnapshotTest.java | 30 ++++ .../dataregion/tsfile/TsFileResourceTest.java | 5 + 8 files changed, 245 insertions(+), 3 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeAfterDeleteIT.java diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeAfterDeleteIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeAfterDeleteIT.java new file mode 100644 index 0000000000000..a3b1a90dff551 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeAfterDeleteIT.java @@ -0,0 +1,153 @@ +/* + * 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.removedatanode; + +import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; +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.junit.After; +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.concurrent.TimeUnit; + +import static org.awaitility.Awaitility.await; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertTrue; + +@Category(ClusterIT.class) +@RunWith(IoTDBTestRunner.class) +public class IoTDBRemoveDataNodeAfterDeleteIT { + @Before + public void setUp() throws Exception { + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS) + .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS) + .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS) + .setSchemaReplicationFactor(1) + .setDataReplicationFactor(1) + .setSchemaRegionGroupExtensionPolicy("CUSTOM") + .setDataRegionGroupExtensionPolicy("CUSTOM") + .setDefaultSchemaRegionGroupNumPerDatabase(1) + .setDefaultDataRegionGroupNumPerDatabase(1) + .setEnableSeqSpaceCompaction(false) + .setEnableUnseqSpaceCompaction(false) + .setEnableCrossSpaceCompaction(false); + EnvFactory.getEnv().initClusterEnvironment(1, 3); + } + + @After + public void tearDown() throws Exception { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testRemoveAfterDeletingUnflushedRow() throws Exception { + removeAfterDelete(false); + } + + @Test + public void testRemoveAfterDeletingUnflushedRowWithRetainedData() throws Exception { + removeAfterDelete(true); + } + + private void removeAfterDelete(boolean retainData) throws Exception { + int removedNodeId = -1; + int removedPort = -1; + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE root.snapshot_delete"); + statement.execute("CREATE TIMESERIES root.snapshot_delete.d.s WITH DATATYPE=INT32"); + if (retainData) { + statement.execute("INSERT INTO root.snapshot_delete.d(time,s) VALUES(1,10)"); + statement.execute("FLUSH"); + } + statement.execute("INSERT INTO root.snapshot_delete.d(time,s) VALUES(2,20)"); + statement.execute("DELETE FROM root.snapshot_delete.d.s WHERE time = 2"); + try (ResultSet regions = statement.executeQuery("SHOW REGIONS")) { + while (regions.next()) { + if ("DataRegion".equals(regions.getString(ColumnHeaderConstant.TYPE)) + && "root.snapshot_delete".equals(regions.getString(ColumnHeaderConstant.DATABASE))) { + removedNodeId = regions.getInt(ColumnHeaderConstant.DATA_NODE_ID); + } + } + } + assertTrue("The test database must have a DataRegion", removedNodeId >= 0); + try (ResultSet nodes = statement.executeQuery("SHOW DATANODES")) { + while (nodes.next()) { + if (nodes.getInt(ColumnHeaderConstant.NODE_ID) == removedNodeId) { + removedPort = nodes.getInt(ColumnHeaderConstant.RPC_PORT); + } + } + } + assertTrue(removedPort > 0); + statement.execute("REMOVE DATANODE " + removedNodeId); + } + + final int nodeId = removedNodeId; + final int rpcPort = removedPort; + DataNodeWrapper removed = + EnvFactory.getEnv().getDataNodeWrapperList().stream() + .filter(node -> node.getPort() == rpcPort) + .findFirst() + .get(); + DataNodeWrapper survivor = + EnvFactory.getEnv().getDataNodeWrapperList().stream() + .filter(node -> node.getPort() != rpcPort) + .findFirst() + .get(); + try (Connection connection = EnvFactory.getEnv().getConnection(survivor); + Statement statement = connection.createStatement()) { + await() + .atMost(600, TimeUnit.SECONDS) + .pollInterval(1, TimeUnit.SECONDS) + .untilAsserted( + () -> { + try (ResultSet nodes = statement.executeQuery("SHOW DATANODES")) { + while (nodes.next()) { + assertNotEquals(nodeId, nodes.getInt(ColumnHeaderConstant.NODE_ID)); + } + } + assertFalse("Removed DataNode must exit", removed.isAlive()); + }); + try (ResultSet rows = statement.executeQuery("SELECT s FROM root.snapshot_delete.d")) { + if (retainData) { + assertTrue(rows.next()); + assertEquals(1, rows.getLong("Time")); + assertEquals(10, rows.getInt("root.snapshot_delete.d.s")); + } + assertFalse("Deleted data must not reappear after migration", rows.next()); + } + } + } +} diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java index 7133bf09abda5..9aa37d450a8a1 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java @@ -635,7 +635,7 @@ private StorageEngineMessages() {} public static final String CANNOT_SNAPSHOT_UNCLOSED_TSFILE = "Cannot create snapshot because TsFile {} is not closed"; public static final String HARD_LINK_TARGET_DIR_NOT_EXIST = "Hard link target dir {} doesn't exist"; - public static final String HARD_LINK_SOURCE_FILE_NOT_EXIST = "Hard link source file {} doesn't exist, this file will be ignored."; + public static final String HARD_LINK_SOURCE_FILE_NOT_EXIST = "Hard link source file {} doesn't exist."; public static final String COPY_TARGET_DIR_NOT_EXIST = "Copy target dir {} doesn't exist"; public static final String COPY_SOURCE_FILE_NOT_EXIST = "Copy source file {} doesn't exist"; public static final String CANNOT_CREATE_DIRECTORY = "Cannot create directory "; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java index 1603065835bda..784ee7287e46d 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java @@ -616,7 +616,7 @@ private StorageEngineMessages() {} public static final String CANNOT_SNAPSHOT_UNCLOSED_TSFILE = "无法创建快照，因为 TsFile {} 尚未关闭"; public static final String HARD_LINK_TARGET_DIR_NOT_EXIST = "硬链接目标目录 {} 不存在"; - public static final String HARD_LINK_SOURCE_FILE_NOT_EXIST = "硬链接源文件 {} 不存在，该文件将被忽略。"; + public static final String HARD_LINK_SOURCE_FILE_NOT_EXIST = "硬链接源文件 {} 不存在。"; public static final String COPY_TARGET_DIR_NOT_EXIST = "复制目标目录 {} 不存在"; public static final String COPY_SOURCE_FILE_NOT_EXIST = "复制源文件 {} 不存在"; public static final String CANNOT_CREATE_DIRECTORY = "无法创建目录: "; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/snapshot/SnapshotTaker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/snapshot/SnapshotTaker.java index 854909bc14c7b..d32615fec33b6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/snapshot/SnapshotTaker.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/snapshot/SnapshotTaker.java @@ -35,6 +35,7 @@ import org.slf4j.LoggerFactory; import java.io.File; +import java.io.FileNotFoundException; import java.io.IOException; import java.nio.file.Files; import java.util.Arrays; @@ -226,6 +227,16 @@ private void readUnlockTheFile() { private boolean createSnapshot(List resources, String snapshotId) { try { + // Closing empty files can delete resources after readLockTheFile() captured them. + // Remove them from subsequent snapshot processing and balance their captured read locks. + resources.removeIf( + resource -> { + if (resource.isDeleted()) { + resource.readUnlock(); + return true; + } + return false; + }); for (TsFileResource resource : resources) { if (!resource.isClosed()) { LOGGER.error(StorageEngineMessages.CANNOT_SNAPSHOT_UNCLOSED_TSFILE, resource); @@ -279,7 +290,7 @@ private void createHardLink(File target, File source) throws IOException { LOGGER.error(StorageEngineMessages.HARD_LINK_TARGET_DIR_NOT_EXIST, target.getParentFile()); } if (!checkHardLinkSourceFile(source, 10)) { - return; + throw new FileNotFoundException(source.getAbsolutePath()); } Files.deleteIfExists(target.toPath()); SnapshotFileUtils.createLinkOrCopy(target.toPath(), source.toPath()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java index a431665aa17ca..5629173986a91 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java @@ -70,6 +70,7 @@ import java.io.BufferedOutputStream; import java.io.File; +import java.io.FileNotFoundException; import java.io.FileOutputStream; import java.io.IOException; import java.io.InputStream; @@ -332,6 +333,9 @@ private ITimeIndex getTimeIndexForSerialization() throws IOException { /** deserialize from disk */ public void deserialize() throws IOException { try (InputStream inputStream = fsFactory.getBufferedInputStream(file + RESOURCE_SUFFIX)) { + if (inputStream == null) { + throw new FileNotFoundException(file + RESOURCE_SUFFIX); + } // The first byte is VERSION_NUMBER, second byte is timeIndexType. ReadWriteIOUtils.readByte(inputStream); timeIndex = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java index 250ccc18dadce..e131ae6e468d7 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java @@ -33,6 +33,7 @@ import org.apache.iotdb.commons.path.NonAlignedFullPath; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.commons.utils.TimePartitionUtils; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; @@ -83,6 +84,8 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.v1.Deletion; import org.apache.iotdb.db.storageengine.dataregion.modification.v1.ModificationFileV1; import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; +import org.apache.iotdb.db.storageengine.dataregion.snapshot.SnapshotLoader; +import org.apache.iotdb.db.storageengine.dataregion.snapshot.SnapshotTaker; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator; @@ -614,6 +617,42 @@ private void setMemoryAdmissionTestThreshold(String name, Object value) throws E field.set(SystemInfo.getInstance(), value); } + @Test + public void testSnapshotAfterDeletingUnflushedData() throws Exception { + dataRegion.setAllowCompaction(false); + TSRecord record = new TSRecord(deviceId, 1); + record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, "10")); + dataRegion.insert(buildInsertRowNodeByTSRecord(record)); + TsFileResource resource = dataRegion.getTsFileManager().getTsFileList(true).get(0); + Assert.assertFalse(resource.isClosed()); + MeasurementPath path = new MeasurementPath(deviceId, measurementId); + dataRegion.deleteByDevice( + path, + new DeleteDataNode( + new PlanNodeId("delete"), + Collections.singletonList(path), + Long.MIN_VALUE, + Long.MAX_VALUE)); + + File snapshotDir = new File(TestConstant.OUTPUT_DATA_DIR, "snapshot-deleted-unflushed"); + try { + Assert.assertTrue( + new SnapshotTaker(dataRegion).takeFullSnapshot(snapshotDir.getAbsolutePath(), true)); + Assert.assertTrue(resource.isDeleted()); + Assert.assertFalse(resource.tsFileExists()); + Assert.assertTrue(resource.tryWriteLock()); + resource.writeUnlock(); + DataRegion loaded = + new SnapshotLoader(snapshotDir.getAbsolutePath(), storageGroup, "0") + .loadSnapshotForStateMachine(); + Assert.assertNotNull(loaded); + Assert.assertTrue(loaded.getTsFileManager().getTsFileList(true).isEmpty()); + Assert.assertTrue(loaded.getTsFileManager().getTsFileList(false).isEmpty()); + } finally { + FileUtils.recursivelyDeleteFolder(snapshotDir.getAbsolutePath()); + } + } + @Test public void testUnseqUnsealedDelete() throws WriteProcessException, IOException, MetadataException { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/snapshot/IoTDBSnapshotTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/snapshot/IoTDBSnapshotTest.java index 84d8deac3d78e..4f70a3f131950 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/snapshot/IoTDBSnapshotTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/snapshot/IoTDBSnapshotTest.java @@ -148,6 +148,36 @@ public void testCreateSnapshot() } } + @Test + public void testRejectSnapshotWithMissingResource() throws Exception { + String[][] originDataDirs = IoTDBDescriptor.getInstance().getConfig().getTierDataDirs(); + IoTDBDescriptor.getInstance().getConfig().setTierDataDirs(testDataDirs); + TierManager.getInstance().resetFolders(); + File snapshotDir = new File("target/snapshot-missing-resource"); + try { + List resources = writeTsFiles(testDataDirs[0]); + TsFileResource resource = resources.get(0); + Files.delete(new File(resource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).toPath()); + DataRegion region = new DataRegion(testSgName, "0"); + region.getTsFileManager().add(resource, true); + Assert.assertFalse( + new SnapshotTaker(region).takeFullSnapshot(snapshotDir.getAbsolutePath(), true)); + SnapshotLogAnalyzer analyzer = + new SnapshotLogAnalyzer(new File(snapshotDir, SnapshotLogger.SNAPSHOT_LOG_NAME)); + try { + Assert.assertFalse(analyzer.isSnapshotComplete()); + } finally { + analyzer.close(); + } + Assert.assertTrue(resource.tryWriteLock()); + resource.writeUnlock(); + } finally { + FileUtils.recursivelyDeleteFolder(snapshotDir.getAbsolutePath()); + IoTDBDescriptor.getInstance().getConfig().setTierDataDirs(originDataDirs); + TierManager.getInstance().resetFolders(); + } + } + @Test public void testRejectSnapshotWithUnclosedTsFile() throws IOException, WriteProcessException, DirectoryNotLegalException { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResourceTest.java index 453e934b5b063..f0149f799c13d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResourceTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResourceTest.java @@ -99,6 +99,11 @@ public void tearDown() throws IOException { } } + @Test(expected = IOException.class) + public void testDeserializeMissingResource() throws IOException { + tsFileResource.deserialize(); + } + @Test public void testSerializeAndDeserialize() throws IOException { tsFileResource.serialize(); From 7f648624735341953aae69b42b169446b8051b4e Mon Sep 17 00:00:00 2001 From: 761417898 <761417898@qq.com> Date: Mon, 9 Nov 2026 15:35:26 +0800 Subject: [PATCH 2/2] Simplify snapshot regression coverage --- .../IoTDBRemoveDataNodeAfterDeleteIT.java | 153 ------------------ 1 file changed, 153 deletions(-) delete mode 100644 integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeAfterDeleteIT.java diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeAfterDeleteIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeAfterDeleteIT.java deleted file mode 100644 index a3b1a90dff551..0000000000000 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeAfterDeleteIT.java +++ /dev/null @@ -1,153 +0,0 @@ -/* - * 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.removedatanode; - -import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; -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.junit.After; -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.concurrent.TimeUnit; - -import static org.awaitility.Awaitility.await; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertNotEquals; -import static org.junit.Assert.assertTrue; - -@Category(ClusterIT.class) -@RunWith(IoTDBTestRunner.class) -public class IoTDBRemoveDataNodeAfterDeleteIT { - @Before - public void setUp() throws Exception { - EnvFactory.getEnv() - .getConfig() - .getCommonConfig() - .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS) - .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS) - .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS) - .setSchemaReplicationFactor(1) - .setDataReplicationFactor(1) - .setSchemaRegionGroupExtensionPolicy("CUSTOM") - .setDataRegionGroupExtensionPolicy("CUSTOM") - .setDefaultSchemaRegionGroupNumPerDatabase(1) - .setDefaultDataRegionGroupNumPerDatabase(1) - .setEnableSeqSpaceCompaction(false) - .setEnableUnseqSpaceCompaction(false) - .setEnableCrossSpaceCompaction(false); - EnvFactory.getEnv().initClusterEnvironment(1, 3); - } - - @After - public void tearDown() throws Exception { - EnvFactory.getEnv().cleanClusterEnvironment(); - } - - @Test - public void testRemoveAfterDeletingUnflushedRow() throws Exception { - removeAfterDelete(false); - } - - @Test - public void testRemoveAfterDeletingUnflushedRowWithRetainedData() throws Exception { - removeAfterDelete(true); - } - - private void removeAfterDelete(boolean retainData) throws Exception { - int removedNodeId = -1; - int removedPort = -1; - try (Connection connection = EnvFactory.getEnv().getConnection(); - Statement statement = connection.createStatement()) { - statement.execute("CREATE DATABASE root.snapshot_delete"); - statement.execute("CREATE TIMESERIES root.snapshot_delete.d.s WITH DATATYPE=INT32"); - if (retainData) { - statement.execute("INSERT INTO root.snapshot_delete.d(time,s) VALUES(1,10)"); - statement.execute("FLUSH"); - } - statement.execute("INSERT INTO root.snapshot_delete.d(time,s) VALUES(2,20)"); - statement.execute("DELETE FROM root.snapshot_delete.d.s WHERE time = 2"); - try (ResultSet regions = statement.executeQuery("SHOW REGIONS")) { - while (regions.next()) { - if ("DataRegion".equals(regions.getString(ColumnHeaderConstant.TYPE)) - && "root.snapshot_delete".equals(regions.getString(ColumnHeaderConstant.DATABASE))) { - removedNodeId = regions.getInt(ColumnHeaderConstant.DATA_NODE_ID); - } - } - } - assertTrue("The test database must have a DataRegion", removedNodeId >= 0); - try (ResultSet nodes = statement.executeQuery("SHOW DATANODES")) { - while (nodes.next()) { - if (nodes.getInt(ColumnHeaderConstant.NODE_ID) == removedNodeId) { - removedPort = nodes.getInt(ColumnHeaderConstant.RPC_PORT); - } - } - } - assertTrue(removedPort > 0); - statement.execute("REMOVE DATANODE " + removedNodeId); - } - - final int nodeId = removedNodeId; - final int rpcPort = removedPort; - DataNodeWrapper removed = - EnvFactory.getEnv().getDataNodeWrapperList().stream() - .filter(node -> node.getPort() == rpcPort) - .findFirst() - .get(); - DataNodeWrapper survivor = - EnvFactory.getEnv().getDataNodeWrapperList().stream() - .filter(node -> node.getPort() != rpcPort) - .findFirst() - .get(); - try (Connection connection = EnvFactory.getEnv().getConnection(survivor); - Statement statement = connection.createStatement()) { - await() - .atMost(600, TimeUnit.SECONDS) - .pollInterval(1, TimeUnit.SECONDS) - .untilAsserted( - () -> { - try (ResultSet nodes = statement.executeQuery("SHOW DATANODES")) { - while (nodes.next()) { - assertNotEquals(nodeId, nodes.getInt(ColumnHeaderConstant.NODE_ID)); - } - } - assertFalse("Removed DataNode must exit", removed.isAlive()); - }); - try (ResultSet rows = statement.executeQuery("SELECT s FROM root.snapshot_delete.d")) { - if (retainData) { - assertTrue(rows.next()); - assertEquals(1, rows.getLong("Time")); - assertEquals(10, rows.getInt("root.snapshot_delete.d.s")); - } - assertFalse("Deleted data must not reappear after migration", rows.next()); - } - } - } -}