From 6d7092533bd112fe076cfda8cfe4464e79f9a4f1 Mon Sep 17 00:00:00 2001 From: MiniSho <42286868+MiniSho@users.noreply.github.com> Date: Thu, 6 Jun 2024 11:22:07 +0800 Subject: [PATCH 01/11] init --- .../org/apache/iotdb/db/conf/IoTDBConfig.java | 18 +++-- .../apache/iotdb/db/conf/IoTDBDescriptor.java | 9 +++ .../execution/load/LoadTsFileManager.java | 73 ++++++++++++++----- 3 files changed, 72 insertions(+), 28 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java index 4a763640eb9a7..b821de0851639 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java @@ -310,8 +310,7 @@ public class IoTDBConfig { {IoTDBConstant.DN_DEFAULT_DATA_DIR + File.separator + IoTDBConstant.DATA_FOLDER_NAME} }; - private String loadTsFileDir = - tierDataDirs[0][0] + File.separator + IoTDBConstant.LOAD_TSFILE_FOLDER_NAME; + private String[] loadTsFileDirs = new String[0]; /** Strategy of multiple directories. */ private String multiDirStrategyClassName = null; @@ -1276,7 +1275,9 @@ public void updatePath() { private void formulateFolders() { systemDir = addDataHomeDir(systemDir); schemaDir = addDataHomeDir(schemaDir); - loadTsFileDir = addDataHomeDir(loadTsFileDir); + for (int i = 0; i < loadTsFileDirs.length; i++) { + loadTsFileDirs[i] = addDataHomeDir(loadTsFileDirs[i]); + } consensusDir = addDataHomeDir(consensusDir); dataRegionConsensusDir = addDataHomeDir(dataRegionConsensusDir); ratisDataRegionSnapshotDir = addDataHomeDir(ratisDataRegionSnapshotDir); @@ -1427,7 +1428,6 @@ public void setTierDataDirs(String[][] tierDataDirs) { // TODO(szywilliam): rewrite the logic here when ratis supports complete snapshot semantic setRatisDataRegionSnapshotDir( tierDataDirs[0][0] + File.separator + IoTDBConstant.SNAPSHOT_FOLDER_NAME); - setLoadTsFileDir(tierDataDirs[0][0] + File.separator + IoTDBConstant.LOAD_TSFILE_FOLDER_NAME); } public String getRpcAddress() { @@ -1462,12 +1462,14 @@ public void setSystemDir(String systemDir) { this.systemDir = systemDir; } - public String getLoadTsFileDir() { - return loadTsFileDir; + public String[] getLoadTsFileDirs() { + return (Objects.isNull(this.loadTsFileDirs) || this.loadTsFileDirs.length == 0) + ? new String[] {tierDataDirs[0][0] + File.separator + IoTDBConstant.LOAD_TSFILE_FOLDER_NAME} + : this.loadTsFileDirs; } - public void setLoadTsFileDir(String loadTsFileDir) { - this.loadTsFileDir = loadTsFileDir; + public void setLoadTsFileDirs(String[] loadTsFileDirs) { + this.loadTsFileDirs = loadTsFileDirs; } public String getSchemaDir() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java index fd1068b125773..7344b6b938221 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java @@ -363,6 +363,15 @@ public void loadProperties(Properties properties) throws BadNodeUrlException, IO conf.setConsensusDir(properties.getProperty("dn_consensus_dir", conf.getConsensusDir())); + conf.setLoadTsFileDirs( + Arrays.stream( + properties + .getProperty("load_file_dirs", String.join(",", conf.getLoadTsFileDirs())) + .trim() + .split(",")) + .filter(dir -> !dir.isEmpty()) + .toArray(String[]::new)); + long forceMlogPeriodInMs = Long.parseLong( properties.getProperty( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java index b568b8879884d..728afc867ec43 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java @@ -29,6 +29,7 @@ import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.exception.DiskSpaceInsufficientException; import org.apache.iotdb.db.exception.LoadFileException; import org.apache.iotdb.db.pipe.agent.PipeAgent; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode; @@ -38,6 +39,8 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileResourceUtils; +import org.apache.iotdb.db.storageengine.rescon.disk.FolderManager; +import org.apache.iotdb.db.storageengine.rescon.disk.strategy.DirectoryStrategyType; import org.apache.iotdb.metrics.utils.MetricLevel; import org.apache.tsfile.common.constant.TsFileConstant; @@ -51,12 +54,15 @@ import java.nio.file.DirectoryNotEmptyException; import java.nio.file.Files; import java.nio.file.Path; +import java.util.Arrays; import java.util.HashMap; import java.util.Map; import java.util.Objects; import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.PriorityBlockingQueue; +import java.util.stream.Collectors; +import java.util.stream.Stream; /** * {@link LoadTsFileManager} is used for dealing with {@link LoadTsFilePieceNode} and {@link @@ -73,17 +79,27 @@ public class LoadTsFileManager { private static final String MESSAGE_WRITER_MANAGER_HAS_BEEN_CLOSED = "%s TsFileWriterManager has been closed."; private static final String MESSAGE_DELETE_FAIL = "failed to delete {}."; + private static final String[] LOAD_BASE_DIRS = CONFIG.getLoadTsFileDirs(); - private final File loadDir; + private static FolderManager folderManager; private final Map uuid2WriterManager = new ConcurrentHashMap<>(); private final Map uuid2CleanupTask = new ConcurrentHashMap<>(); private final PriorityBlockingQueue cleanupTaskQueue = new PriorityBlockingQueue<>(); - public LoadTsFileManager() { - this.loadDir = SystemFileFactory.INSTANCE.getFile(CONFIG.getLoadTsFileDir()); + static { + try { + folderManager = + new FolderManager(Arrays.asList(LOAD_BASE_DIRS), DirectoryStrategyType.SEQUENCE_STRATEGY); + } catch (final DiskSpaceInsufficientException e) { + LOGGER.error( + "Fail to create load tsfile folders allocation strategy because all disks of folders are full.", + e); + } + } + public LoadTsFileManager() { registerCleanupTaskExecutor(); recover(); } @@ -130,14 +146,20 @@ private void cleanupTasks() { } private void recover() { - if (!loadDir.exists()) { + if (!Arrays.stream(LOAD_BASE_DIRS).map(File::new).allMatch(File::exists)) { return; } - final File[] files = loadDir.listFiles(); - if (files == null) { - return; - } + final File[] files = + Arrays.stream(LOAD_BASE_DIRS) + .map(File::new) + .flatMap( + dir -> { + File[] tmpfiles = dir.listFiles(); + return tmpfiles != null ? Arrays.stream(tmpfiles) : Stream.empty(); + }) + .toArray(File[]::new); + for (final File taskDir : files) { String uuid = taskDir.getName(); TsFileWriterManager writerManager = new TsFileWriterManager(taskDir); @@ -171,7 +193,13 @@ public void writeToDataRegion(DataRegion dataRegion, LoadTsFilePieceNode pieceNo final TsFileWriterManager writerManager = uuid2WriterManager.computeIfAbsent( uuid, - o -> new TsFileWriterManager(SystemFileFactory.INSTANCE.getFile(loadDir, uuid))); + o -> { + try { + return new TsFileWriterManager(new File(folderManager.getNextFolder(), uuid)); + } catch (DiskSpaceInsufficientException e) { + throw new RuntimeException(e); + } + }); for (TsFileData tsFileData : pieceNode.getAllTsFileData()) { if (!tsFileData.isModification()) { ChunkData chunkData = (ChunkData) tsFileData; @@ -229,17 +257,22 @@ private void forceCloseWriterManager(String uuid) { writerManager.close(); } - final Path loadDirPath = loadDir.toPath(); - if (!Files.exists(loadDirPath)) { - return; - } - try { - Files.deleteIfExists(loadDirPath); - LOGGER.info("Load dir {} was deleted.", loadDirPath); - } catch (DirectoryNotEmptyException e) { - LOGGER.info("Load dir {} is not empty, skip deleting.", loadDirPath); - } catch (IOException e) { - LOGGER.info(MESSAGE_DELETE_FAIL, loadDirPath); + for (Path loadDirPath : + Arrays.stream(LOAD_BASE_DIRS) + .map(File::new) + .map(File::toPath) + .collect(Collectors.toList())) { + if (!Files.exists(loadDirPath)) { + continue; + } + try { + Files.deleteIfExists(loadDirPath); + LOGGER.info("Load dir {} was deleted.", loadDirPath); + } catch (DirectoryNotEmptyException e) { + LOGGER.info("Load dir {} is not empty, skip deleting.", loadDirPath); + } catch (IOException e) { + LOGGER.info(MESSAGE_DELETE_FAIL, loadDirPath); + } } } From 69fd426d1e2c78e416ca2f38dec2a50689b0c6a7 Mon Sep 17 00:00:00 2001 From: MiniSho <42286868+MiniSho@users.noreply.github.com> Date: Thu, 6 Jun 2024 13:49:57 +0800 Subject: [PATCH 02/11] improve --- .../execution/load/LoadTsFileManager.java | 38 ++++++++++--------- 1 file changed, 21 insertions(+), 17 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java index 728afc867ec43..35238b2a79667 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java @@ -146,13 +146,13 @@ private void cleanupTasks() { } private void recover() { - if (!Arrays.stream(LOAD_BASE_DIRS).map(File::new).allMatch(File::exists)) { + final File[] baseDirs = Arrays.stream(LOAD_BASE_DIRS).map(File::new).toArray(File[]::new); + if (!Arrays.stream(baseDirs).allMatch(File::exists)) { return; } final File[] files = - Arrays.stream(LOAD_BASE_DIRS) - .map(File::new) + Arrays.stream(baseDirs) .flatMap( dir -> { File[] tmpfiles = dir.listFiles(); @@ -160,20 +160,24 @@ private void recover() { }) .toArray(File[]::new); - for (final File taskDir : files) { - String uuid = taskDir.getName(); - TsFileWriterManager writerManager = new TsFileWriterManager(taskDir); - - uuid2WriterManager.put(uuid, writerManager); - writerManager.close(); - - synchronized (uuid2CleanupTask) { - final CleanupTask cleanupTask = - new CleanupTask(uuid, CONFIG.getLoadCleanupTaskExecutionDelayTimeSeconds() * 1000); - uuid2CleanupTask.put(uuid, cleanupTask); - cleanupTaskQueue.add(cleanupTask); - } - } + Arrays.stream(files) + .parallel() + .forEach( + taskDir -> { + String uuid = taskDir.getName(); + TsFileWriterManager writerManager = new TsFileWriterManager(taskDir); + + uuid2WriterManager.put(uuid, writerManager); + writerManager.close(); + + synchronized (uuid2CleanupTask) { + final CleanupTask cleanupTask = + new CleanupTask( + uuid, CONFIG.getLoadCleanupTaskExecutionDelayTimeSeconds() * 1000); + uuid2CleanupTask.put(uuid, cleanupTask); + cleanupTaskQueue.add(cleanupTask); + } + }); } public void writeToDataRegion(DataRegion dataRegion, LoadTsFilePieceNode pieceNode, String uuid) From 5596783d020d6e982fccdead82c0019163b42f92 Mon Sep 17 00:00:00 2001 From: MiniSho <42286868+MiniSho@users.noreply.github.com> Date: Fri, 7 Jun 2024 10:14:14 +0800 Subject: [PATCH 03/11] fix --- .../main/java/org/apache/iotdb/db/conf/IoTDBConfig.java | 9 +++++++-- .../java/org/apache/iotdb/db/conf/IoTDBDescriptor.java | 9 --------- 2 files changed, 7 insertions(+), 11 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java index b821de0851639..fe08e6ddcc1f4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java @@ -1428,6 +1428,7 @@ public void setTierDataDirs(String[][] tierDataDirs) { // TODO(szywilliam): rewrite the logic here when ratis supports complete snapshot semantic setRatisDataRegionSnapshotDir( tierDataDirs[0][0] + File.separator + IoTDBConstant.SNAPSHOT_FOLDER_NAME); + setLoadTsFileDirs(tierDataDirs); } public String getRpcAddress() { @@ -1468,8 +1469,12 @@ public String[] getLoadTsFileDirs() { : this.loadTsFileDirs; } - public void setLoadTsFileDirs(String[] loadTsFileDirs) { - this.loadTsFileDirs = loadTsFileDirs; + public void setLoadTsFileDirs(String[][] tierDataDirs) { + this.loadTsFileDirs = new String[tierDataDirs.length]; + for (int i = 0; i < tierDataDirs.length; i++) { + this.loadTsFileDirs[i] = + tierDataDirs[i][0] + File.separator + IoTDBConstant.LOAD_TSFILE_FOLDER_NAME; + } } public String getSchemaDir() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java index 7344b6b938221..fd1068b125773 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java @@ -363,15 +363,6 @@ public void loadProperties(Properties properties) throws BadNodeUrlException, IO conf.setConsensusDir(properties.getProperty("dn_consensus_dir", conf.getConsensusDir())); - conf.setLoadTsFileDirs( - Arrays.stream( - properties - .getProperty("load_file_dirs", String.join(",", conf.getLoadTsFileDirs())) - .trim() - .split(",")) - .filter(dir -> !dir.isEmpty()) - .toArray(String[]::new)); - long forceMlogPeriodInMs = Long.parseLong( properties.getProperty( From 121c5bddf61e6302d6027c05db8b29e0052fef43 Mon Sep 17 00:00:00 2001 From: MiniSho <42286868+MiniSho@users.noreply.github.com> Date: Fri, 7 Jun 2024 10:28:45 +0800 Subject: [PATCH 04/11] fix --- .../org/apache/iotdb/db/conf/IoTDBConfig.java | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java index fe08e6ddcc1f4..ce183f42c682a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java @@ -310,7 +310,9 @@ public class IoTDBConfig { {IoTDBConstant.DN_DEFAULT_DATA_DIR + File.separator + IoTDBConstant.DATA_FOLDER_NAME} }; - private String[] loadTsFileDirs = new String[0]; + private String[] loadTsFileDirs = { + tierDataDirs[0][0] + File.separator + IoTDBConstant.LOAD_TSFILE_FOLDER_NAME + }; /** Strategy of multiple directories. */ private String multiDirStrategyClassName = null; @@ -1275,9 +1277,6 @@ public void updatePath() { private void formulateFolders() { systemDir = addDataHomeDir(systemDir); schemaDir = addDataHomeDir(schemaDir); - for (int i = 0; i < loadTsFileDirs.length; i++) { - loadTsFileDirs[i] = addDataHomeDir(loadTsFileDirs[i]); - } consensusDir = addDataHomeDir(consensusDir); dataRegionConsensusDir = addDataHomeDir(dataRegionConsensusDir); ratisDataRegionSnapshotDir = addDataHomeDir(ratisDataRegionSnapshotDir); @@ -1301,6 +1300,11 @@ private void formulateFolders() { queryDir = addDataHomeDir(queryDir); sortTmpDir = addDataHomeDir(sortTmpDir); formulateDataDirs(tierDataDirs); + + setLoadTsFileDirs(tierDataDirs); + for (int i = 0; i < loadTsFileDirs.length; i++) { + loadTsFileDirs[i] = addDataHomeDir(loadTsFileDirs[i]); + } } private void formulateDataDirs(String[][] tierDataDirs) { @@ -1343,6 +1347,7 @@ void reloadDataDirs(String[][] tierDataDirs) throws LoadConfigurationException { } } this.tierDataDirs = tierDataDirs; + setLoadTsFileDirs(tierDataDirs); reloadSystemMetrics(); } @@ -1464,9 +1469,7 @@ public void setSystemDir(String systemDir) { } public String[] getLoadTsFileDirs() { - return (Objects.isNull(this.loadTsFileDirs) || this.loadTsFileDirs.length == 0) - ? new String[] {tierDataDirs[0][0] + File.separator + IoTDBConstant.LOAD_TSFILE_FOLDER_NAME} - : this.loadTsFileDirs; + return this.loadTsFileDirs; } public void setLoadTsFileDirs(String[][] tierDataDirs) { From 1949e267726260d4b747f6ecc5466ef52c6c4718 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 11 Jun 2024 11:32:51 +0800 Subject: [PATCH 05/11] Update IoTDBDescriptor.java --- .../apache/iotdb/db/conf/IoTDBDescriptor.java | 57 ++++++++++--------- 1 file changed, 31 insertions(+), 26 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java index fd1068b125773..20a10200a7b8b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java @@ -895,32 +895,6 @@ public void loadProperties(Properties properties) throws BadNodeUrlException, IO conf.setIntoOperationExecutionThreadCount(2); } - conf.setMaxAllocateMemoryRatioForLoad( - Double.parseDouble( - properties.getProperty( - "max_allocate_memory_ratio_for_load", - String.valueOf(conf.getMaxAllocateMemoryRatioForLoad())))); - conf.setLoadTsFileAnalyzeSchemaBatchFlushTimeSeriesNumber( - Integer.parseInt( - properties.getProperty( - "load_tsfile_analyze_schema_batch_flush_time_series_number", - String.valueOf(conf.getLoadTsFileAnalyzeSchemaBatchFlushTimeSeriesNumber())))); - conf.setLoadTsFileAnalyzeSchemaMemorySizeInBytes( - Long.parseLong( - properties.getProperty( - "load_tsfile_analyze_schema_memory_size_in_bytes", - String.valueOf(conf.getLoadTsFileAnalyzeSchemaMemorySizeInBytes())))); - conf.setLoadCleanupTaskExecutionDelayTimeSeconds( - Long.parseLong( - properties.getProperty( - "load_clean_up_task_execution_delay_time_seconds", - String.valueOf(conf.getLoadCleanupTaskExecutionDelayTimeSeconds())))); - conf.setLoadWriteThroughputBytesPerSecond( - Double.parseDouble( - properties.getProperty( - "load_write_throughput_bytes_per_second", - String.valueOf(conf.getLoadWriteThroughputBytesPerSecond())))); - conf.setExtPipeDir(properties.getProperty("ext_pipe_dir", conf.getExtPipeDir()).trim()); // At the same time, set TSFileConfig @@ -1023,6 +997,9 @@ public void loadProperties(Properties properties) throws BadNodeUrlException, IO // CQ loadCQProps(properties); + // Load TsFile + loadLoadTsFileProps(properties); + // Pipe loadPipeProps(properties); @@ -2089,6 +2066,34 @@ private void initSchemaMemoryAllocate(Properties properties) { LOGGER.info("allocateMemoryForPartitionCache = {}", conf.getAllocateMemoryForPartitionCache()); } + private void loadLoadTsFileProps(Properties properties) { + conf.setMaxAllocateMemoryRatioForLoad( + Double.parseDouble( + properties.getProperty( + "max_allocate_memory_ratio_for_load", + String.valueOf(conf.getMaxAllocateMemoryRatioForLoad())))); + conf.setLoadTsFileAnalyzeSchemaBatchFlushTimeSeriesNumber( + Integer.parseInt( + properties.getProperty( + "load_tsfile_analyze_schema_batch_flush_time_series_number", + String.valueOf(conf.getLoadTsFileAnalyzeSchemaBatchFlushTimeSeriesNumber())))); + conf.setLoadTsFileAnalyzeSchemaMemorySizeInBytes( + Long.parseLong( + properties.getProperty( + "load_tsfile_analyze_schema_memory_size_in_bytes", + String.valueOf(conf.getLoadTsFileAnalyzeSchemaMemorySizeInBytes())))); + conf.setLoadCleanupTaskExecutionDelayTimeSeconds( + Long.parseLong( + properties.getProperty( + "load_clean_up_task_execution_delay_time_seconds", + String.valueOf(conf.getLoadCleanupTaskExecutionDelayTimeSeconds())))); + conf.setLoadWriteThroughputBytesPerSecond( + Double.parseDouble( + properties.getProperty( + "load_write_throughput_bytes_per_second", + String.valueOf(conf.getLoadWriteThroughputBytesPerSecond())))); + } + @SuppressWarnings("squid:S3518") // "proportionSum" can't be zero private void loadUDFProps(Properties properties) { String initialByteArrayLengthForMemoryControl = From 5eb6589a3261bf3425d4b00e01961cb1c94f77c5 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 11 Jun 2024 12:07:51 +0800 Subject: [PATCH 06/11] Update IoTDBConfig.java --- .../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 14 ++++---------- 1 file changed, 4 insertions(+), 10 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java index ce183f42c682a..de2ba36464bba 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java @@ -1300,11 +1300,6 @@ private void formulateFolders() { queryDir = addDataHomeDir(queryDir); sortTmpDir = addDataHomeDir(sortTmpDir); formulateDataDirs(tierDataDirs); - - setLoadTsFileDirs(tierDataDirs); - for (int i = 0; i < loadTsFileDirs.length; i++) { - loadTsFileDirs[i] = addDataHomeDir(loadTsFileDirs[i]); - } } private void formulateDataDirs(String[][] tierDataDirs) { @@ -1329,6 +1324,7 @@ private void formulateDataDirs(String[][] tierDataDirs) { } } } + formulateLoadTsFileDirs(tierDataDirs); } void reloadDataDirs(String[][] tierDataDirs) throws LoadConfigurationException { @@ -1347,7 +1343,6 @@ void reloadDataDirs(String[][] tierDataDirs) throws LoadConfigurationException { } } this.tierDataDirs = tierDataDirs; - setLoadTsFileDirs(tierDataDirs); reloadSystemMetrics(); } @@ -1433,7 +1428,6 @@ public void setTierDataDirs(String[][] tierDataDirs) { // TODO(szywilliam): rewrite the logic here when ratis supports complete snapshot semantic setRatisDataRegionSnapshotDir( tierDataDirs[0][0] + File.separator + IoTDBConstant.SNAPSHOT_FOLDER_NAME); - setLoadTsFileDirs(tierDataDirs); } public String getRpcAddress() { @@ -1472,10 +1466,10 @@ public String[] getLoadTsFileDirs() { return this.loadTsFileDirs; } - public void setLoadTsFileDirs(String[][] tierDataDirs) { - this.loadTsFileDirs = new String[tierDataDirs.length]; + public void formulateLoadTsFileDirs(String[][] tierDataDirs) { + loadTsFileDirs = new String[tierDataDirs.length]; for (int i = 0; i < tierDataDirs.length; i++) { - this.loadTsFileDirs[i] = + loadTsFileDirs[i] = tierDataDirs[i][0] + File.separator + IoTDBConstant.LOAD_TSFILE_FOLDER_NAME; } } From d26b116353d1e5315c98ef48b09a2a5cc6e33d8c Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 11 Jun 2024 13:10:07 +0800 Subject: [PATCH 07/11] Update LoadTsFileManager.java --- .../execution/load/LoadTsFileManager.java | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java index 35238b2a79667..4275522eec4e4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java @@ -94,7 +94,7 @@ public class LoadTsFileManager { new FolderManager(Arrays.asList(LOAD_BASE_DIRS), DirectoryStrategyType.SEQUENCE_STRATEGY); } catch (final DiskSpaceInsufficientException e) { LOGGER.error( - "Fail to create load tsfile folders allocation strategy because all disks of folders are full.", + "Fail to create LoadTsFileManager because of insufficient disk space, please check the disk space.", e); } } @@ -147,16 +147,13 @@ private void cleanupTasks() { private void recover() { final File[] baseDirs = Arrays.stream(LOAD_BASE_DIRS).map(File::new).toArray(File[]::new); - if (!Arrays.stream(baseDirs).allMatch(File::exists)) { - return; - } - final File[] files = Arrays.stream(baseDirs) + .filter(File::exists) .flatMap( dir -> { - File[] tmpfiles = dir.listFiles(); - return tmpfiles != null ? Arrays.stream(tmpfiles) : Stream.empty(); + final File[] listedFiles = dir.listFiles(); + return listedFiles != null ? Arrays.stream(listedFiles) : Stream.empty(); }) .toArray(File[]::new); @@ -164,8 +161,8 @@ private void recover() { .parallel() .forEach( taskDir -> { - String uuid = taskDir.getName(); - TsFileWriterManager writerManager = new TsFileWriterManager(taskDir); + final String uuid = taskDir.getName(); + final TsFileWriterManager writerManager = new TsFileWriterManager(taskDir); uuid2WriterManager.put(uuid, writerManager); writerManager.close(); From 74a017e3ca2e48dec8fe431afbef47e9e6de623f Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 11 Jun 2024 14:47:21 +0800 Subject: [PATCH 08/11] Update LoadTsFileManager.java --- .../execution/load/LoadTsFileManager.java | 21 +++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java index 4275522eec4e4..fffc47f7fdba1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java @@ -61,6 +61,7 @@ import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.PriorityBlockingQueue; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -191,6 +192,7 @@ public void writeToDataRegion(DataRegion dataRegion, LoadTsFilePieceNode pieceNo final Optional cleanupTask = Optional.of(uuid2CleanupTask.get(uuid)); cleanupTask.ifPresent(CleanupTask::markLoadTaskRunning); try { + final AtomicReference exception = new AtomicReference<>(); final TsFileWriterManager writerManager = uuid2WriterManager.computeIfAbsent( uuid, @@ -198,9 +200,18 @@ public void writeToDataRegion(DataRegion dataRegion, LoadTsFilePieceNode pieceNo try { return new TsFileWriterManager(new File(folderManager.getNextFolder(), uuid)); } catch (DiskSpaceInsufficientException e) { - throw new RuntimeException(e); + exception.set(e); + return null; } }); + if (exception.get() != null || writerManager == null) { + throw new IOException( + "Failed to create TsFileWriterManager for uuid " + + uuid + + " because of insufficient disk space.", + exception.get()); + } + for (TsFileData tsFileData : pieceNode.getAllTsFileData()) { if (!tsFileData.isModification()) { ChunkData chunkData = (ChunkData) tsFileData; @@ -258,20 +269,18 @@ private void forceCloseWriterManager(String uuid) { writerManager.close(); } - for (Path loadDirPath : + for (final Path loadDirPath : Arrays.stream(LOAD_BASE_DIRS) .map(File::new) + .filter(File::exists) .map(File::toPath) .collect(Collectors.toList())) { - if (!Files.exists(loadDirPath)) { - continue; - } try { Files.deleteIfExists(loadDirPath); LOGGER.info("Load dir {} was deleted.", loadDirPath); } catch (DirectoryNotEmptyException e) { LOGGER.info("Load dir {} is not empty, skip deleting.", loadDirPath); - } catch (IOException e) { + } catch (Exception e) { LOGGER.info(MESSAGE_DELETE_FAIL, loadDirPath); } } From b6062b90e759fec93fd0a9cd2af420aa30145782 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 11 Jun 2024 14:49:29 +0800 Subject: [PATCH 09/11] Update LoadTsFileManager.java --- .../iotdb/db/queryengine/execution/load/LoadTsFileManager.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java index fffc47f7fdba1..23aed91dcc813 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java @@ -80,8 +80,8 @@ public class LoadTsFileManager { private static final String MESSAGE_WRITER_MANAGER_HAS_BEEN_CLOSED = "%s TsFileWriterManager has been closed."; private static final String MESSAGE_DELETE_FAIL = "failed to delete {}."; - private static final String[] LOAD_BASE_DIRS = CONFIG.getLoadTsFileDirs(); + private static final String[] LOAD_BASE_DIRS = CONFIG.getLoadTsFileDirs(); private static FolderManager folderManager; private final Map uuid2WriterManager = new ConcurrentHashMap<>(); From 31fd87f31ec65bef24717ada4199cc59c5e5423c Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 11 Jun 2024 18:21:35 +0800 Subject: [PATCH 10/11] refactor --- .../execution/load/LoadTsFileManager.java | 57 +++++++++++++------ .../rescon/disk/FolderManager.java | 4 ++ 2 files changed, 45 insertions(+), 16 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java index 23aed91dcc813..c5ecd6b566d80 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java @@ -81,25 +81,15 @@ public class LoadTsFileManager { "%s TsFileWriterManager has been closed."; private static final String MESSAGE_DELETE_FAIL = "failed to delete {}."; - private static final String[] LOAD_BASE_DIRS = CONFIG.getLoadTsFileDirs(); - private static FolderManager folderManager; + private static final AtomicReference LOAD_BASE_DIRS = + new AtomicReference<>(CONFIG.getLoadTsFileDirs()); + private static final AtomicReference FOLDER_MANAGER = new AtomicReference<>(); private final Map uuid2WriterManager = new ConcurrentHashMap<>(); private final Map uuid2CleanupTask = new ConcurrentHashMap<>(); private final PriorityBlockingQueue cleanupTaskQueue = new PriorityBlockingQueue<>(); - static { - try { - folderManager = - new FolderManager(Arrays.asList(LOAD_BASE_DIRS), DirectoryStrategyType.SEQUENCE_STRATEGY); - } catch (final DiskSpaceInsufficientException e) { - LOGGER.error( - "Fail to create LoadTsFileManager because of insufficient disk space, please check the disk space.", - e); - } - } - public LoadTsFileManager() { registerCleanupTaskExecutor(); recover(); @@ -147,7 +137,15 @@ private void cleanupTasks() { } private void recover() { - final File[] baseDirs = Arrays.stream(LOAD_BASE_DIRS).map(File::new).toArray(File[]::new); + if (CONFIG.getLoadTsFileDirs() != LOAD_BASE_DIRS.get()) { + synchronized (LOAD_BASE_DIRS) { + if (CONFIG.getLoadTsFileDirs() != LOAD_BASE_DIRS.get()) { + LOAD_BASE_DIRS.set(CONFIG.getLoadTsFileDirs()); + } + } + } + + final File[] baseDirs = Arrays.stream(LOAD_BASE_DIRS.get()).map(File::new).toArray(File[]::new); final File[] files = Arrays.stream(baseDirs) .filter(File::exists) @@ -198,7 +196,7 @@ public void writeToDataRegion(DataRegion dataRegion, LoadTsFilePieceNode pieceNo uuid, o -> { try { - return new TsFileWriterManager(new File(folderManager.getNextFolder(), uuid)); + return new TsFileWriterManager(new File(getNextFolder(), uuid)); } catch (DiskSpaceInsufficientException e) { exception.set(e); return null; @@ -226,6 +224,33 @@ public void writeToDataRegion(DataRegion dataRegion, LoadTsFilePieceNode pieceNo } } + private String getNextFolder() throws DiskSpaceInsufficientException { + if (CONFIG.getLoadTsFileDirs() != LOAD_BASE_DIRS.get()) { + synchronized (FOLDER_MANAGER) { + if (CONFIG.getLoadTsFileDirs() != LOAD_BASE_DIRS.get()) { + LOAD_BASE_DIRS.set(CONFIG.getLoadTsFileDirs()); + FOLDER_MANAGER.set( + new FolderManager( + Arrays.asList(LOAD_BASE_DIRS.get()), DirectoryStrategyType.SEQUENCE_STRATEGY)); + return FOLDER_MANAGER.get().getNextFolder(); + } + } + } + + if (FOLDER_MANAGER.get() == null) { + synchronized (FOLDER_MANAGER) { + if (FOLDER_MANAGER.get() == null) { + FOLDER_MANAGER.set( + new FolderManager( + Arrays.asList(LOAD_BASE_DIRS.get()), DirectoryStrategyType.SEQUENCE_STRATEGY)); + return FOLDER_MANAGER.get().getNextFolder(); + } + } + } + + return FOLDER_MANAGER.get().getNextFolder(); + } + public boolean loadAll(String uuid, boolean isGeneratedByPipe, ProgressIndex progressIndex) throws IOException, LoadFileException { if (!uuid2WriterManager.containsKey(uuid)) { @@ -270,7 +295,7 @@ private void forceCloseWriterManager(String uuid) { } for (final Path loadDirPath : - Arrays.stream(LOAD_BASE_DIRS) + Arrays.stream(LOAD_BASE_DIRS.get()) .map(File::new) .filter(File::exists) .map(File::toPath) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/FolderManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/FolderManager.java index b32307ada8bae..0851a698aba12 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/FolderManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/FolderManager.java @@ -79,4 +79,8 @@ public String getNextFolder() throws DiskSpaceInsufficientException { throw e; } } + + public List getFolders() { + return folders; + } } From a3ad36dfb66bef4d07dcbc94273675a19bec1adc Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 11 Jun 2024 18:47:44 +0800 Subject: [PATCH 11/11] Update LoadTsFileManager.java --- .../iotdb/db/queryengine/execution/load/LoadTsFileManager.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java index c5ecd6b566d80..054c664b48efe 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java @@ -138,7 +138,7 @@ private void cleanupTasks() { private void recover() { if (CONFIG.getLoadTsFileDirs() != LOAD_BASE_DIRS.get()) { - synchronized (LOAD_BASE_DIRS) { + synchronized (FOLDER_MANAGER) { if (CONFIG.getLoadTsFileDirs() != LOAD_BASE_DIRS.get()) { LOAD_BASE_DIRS.set(CONFIG.getLoadTsFileDirs()); }