This is an automated email from the ASF dual-hosted git repository. jiangtian pushed a commit to branch fix_compression_file_deletion in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 06a1e3256363bed97c4fd014b1de9166ab88485b Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 3 11:05:28 2025 +0800 Fix deleting wrong compression ratio file after flush & delete compression ratio file after region removal --- .../iotdb/it/env/cluster/node/DataNodeWrapper.java | 10 +- .../java/org/apache/iotdb/db/it/IoTDBMiscIT.java | 116 +++++++++++++++++++++ .../iotdb/db/storageengine/StorageEngine.java | 4 + .../dataregion/flush/CompressionRatio.java | 26 ++++- 4 files changed, 152 insertions(+), 4 deletions(-) diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/DataNodeWrapper.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/DataNodeWrapper.java index 1d65adacf7e..f08c085bc6c 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/DataNodeWrapper.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/DataNodeWrapper.java @@ -157,8 +157,16 @@ public class DataNodeWrapper extends AbstractNodeWrapper { return getNodePath() + File.separator + "data"; } + public String getDataNodeDir() { + return getDataDir() + File.separator + "datanode"; + } + public String getWalDir() { - return getDataDir() + File.separator + "datanode" + File.separator + "wal"; + return getDataNodeDir() + File.separator + "wal"; + } + + public String getSystemDir() { + return getDataNodeDir() + File.separator + "system"; } @Override diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBMiscIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBMiscIT.java new file mode 100644 index 00000000000..146a95fd024 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBMiscIT.java @@ -0,0 +1,116 @@ +/* + * 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.db.it; + +import org.apache.iotdb.db.storageengine.dataregion.flush.CompressionRatio; +import org.apache.iotdb.it.env.cluster.env.SimpleEnv; +import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.LocalStandaloneIT; + +import org.awaitility.Awaitility; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.io.File; +import java.sql.Connection; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertEquals; + +@RunWith(IoTDBTestRunner.class) +public class IoTDBMiscIT { + + @Category({LocalStandaloneIT.class}) + @Test + public void testCompressionRatioFile() throws SQLException { + SimpleEnv simpleEnv = new SimpleEnv(); + simpleEnv.initClusterEnvironment(1, 1); + + DataNodeWrapper nodeWrapper = simpleEnv.getDataNodeWrapper(0); + try (Connection connection = simpleEnv.getConnection(); + Statement statement = connection.createStatement()) { + statement.execute("insert into root.comprssion_ratio_file.d1(timestamp,s1) values(1,1.0)"); + statement.execute("flush"); + // one global file and two data region file (including one system region) + assertEquals(3, collectCompressionRatioFiles(nodeWrapper).size()); + + statement.execute("drop database root.comprssion_ratio_file"); + // one global file and system region file + // deleting a file may not be sensed by other processes instantly + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .pollDelay(100, TimeUnit.MILLISECONDS) + .until(() -> collectCompressionRatioFiles(nodeWrapper).size() == 2); + + statement.execute("insert into root.comprssion_ratio_file.d1(timestamp,s1) values(1,1.0)"); + statement.execute("flush"); + assertEquals(3, collectCompressionRatioFiles(nodeWrapper).size()); + + statement.execute("drop database root.comprssion_ratio_file"); + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .pollDelay(100, TimeUnit.MILLISECONDS) + .until(() -> collectCompressionRatioFiles(nodeWrapper).size() == 2); + + statement.execute("insert into root.comprssion_ratio_file.d1(timestamp,s1) values(1,1.0)"); + statement.execute("flush"); + assertEquals(3, collectCompressionRatioFiles(nodeWrapper).size()); + + statement.execute("insert into root.comprssion_ratio_file_2.d1(timestamp,s1) values(1,1.0)"); + statement.execute("flush"); + assertEquals(4, collectCompressionRatioFiles(nodeWrapper).size()); + + statement.execute("drop database root.comprssion_ratio_file"); + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .pollDelay(100, TimeUnit.MILLISECONDS) + .until(() -> collectCompressionRatioFiles(nodeWrapper).size() == 3); + + statement.execute("drop database root.comprssion_ratio_file_2"); + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .pollDelay(100, TimeUnit.MILLISECONDS) + .until(() -> collectCompressionRatioFiles(nodeWrapper).size() == 2); + } finally { + simpleEnv.cleanClusterEnvironment(); + } + } + + private List<File> collectCompressionRatioFiles(DataNodeWrapper nodeWrapper) throws SQLException { + String compressionRatioDir = + nodeWrapper.getSystemDir() + File.separator + CompressionRatio.COMPRESSION_RATIO_DIR; + File compressionRatioDirFile = new File(compressionRatioDir); + if (!compressionRatioDirFile.exists() || !compressionRatioDirFile.isDirectory()) { + return Collections.emptyList(); + } else { + File[] files = + compressionRatioDirFile.listFiles( + f -> f.getName().startsWith(CompressionRatio.FILE_PREFIX)); + return files != null ? Arrays.asList(files) : Collections.emptyList(); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java index 034d184bd03..0bc3f408bba 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java @@ -67,6 +67,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.repair.RepairLogg import org.apache.iotdb.db.storageengine.dataregion.compaction.repair.UnsortedFileRepairTaskScheduler; import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionScheduleTaskManager; import org.apache.iotdb.db.storageengine.dataregion.flush.CloseFileListener; +import org.apache.iotdb.db.storageengine.dataregion.flush.CompressionRatio; import org.apache.iotdb.db.storageengine.dataregion.flush.FlushListener; import org.apache.iotdb.db.storageengine.dataregion.flush.TsFileFlushPolicy; import org.apache.iotdb.db.storageengine.dataregion.flush.TsFileFlushPolicy.DirectFlushPolicy; @@ -770,6 +771,7 @@ public class StorageEngine implements IService { DataRegion region = deletingDataRegionMap.computeIfAbsent(regionId, k -> dataRegionMap.remove(regionId)); if (region != null) { + LOGGER.info("Removing data region {}", regionId); region.markDeleted(); try { region.abortCompaction(); @@ -813,6 +815,8 @@ public class StorageEngine implements IService { WRITING_METRICS.removeFlushingMemTableStatusMetrics(regionId); WRITING_METRICS.removeActiveMemtableCounterMetrics(regionId); FileMetrics.getInstance().deleteRegion(region.getDatabaseName(), region.getDataRegionId()); + CompressionRatio.getInstance().removeDataRegionRatio(String.valueOf(regionId.getId())); + LOGGER.info("Removed data region {}", regionId); } catch (Exception e) { LOGGER.error( "Error occurs when deleting data region {}-{}", diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/CompressionRatio.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/CompressionRatio.java index c22afac0817..724f4d0decf 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/CompressionRatio.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/CompressionRatio.java @@ -54,10 +54,10 @@ public class CompressionRatio { private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig(); - static final String COMPRESSION_RATIO_DIR = "compression_ratio"; + public static final String COMPRESSION_RATIO_DIR = "compression_ratio"; private static final String FILE_PREFIX_BEFORE_V121 = "Ratio-"; - private static final String FILE_PREFIX = "Compress-"; + public static final String FILE_PREFIX = "Compress-"; private static final String SEPARATOR = "-"; @@ -109,6 +109,7 @@ public class CompressionRatio { String.format( Locale.ENGLISH, RATIO_FILE_PATH_FORMAT, totalMemorySize.get(), totalDiskSize)); persist(oldDataNodeFile, newDataNodeFile); + this.oldFileName = newDataNodeFile.getName(); Pair<Long, Long> dataRegionCompressionRatio = dataRegionRatioMap.computeIfAbsent(dataRegionId, id -> new Pair<>(0L, 0L)); @@ -137,6 +138,26 @@ public class CompressionRatio { persist(oldDataRegionFile, newDataRegionFile); } + public synchronized void removeDataRegionRatio(String dataRegionId) { + Pair<Long, Long> dataRegionCompressionRatio = dataRegionRatioMap.remove(dataRegionId); + if (dataRegionCompressionRatio == null) { + return; + } + File oldDataRegionFile = + SystemFileFactory.INSTANCE.getFile( + directory, + String.format( + Locale.ENGLISH, + RATIO_FILE_PATH_FORMAT, + dataRegionCompressionRatio.getLeft(), + dataRegionCompressionRatio.getRight()) + + "." + + dataRegionId); + if (!oldDataRegionFile.delete() && oldDataRegionFile.exists()) { + LOGGER.warn("Can't delete old data region compression file {}", oldDataRegionFile); + } + } + /** Get the average compression ratio for all closed files */ public double getRatio() { return (double) totalMemorySize.get() / totalDiskSize; @@ -171,7 +192,6 @@ public class CompressionRatio { oldFile.getAbsolutePath(), newFile.getAbsolutePath()); } - this.oldFileName = newFile.getName(); } private void checkDirectoryExist() throws IOException {
