You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2024/11/13 10:48:23 UTC
(iotdb) 01/01: Fix flush command stack problem and clean old code in DataRegion (#14027)
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch fix_flush_stack_13
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 1a0e4c57a3eef790041a2809ff785c17287791e0
Author: Haonan <hh...@outlook.com>
AuthorDate: Wed Nov 13 17:42:24 2024 +0800
Fix flush command stack problem and clean old code in DataRegion (#14027)
* Try to fix flush stack problem and clean old code in DataRegion
* clean code
* Fix flush command stuck
* fix ArrayIndexOutOfBoundsException
* refactor flush manager
* fix dataregion UT
---
.../iotdb/db/storageengine/StorageEngine.java | 106 +++----
.../db/storageengine/dataregion/DataRegion.java | 350 +++++----------------
.../dataregion/flush/FlushManager.java | 50 +--
.../dataregion/flush/TsFileFlushPolicy.java | 6 +-
.../dataregion/memtable/TsFileProcessor.java | 119 +++----
.../generator/SimpleFileVersionController.java | 160 ----------
.../tsfile/generator/VersionController.java | 44 ---
.../tsfile/timeindex/DeviceTimeIndex.java | 2 +-
.../storageengine/dataregion/DataRegionTest.java | 55 +---
.../generator/SimpleFileVersionControllerTest.java | 56 ----
10 files changed, 204 insertions(+), 744 deletions(-)
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 866fcfc28b1..0baf58a172d 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
@@ -45,7 +45,6 @@ import org.apache.iotdb.commons.utils.TimePartitionUtils;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.consensus.statemachine.dataregion.DataExecutionVisitor;
import org.apache.iotdb.db.exception.DataRegionException;
import org.apache.iotdb.db.exception.LoadReadOnlyException;
import org.apache.iotdb.db.exception.StorageEngineException;
@@ -53,7 +52,6 @@ import org.apache.iotdb.db.exception.TsFileProcessorException;
import org.apache.iotdb.db.exception.WriteProcessRejectException;
import org.apache.iotdb.db.exception.runtime.StorageEngineFailureException;
import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeTTLCache;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode;
import org.apache.iotdb.db.queryengine.plan.scheduler.load.LoadTsFileScheduler;
import org.apache.iotdb.db.service.metrics.WritingMetrics;
@@ -68,7 +66,6 @@ import org.apache.iotdb.db.storageengine.dataregion.flush.CloseFileListener;
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;
-import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor;
import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager;
import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALException;
import org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRecoverManager;
@@ -93,6 +90,7 @@ import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Properties;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
@@ -234,7 +232,7 @@ public class StorageEngine implements IService {
for (DataRegionId dataRegionId : entry.getValue()) {
Callable<Void> recoverDataRegionTask =
() -> {
- DataRegion dataRegion = null;
+ DataRegion dataRegion;
try {
dataRegion = buildNewDataRegion(sgName, dataRegionId);
} catch (DataRegionException e) {
@@ -268,7 +266,7 @@ public class StorageEngine implements IService {
}
String sgName = sgDir.getName();
List<DataRegionId> dataRegionIdList = new ArrayList<>();
- for (File dataRegionDir : sgDir.listFiles()) {
+ for (File dataRegionDir : Objects.requireNonNull(sgDir.listFiles())) {
if (!dataRegionDir.isDirectory()) {
continue;
}
@@ -446,21 +444,21 @@ public class StorageEngine implements IService {
* build a new data region
*
* @param dataRegionId data region id e.g. 1
- * @param logicalStorageGroupName database name e.g. root.sg1
+ * @param databaseName database name e.g. root.sg1
*/
- public DataRegion buildNewDataRegion(String logicalStorageGroupName, DataRegionId dataRegionId)
+ public DataRegion buildNewDataRegion(String databaseName, DataRegionId dataRegionId)
throws DataRegionException {
DataRegion dataRegion;
LOGGER.info(
"construct a data region instance, the database is {}, Thread is {}",
- logicalStorageGroupName,
+ databaseName,
Thread.currentThread().getId());
dataRegion =
new DataRegion(
- systemDir + File.separator + logicalStorageGroupName,
+ systemDir + File.separator + databaseName,
String.valueOf(dataRegionId.getId()),
fileFlushPolicy,
- logicalStorageGroupName);
+ databaseName);
WRITING_METRICS.createFlushingMemTableStatusMetrics(dataRegionId);
WRITING_METRICS.createDataRegionMemoryCostMetrics(dataRegion);
WRITING_METRICS.createActiveMemtableCounterMetrics(dataRegionId);
@@ -469,11 +467,6 @@ public class StorageEngine implements IService {
return dataRegion;
}
- /** Write data into DataRegion. For standalone mode only. */
- public TSStatus write(DataRegionId groupId, PlanNode planNode) {
- return planNode.accept(new DataExecutionVisitor(), dataRegionMap.get(groupId));
- }
-
/** This function is just for unit test. */
@TestOnly
public synchronized void reset() {
@@ -513,29 +506,31 @@ public class StorageEngine implements IService {
checkResults(tasks, "Failed to force close processor.");
}
- public void closeStorageGroupProcessor(String storageGroupPath, boolean isSeq) {
+ public void syncCloseProcessorsInDatabase(String databaseName) {
List<Future<Void>> tasks = new ArrayList<>();
for (DataRegion dataRegion : dataRegionMap.values()) {
- if (dataRegion.getDatabaseName().equals(storageGroupPath)) {
- if (isSeq) {
- for (TsFileProcessor tsFileProcessor : dataRegion.getWorkSequenceTsFileProcessors()) {
- tasks.add(
- cachedThreadPool.submit(
- () -> {
- dataRegion.syncCloseOneTsFileProcessor(isSeq, tsFileProcessor);
- return null;
- }));
- }
- } else {
- for (TsFileProcessor tsFileProcessor : dataRegion.getWorkUnsequenceTsFileProcessors()) {
- tasks.add(
- cachedThreadPool.submit(
- () -> {
- dataRegion.syncCloseOneTsFileProcessor(isSeq, tsFileProcessor);
- return null;
- }));
- }
- }
+ if (dataRegion != null && dataRegion.getDatabaseName().equals(databaseName)) {
+ tasks.add(
+ cachedThreadPool.submit(
+ () -> {
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
+ return null;
+ }));
+ }
+ }
+ checkResults(tasks, "Failed to sync close processor.");
+ }
+
+ public void syncCloseProcessorsInDatabase(String databaseName, boolean isSeq) {
+ List<Future<Void>> tasks = new ArrayList<>();
+ for (DataRegion dataRegion : dataRegionMap.values()) {
+ if (dataRegion.getDatabaseName().equals(databaseName)) {
+ tasks.add(
+ cachedThreadPool.submit(
+ () -> {
+ dataRegion.syncCloseWorkingTsFileProcessors(isSeq);
+ return null;
+ }));
}
}
checkResults(tasks, "Failed to close database processor.");
@@ -638,13 +633,12 @@ public class StorageEngine implements IService {
StorageEngine.getInstance().syncCloseAllProcessor();
WALManager.getInstance().syncDeleteOutdatedFilesInWALNodes();
} else {
- for (String storageGroup : req.storageGroups) {
+ for (String databaseName : req.storageGroups) {
if (req.isSeq == null) {
- StorageEngine.getInstance().closeStorageGroupProcessor(storageGroup, true);
- StorageEngine.getInstance().closeStorageGroupProcessor(storageGroup, false);
+ StorageEngine.getInstance().syncCloseProcessorsInDatabase(databaseName);
} else {
StorageEngine.getInstance()
- .closeStorageGroupProcessor(storageGroup, Boolean.parseBoolean(req.isSeq));
+ .syncCloseProcessorsInDatabase(databaseName, Boolean.parseBoolean(req.isSeq));
}
}
}
@@ -698,7 +692,7 @@ public class StorageEngine implements IService {
/**
* Add a listener to listen flush start/end events. Notice that this addition only applies to
- * TsFileProcessors created afterwards.
+ * TsFileProcessors created afterward.
*/
public void registerFlushListener(FlushListener listener) {
customFlushListeners.add(listener);
@@ -706,7 +700,7 @@ public class StorageEngine implements IService {
/**
* Add a listener to listen file close events. Notice that this addition only applies to
- * TsFileProcessors created afterwards.
+ * TsFileProcessors created afterward.
*/
public void registerCloseFileListener(CloseFileListener listener) {
customCloseFileListeners.add(listener);
@@ -723,24 +717,24 @@ public class StorageEngine implements IService {
// When registering a new region, the coordinator needs to register the corresponding region with
// the local storage before adding the corresponding consensusGroup to the consensus layer
- public DataRegion createDataRegion(DataRegionId regionId, String sg) throws DataRegionException {
+ public void createDataRegion(DataRegionId regionId, String databaseName)
+ throws DataRegionException {
makeSureNoOldRegion(regionId);
AtomicReference<DataRegionException> exceptionAtomicReference = new AtomicReference<>(null);
- DataRegion dataRegion =
- dataRegionMap.computeIfAbsent(
- regionId,
- x -> {
- try {
- return buildNewDataRegion(sg, x);
- } catch (DataRegionException e) {
- exceptionAtomicReference.set(e);
- }
- return null;
- });
+ dataRegionMap.computeIfAbsent(
+ regionId,
+ region -> {
+ try {
+ return buildNewDataRegion(databaseName, region);
+ } catch (DataRegionException e) {
+ exceptionAtomicReference.set(e);
+ }
+ return null;
+ });
+
if (exceptionAtomicReference.get() != null) {
throw exceptionAtomicReference.get();
}
- return dataRegion;
}
public void deleteDataRegion(DataRegionId regionId) {
@@ -960,7 +954,7 @@ public class StorageEngine implements IService {
return status;
}
- /** reboot timed flush sequence/unsequence memetable thread */
+ /** reboot timed flush sequence/unsequence memtable thread */
public void rebootTimedService() throws ShutdownException {
LOGGER.info("Start rebooting all timed service.");
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index 15fca755487..a07aeafc6a4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -95,7 +95,6 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager;
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;
-import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.VersionController;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndex;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndexCacheRecorder;
import org.apache.iotdb.db.storageengine.dataregion.utils.fileTimeIndexCache.FileTimeIndexCacheReader;
@@ -258,14 +257,6 @@ public class DataRegion implements IDataRegionForQuery {
/** manage tsFileResource degrade. */
private final TsFileResourceManager tsFileResourceManager = TsFileResourceManager.getInstance();
- /**
- * time partition id -> version controller which assigns a version for each MemTable and
- * deletion/update such that after they are persisted, the order of insertions, deletions and
- * updates can be re-determined. Will be empty if there are not MemTables in memory.
- */
- private final HashMap<Long, VersionController> timePartitionIdVersionControllerMap =
- new HashMap<>();
-
/** file system factory (local or hdfs). */
private final FSFactory fsFactory = FSFactoryProducer.getFSFactory();
@@ -359,7 +350,12 @@ public class DataRegion implements IDataRegionForQuery {
e);
}
if (FSUtils.getFSType(dataRegionFolder) == FSType.LOCAL) {
- dataRegionFolder.mkdirs();
+ if (dataRegionFolder.mkdirs()) {
+ logger.info(
+ "Data region directory {} doesn't exist, create it", dataRegionFolder.getPath());
+ } else if (!dataRegionFolder.exists()) {
+ logger.error("create data region directory {} failed", dataRegionFolder.getPath());
+ }
}
}
} else {
@@ -430,7 +426,7 @@ public class DataRegion implements IDataRegionForQuery {
}
} else {
logger.info(
- "The TsFiles of data region {}[{}] has recovered" + " {}/{}.",
+ "The TsFiles of data region {}[{}] has recovered completely {}/{}.",
databaseName,
dataRegionId,
numOfFilesToRecover,
@@ -848,8 +844,8 @@ public class DataRegion implements IDataRegionForQuery {
// get chunkMetadata size
long chunkMetadataSize = 0;
for (Map<String, List<ChunkMetadata>> metaMap : writer.getMetadatasForQuery().values()) {
- for (List<ChunkMetadata> metadatas : metaMap.values()) {
- for (ChunkMetadata chunkMetadata : metadatas) {
+ for (List<ChunkMetadata> metadataList : metaMap.values()) {
+ for (ChunkMetadata chunkMetadata : metadataList) {
chunkMetadataSize += chunkMetadata.getRetainedSizeInBytes();
}
}
@@ -867,7 +863,7 @@ public class DataRegion implements IDataRegionForQuery {
/** recover sealed TsFile. */
private void recoverSealedTsFiles(
- TsFileResource sealedTsFile, DataRegionRecoveryContext context, boolean isSeq) {
+ TsFileResource sealedTsFile, DataRegionRecoveryContext context) {
try (SealedTsFileRecoverPerformer recoverPerformer =
new SealedTsFileRecoverPerformer(sealedTsFile)) {
recoverPerformer.recover();
@@ -902,19 +898,16 @@ public class DataRegion implements IDataRegionForQuery {
}
if (!resourceListForAsyncRecover.isEmpty()) {
asyncRecoverTask =
- asyncRecoverFilesInPartition(partitionId, context, resourceListForAsyncRecover, isSeq);
+ asyncRecoverFilesInPartition(partitionId, context, resourceListForAsyncRecover);
}
if (!resourceListForSyncRecover.isEmpty()) {
- syncRecoverFilesInPartition(partitionId, context, resourceListForSyncRecover, isSeq);
+ syncRecoverFilesInPartition(partitionId, context, resourceListForSyncRecover);
}
return asyncRecoverTask;
}
private Callable<Void> asyncRecoverFilesInPartition(
- long partitionId,
- DataRegionRecoveryContext context,
- List<TsFileResource> resourceList,
- boolean isSeq) {
+ long partitionId, DataRegionRecoveryContext context, List<TsFileResource> resourceList) {
if (config.isEnableSeparateData()) {
if (!lastFlushTimeMap.checkAndCreateFlushedTimePartition(partitionId, false)) {
TimePartitionManager.getInstance()
@@ -961,12 +954,9 @@ public class DataRegion implements IDataRegionForQuery {
}
private void syncRecoverFilesInPartition(
- long partitionId,
- DataRegionRecoveryContext context,
- List<TsFileResource> resourceList,
- boolean isSeq) {
+ long partitionId, DataRegionRecoveryContext context, List<TsFileResource> resourceList) {
for (TsFileResource tsFileResource : resourceList) {
- recoverSealedTsFiles(tsFileResource, context, isSeq);
+ recoverSealedTsFiles(tsFileResource, context);
}
FileTimeIndexCacheRecorder.getInstance()
.logFileTimeIndex(resourceList.toArray(new TsFileResource[0]));
@@ -1530,7 +1520,7 @@ public class DataRegion implements IDataRegionForQuery {
// build new processor, memory control module will control the number of memtables
TimePartitionManager.getInstance()
.updateAfterOpeningTsFileProcessor(
- new DataRegionId(Integer.valueOf(dataRegionId)), timeRangeId);
+ new DataRegionId(Integer.parseInt(dataRegionId)), timeRangeId);
res = newTsFileProcessor(sequence, timeRangeId);
if (workSequenceTsFileProcessors.get(timeRangeId) == null
&& workUnsequenceTsFileProcessors.get(timeRangeId) == null) {
@@ -1584,55 +1574,10 @@ public class DataRegion implements IDataRegionForQuery {
return tsFileProcessor;
}
- /**
- * Create a new tsfile name
- *
- * @return file name
- */
- private String getNewTsFileName(long timePartitionId) {
- long version =
- partitionMaxFileVersions.compute(
- timePartitionId, (key, oldVersion) -> (oldVersion == null ? 1 : oldVersion + 1));
- return getNewTsFileName(System.currentTimeMillis(), version, 0, 0);
- }
-
private String getNewTsFileName(long time, long version, int mergeCnt, int unseqCompactionCnt) {
return TsFileNameGenerator.generateNewTsFileName(time, version, mergeCnt, unseqCompactionCnt);
}
- /**
- * close one tsfile processor
- *
- * @param sequence whether this tsfile processor is sequence or not
- * @param tsFileProcessor tsfile processor
- */
- public void syncCloseOneTsFileProcessor(boolean sequence, TsFileProcessor tsFileProcessor) {
- synchronized (closeStorageGroupCondition) {
- try {
- asyncCloseOneTsFileProcessor(sequence, tsFileProcessor);
- long startTime = System.currentTimeMillis();
- while (closingSequenceTsFileProcessor.contains(tsFileProcessor)
- || closingUnSequenceTsFileProcessor.contains(tsFileProcessor)) {
- closeStorageGroupCondition.wait(60_000);
- if (System.currentTimeMillis() - startTime > 60_000) {
- logger.warn(
- "{} has spent {}s to wait for closing one tsfile.",
- databaseName + "-" + this.dataRegionId,
- (System.currentTimeMillis() - startTime) / 1000);
- }
- }
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- logger.error(
- "syncCloseOneTsFileProcessor error occurs while waiting for closing the storage "
- + "group {}",
- databaseName + "-" + dataRegionId,
- e);
- }
- }
- WritingMetrics.getInstance().recordManualFlushMemTableCount(1);
- }
-
/**
* close one tsfile processor, thread-safety should be ensured by caller
*
@@ -1647,29 +1592,25 @@ public class DataRegion implements IDataRegionForQuery {
|| tsFileProcessor.alreadyMarkedClosing()) {
return CompletableFuture.completedFuture(null);
}
- TsFileResource resource = tsFileProcessor.getTsFileResource();
Future<?> future;
if (sequence) {
closingSequenceTsFileProcessor.add(tsFileProcessor);
future = tsFileProcessor.asyncClose();
+ if (future.isDone()) {
+ closingSequenceTsFileProcessor.remove(tsFileProcessor);
+ }
workSequenceTsFileProcessors.remove(tsFileProcessor.getTimeRangeId());
- // if unsequence files don't contain this time range id, we should remove it's version
- // controller
- if (!workUnsequenceTsFileProcessors.containsKey(tsFileProcessor.getTimeRangeId())) {
- timePartitionIdVersionControllerMap.remove(tsFileProcessor.getTimeRangeId());
- }
} else {
closingUnSequenceTsFileProcessor.add(tsFileProcessor);
future = tsFileProcessor.asyncClose();
+ if (future.isDone()) {
+ closingUnSequenceTsFileProcessor.remove(tsFileProcessor);
+ }
workUnsequenceTsFileProcessors.remove(tsFileProcessor.getTimeRangeId());
- // if sequence files don't contain this time range id, we should remove it's version
- // controller
- if (!workSequenceTsFileProcessors.containsKey(tsFileProcessor.getTimeRangeId())) {
- timePartitionIdVersionControllerMap.remove(tsFileProcessor.getTimeRangeId());
- }
}
+ TsFileResource resource = tsFileProcessor.getTsFileResource();
logger.info(
"Async close tsfile: {}, file start time: {}, file end time: {}",
resource.getTsFile().getAbsolutePath(),
@@ -1838,7 +1779,6 @@ public class DataRegion implements IDataRegionForQuery {
public void syncCloseAllWorkingTsFileProcessors() {
try {
List<Future<?>> tsFileProcessorsClosingFutures = asyncCloseAllWorkingTsFileProcessors();
- waitClosingTsFileProcessorFinished();
for (Future<?> f : tsFileProcessorsClosingFutures) {
if (f != null) {
f.get();
@@ -1846,7 +1786,41 @@ public class DataRegion implements IDataRegionForQuery {
}
} catch (InterruptedException | ExecutionException e) {
logger.error(
- "CloseFileNodeCondition error occurs while waiting for closing the storage " + "group {}",
+ "CloseFileNodeCondition error occurs while waiting for closing tsfile processors of {}",
+ databaseName + "-" + dataRegionId,
+ e);
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ public void syncCloseWorkingTsFileProcessors(boolean sequence) {
+ try {
+ writeLock("syncCloseWorkingTsFileProcessors");
+ List<Future<?>> tsFileProcessorsClosingFutures = new ArrayList<>();
+ int count = 0;
+ try {
+ // to avoid concurrent modification problem, we need a new array list
+ for (TsFileProcessor tsFileProcessor :
+ new ArrayList<>(
+ sequence
+ ? workSequenceTsFileProcessors.values()
+ : workUnsequenceTsFileProcessors.values())) {
+ tsFileProcessorsClosingFutures.add(
+ asyncCloseOneTsFileProcessor(sequence, tsFileProcessor));
+ count++;
+ }
+ } finally {
+ writeUnlock();
+ }
+ WritingMetrics.getInstance().recordManualFlushMemTableCount(count);
+ for (Future<?> f : tsFileProcessorsClosingFutures) {
+ if (f != null) {
+ f.get();
+ }
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ logger.error(
+ "CloseFileNodeCondition error occurs while waiting for closing tsfile processors of {}",
databaseName + "-" + dataRegionId,
e);
Thread.currentThread().interrupt();
@@ -1874,7 +1848,7 @@ public class DataRegion implements IDataRegionForQuery {
}
/** close all working tsfile processors */
- List<Future<?>> asyncCloseAllWorkingTsFileProcessors() {
+ private List<Future<?>> asyncCloseAllWorkingTsFileProcessors() {
writeLock("asyncCloseAllWorkingTsFileProcessors");
List<Future<?>> futures = new ArrayList<>();
int count = 0;
@@ -1936,7 +1910,7 @@ public class DataRegion implements IDataRegionForQuery {
}
}
- /** used for queryengine */
+ /** used for query engine */
@Override
public QueryDataSource query(
List<PartialPath> pathList,
@@ -1999,9 +1973,7 @@ public class DataRegion implements IDataRegionForQuery {
QUERY_RESOURCE_METRIC_SET.recordQueryResourceNum(
UNSEQUENCE_TSFILE, unseqFileScanHandles.size());
- QueryDataSourceForRegionScan dataSource =
- new QueryDataSourceForRegionScan(seqFileScanHandles, unseqFileScanHandles);
- return dataSource;
+ return new QueryDataSourceForRegionScan(seqFileScanHandles, unseqFileScanHandles);
} catch (MetadataException e) {
throw new QueryProcessException(e);
}
@@ -2063,9 +2035,7 @@ public class DataRegion implements IDataRegionForQuery {
QUERY_RESOURCE_METRIC_SET.recordQueryResourceNum(
UNSEQUENCE_TSFILE, unseqFileScanHandles.size());
- QueryDataSourceForRegionScan dataSource =
- new QueryDataSourceForRegionScan(seqFileScanHandles, unseqFileScanHandles);
- return dataSource;
+ return new QueryDataSourceForRegionScan(seqFileScanHandles, unseqFileScanHandles);
} catch (MetadataException e) {
throw new QueryProcessException(e);
}
@@ -2143,7 +2113,7 @@ public class DataRegion implements IDataRegionForQuery {
if (context.isDebug()) {
DEBUG_LOGGER.info(
- "Path: {}, get tsfile list: {} isSeq: {} timefilter: {}",
+ "Path: {}, get tsfile list: {} isSeq: {} time filter: {}",
pathList,
tsFileResources,
isSeq,
@@ -2176,7 +2146,7 @@ public class DataRegion implements IDataRegionForQuery {
return tsfileResourcesForQuery;
}
- /** Seperate tsfiles in TsFileManager to sealedList and unsealedList. */
+ /** Separate tsfiles in TsFileManager to sealedList and unsealedList. */
private void getTwoKindsOfTsFiles(
List<TsFileResource> sealedResource,
List<TsFileResource> unsealedResource,
@@ -2307,7 +2277,8 @@ public class DataRegion implements IDataRegionForQuery {
}
}
// Some time the deletion operation doesn't have any related tsfile processor or memtable,
- // but it's still necessary to write to the WAL, so that iotconsensus can synchronize the delete
+ // but it's still necessary to write to the WAL, so that iot consensus can synchronize the
+ // delete
// operation to other nodes.
if (walFlushListeners.isEmpty()) {
// TODO: IoTConsensusV2 deletion support
@@ -2807,7 +2778,7 @@ public class DataRegion implements IDataRegionForQuery {
oldTsFileResource.writeLock();
try {
TsFileAndModSettleTool.moveNewTsFile(oldTsFileResource, newTsFileResources);
- if (TsFileAndModSettleTool.getInstance().recoverSettleFileMap.size() != 0) {
+ if (!TsFileAndModSettleTool.getInstance().recoverSettleFileMap.isEmpty()) {
TsFileAndModSettleTool.getInstance()
.recoverSettleFileMap
.remove(oldTsFileResource.getTsFile().getAbsolutePath());
@@ -2949,101 +2920,6 @@ public class DataRegion implements IDataRegionForQuery {
}
}
- /**
- * Set the version in "partition" to "version" if "version" is larger than the current version.
- */
- public void setPartitionFileVersionToMax(long partition, long version) {
- partitionMaxFileVersions.compute(
- partition, (prt, oldVer) -> computeMaxVersion(oldVer, version));
- }
-
- private long computeMaxVersion(Long oldVersion, Long newVersion) {
- if (oldVersion == null) {
- return newVersion;
- }
- return Math.max(oldVersion, newVersion);
- }
-
- /**
- * If the historical versions of a file is a sub-set of the given file's, (close and) remove it to
- * reduce unnecessary merge. Only used when the file sender and the receiver share the same file
- * close policy. Warning: DO NOT REMOVE
- */
- @SuppressWarnings("unused")
- public void removeFullyOverlapFiles(TsFileResource resource) {
- writeLock("removeFullyOverlapFiles");
- try {
- Iterator<TsFileResource> iterator = tsFileManager.getIterator(true);
- removeFullyOverlapFiles(resource, iterator, true);
-
- iterator = tsFileManager.getIterator(false);
- removeFullyOverlapFiles(resource, iterator, false);
- } finally {
- writeUnlock();
- }
- }
-
- private void removeFullyOverlapFiles(
- TsFileResource newTsFile, Iterator<TsFileResource> iterator, boolean isSeq) {
- while (iterator.hasNext()) {
- TsFileResource existingTsFile = iterator.next();
- if (newTsFile.isPlanRangeCovers(existingTsFile)
- && !newTsFile.getTsFile().equals(existingTsFile.getTsFile())
- && existingTsFile.tryWriteLock()) {
- logger.info(
- "{} is covered by {}: [{}, {}], [{}, {}], remove it",
- existingTsFile,
- newTsFile,
- existingTsFile.minPlanIndex,
- existingTsFile.maxPlanIndex,
- newTsFile.minPlanIndex,
- newTsFile.maxPlanIndex);
- // if we fail to lock the file, it means it is being queried or merged and we will not
- // wait until it is free, we will just leave it to the next merge
- try {
- removeFullyOverlapFile(existingTsFile, iterator, isSeq);
- } catch (Exception e) {
- logger.error(
- "Something gets wrong while removing FullyOverlapFiles: {}",
- existingTsFile.getTsFile().getAbsolutePath(),
- e);
- } finally {
- existingTsFile.writeUnlock();
- }
- }
- }
- }
-
- /**
- * Remove the given {@link TsFileResource}. If the corresponding {@link TsFileProcessor} is in the
- * working status, close it before remove the related resource files. maybe time-consuming for
- * closing a tsfile.
- */
- private void removeFullyOverlapFile(
- TsFileResource tsFileResource, Iterator<TsFileResource> iterator, boolean isSeq) {
- logger.info(
- "Removing a covered file {}, closed: {}", tsFileResource, tsFileResource.isClosed());
- if (!tsFileResource.isClosed()) {
- try {
- // also remove the TsFileProcessor if the overlapped file is not closed
- long timePartition = tsFileResource.getTimePartition();
- Map<Long, TsFileProcessor> fileProcessorMap =
- isSeq ? workSequenceTsFileProcessors : workUnsequenceTsFileProcessors;
- TsFileProcessor tsFileProcessor = fileProcessorMap.get(timePartition);
- if (tsFileProcessor != null && tsFileProcessor.getTsFileResource() == tsFileResource) {
- // have to take some time to close the tsFileProcessor
- tsFileProcessor.syncClose();
- fileProcessorMap.remove(timePartition);
- }
- } catch (Exception e) {
- logger.error("Cannot close {}", tsFileResource, e);
- }
- }
- tsFileManager.remove(tsFileResource, isSeq);
- iterator.remove();
- tsFileResource.remove();
- }
-
private long getAndSetNewVersion(long timePartitionId, TsFileResource tsFileResource) {
long version =
partitionMaxFileVersions.compute(
@@ -3300,67 +3176,6 @@ public class DataRegion implements IDataRegionForQuery {
return databaseName + File.separator + dataRegionId;
}
- /**
- * Check if the data of "tsFileResource" all exist locally by comparing planIndexes in the
- * partition of "partitionNumber". This is available only when the IoTDB instances which generated
- * "tsFileResource" have the same plan indexes as the local one.
- *
- * @return true if any file contains plans with indexes no less than the max plan index of
- * "tsFileResource", otherwise false.
- */
- public boolean isFileAlreadyExist(TsFileResource tsFileResource, long partitionNum) {
- // examine working processor first as they have the largest plan index
- return isFileAlreadyExistInWorking(
- tsFileResource, partitionNum, getWorkSequenceTsFileProcessors())
- || isFileAlreadyExistInWorking(
- tsFileResource, partitionNum, getWorkUnsequenceTsFileProcessors())
- || isFileAlreadyExistInClosed(tsFileResource, partitionNum, getSequenceFileList())
- || isFileAlreadyExistInClosed(tsFileResource, partitionNum, getUnSequenceFileList());
- }
-
- private boolean isFileAlreadyExistInClosed(
- TsFileResource tsFileResource, long partitionNum, Collection<TsFileResource> existingFiles) {
- for (TsFileResource resource : existingFiles) {
- if (resource.getTimePartition() == partitionNum
- && resource.getMaxPlanIndex() > tsFileResource.getMaxPlanIndex()) {
- logger.info(
- "{} is covered by a closed file {}: [{}, {}] [{}, {}]",
- tsFileResource,
- resource,
- tsFileResource.minPlanIndex,
- tsFileResource.maxPlanIndex,
- resource.minPlanIndex,
- resource.maxPlanIndex);
- return true;
- }
- }
- return false;
- }
-
- private boolean isFileAlreadyExistInWorking(
- TsFileResource tsFileResource,
- long partitionNum,
- Collection<TsFileProcessor> workingProcessors) {
- for (TsFileProcessor workingProcesssor : workingProcessors) {
- if (workingProcesssor.getTimeRangeId() == partitionNum) {
- TsFileResource workResource = workingProcesssor.getTsFileResource();
- boolean isCovered = workResource.getMaxPlanIndex() > tsFileResource.getMaxPlanIndex();
- if (isCovered) {
- logger.info(
- "{} is covered by a working file {}: [{}, {}] [{}, {}]",
- tsFileResource,
- workResource,
- tsFileResource.minPlanIndex,
- tsFileResource.maxPlanIndex,
- workResource.minPlanIndex,
- workResource.maxPlanIndex);
- }
- return isCovered;
- }
- }
- return false;
- }
-
public void abortCompaction() {
tsFileManager.setAllowCompaction(false);
CompactionScheduleTaskManager.getInstance().unregisterDataRegion(this);
@@ -3627,10 +3442,6 @@ public class DataRegion implements IDataRegionForQuery {
}
}
- public File getDataRegionSysDir() {
- return dataRegionSysDir;
- }
-
public void addSettleFilesToList(
List<TsFileResource> seqResourcesToBeSettled,
List<TsFileResource> unseqResourcesToBeSettled,
@@ -3692,12 +3503,6 @@ public class DataRegion implements IDataRegionForQuery {
this.tsFileManager.setAllowCompaction(allowCompaction);
}
- @FunctionalInterface
- public interface CloseTsFileCallBack {
-
- void call(TsFileProcessor caller) throws TsFileProcessorException, IOException;
- }
-
@FunctionalInterface
public interface UpdateEndTimeCallBack {
@@ -3763,6 +3568,18 @@ public class DataRegion implements IDataRegionForQuery {
}
private void acquireDirectBufferMemory() throws DataRegionException {
+ long acquireDirectBufferMemCost = getAcquireDirectBufferMemCost();
+ if (!SystemInfo.getInstance().addDirectBufferMemoryCost(acquireDirectBufferMemCost)) {
+ throw new DataRegionException(
+ "Total allocated memory for direct buffer will be "
+ + (SystemInfo.getInstance().getDirectBufferMemoryCost() + acquireDirectBufferMemCost)
+ + ", which is greater than limit mem cost: "
+ + SystemInfo.getInstance().getTotalDirectBufferMemorySizeLimit());
+ }
+ this.directBufferMemoryCost = acquireDirectBufferMemCost;
+ }
+
+ private static long getAcquireDirectBufferMemCost() {
long acquireDirectBufferMemCost = 0;
if (config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS)
|| config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.FAST_IOT_CONSENSUS)
@@ -3773,14 +3590,7 @@ public class DataRegion implements IDataRegionForQuery {
.equals(ConsensusFactory.RATIS_CONSENSUS)) {
acquireDirectBufferMemCost = config.getDataRatisConsensusLogAppenderBufferSizeMax();
}
- if (!SystemInfo.getInstance().addDirectBufferMemoryCost(acquireDirectBufferMemCost)) {
- throw new DataRegionException(
- "Total allocated memory for direct buffer will be "
- + (SystemInfo.getInstance().getDirectBufferMemoryCost() + acquireDirectBufferMemCost)
- + ", which is greater than limit mem cost: "
- + SystemInfo.getInstance().getTotalDirectBufferMemorySizeLimit());
- }
- this.directBufferMemoryCost = acquireDirectBufferMemCost;
+ return acquireDirectBufferMemCost;
}
private void releaseDirectBufferMemory() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/FlushManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/FlushManager.java
index 87f0440c3be..9e526e8267b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/FlushManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/FlushManager.java
@@ -33,7 +33,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.Future;
@SuppressWarnings("squid:S6548")
@@ -41,9 +40,6 @@ public class FlushManager implements FlushManagerMBean, IService {
private static final Logger LOGGER = LoggerFactory.getLogger(FlushManager.class);
- private ConcurrentLinkedDeque<TsFileProcessor> tsFileProcessorQueue =
- new ConcurrentLinkedDeque<>();
-
private FlushTaskPoolManager flushPool = FlushTaskPoolManager.getInstance();
@Override
@@ -98,21 +94,21 @@ public class FlushManager implements FlushManagerMBean, IService {
/** a flush thread handles flush task */
class FlushThread extends WrappedRunnable {
+ TsFileProcessor tsFileProcessor;
+
+ FlushThread(TsFileProcessor tsFileProcessor) {
+ this.tsFileProcessor = tsFileProcessor;
+ }
+
@Override
public void runMayThrow() {
- TsFileProcessor tsFileProcessor = tsFileProcessorQueue.poll();
if (null == tsFileProcessor) {
return;
}
-
- tsFileProcessor.flushOneMemTable();
- tsFileProcessor.setManagedByFlushManager(false);
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug(
- "Flush Thread re-register TSProcessor {} to the queue.",
- tsFileProcessor.getTsFileResource().getTsFile().getAbsolutePath());
+ while (tsFileProcessor.getFlushingMemTableSize() > 0) {
+ tsFileProcessor.flushOneMemTable();
}
- registerTsFileProcessor(tsFileProcessor);
+ tsFileProcessor.setManagedByFlushManager(false);
}
}
@@ -125,32 +121,16 @@ public class FlushManager implements FlushManagerMBean, IService {
public Future<?> registerTsFileProcessor(TsFileProcessor tsFileProcessor) {
synchronized (tsFileProcessor) {
if (tsFileProcessor.isManagedByFlushManager()) {
- LOGGER.debug(
- "{} is already in the flushPool, the given processor flushMemtable number = {}",
- tsFileProcessor.getTsFileResource().getTsFile().getAbsolutePath(),
- tsFileProcessor.getFlushingMemTableSize());
+ return CompletableFuture.completedFuture(null);
} else {
if (tsFileProcessor.getFlushingMemTableSize() > 0) {
- tsFileProcessorQueue.add(tsFileProcessor);
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug(
- "{} begin to submit a flush thread, flushing memtable size: {}, queue size: {}",
- tsFileProcessor.getTsFileResource().getTsFile().getAbsolutePath(),
- tsFileProcessor.getFlushingMemTableSize(),
- tsFileProcessorQueue.size());
- }
tsFileProcessor.setManagedByFlushManager(true);
- return flushPool.submit(new FlushThread());
+ return flushPool.submit(new FlushThread(tsFileProcessor));
} else {
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug(
- "No flushing memetable to do, register TsProcessor {} failed.",
- tsFileProcessor.getTsFileResource().getTsFile().getAbsolutePath());
- }
+ return CompletableFuture.completedFuture(null);
}
}
}
- return CompletableFuture.completedFuture(null);
}
private FlushManager() {}
@@ -169,9 +149,7 @@ public class FlushManager implements FlushManagerMBean, IService {
@Override
public String toString() {
return String.format(
- "TSProcessors in the queue: %d, TaskPool size %d + %d,",
- tsFileProcessorQueue.size(),
- flushPool.getWorkingTasksNumber(),
- flushPool.getWaitingTasksNumber());
+ "FlushTaskPool size %d + %d,",
+ flushPool.getWorkingTasksNumber(), flushPool.getWaitingTasksNumber());
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/TsFileFlushPolicy.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/TsFileFlushPolicy.java
index 841b5a126d2..6a51cca34e1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/TsFileFlushPolicy.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/TsFileFlushPolicy.java
@@ -35,11 +35,7 @@ public interface TsFileFlushPolicy {
@Override
public void apply(DataRegion dataRegion, TsFileProcessor tsFileProcessor, boolean isSeq) {
- if (tsFileProcessor.shouldClose()) {
- dataRegion.asyncCloseOneTsFileProcessor(isSeq, tsFileProcessor);
- } else {
- tsFileProcessor.asyncFlush();
- }
+ dataRegion.asyncCloseOneTsFileProcessor(isSeq, tsFileProcessor);
}
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index 5b025847fa4..477ae74da4b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -120,6 +120,9 @@ public class TsFileProcessor {
/** Logger fot this class. */
private static final Logger logger = LoggerFactory.getLogger(TsFileProcessor.class);
+ private static final PerformanceOverviewMetrics PERFORMANCE_OVERVIEW_METRICS =
+ PerformanceOverviewMetrics.getInstance();
+
/** Storage group name of this tsfile. */
private final String storageGroupName;
@@ -164,9 +167,6 @@ public class TsFileProcessor {
/** Working memtable. */
private IMemTable workMemTable;
- /** Last flush time to flush the working memtable. */
- private long lastWorkMemtableFlushTime;
-
/** This callback is called before the workMemtable is added into the flushingMemTables. */
private final DataRegion.UpdateEndTimeCallBack updateLatestFlushTimeCallback;
@@ -199,9 +199,6 @@ public class TsFileProcessor {
private final QueryResourceMetricSet QUERY_RESOURCE_METRICS =
QueryResourceMetricSet.getInstance();
- private static final PerformanceOverviewMetrics PERFORMANCE_OVERVIEW_METRICS =
- PerformanceOverviewMetrics.getInstance();
-
public static final int MEMTABLE_NOT_EXIST = -1;
@SuppressWarnings("squid:S107")
@@ -209,7 +206,7 @@ public class TsFileProcessor {
String storageGroupName,
File tsfile,
DataRegionInfo dataRegionInfo,
- CloseFileListener closeTsFileCallback,
+ CloseFileListener closeUnsealedTsFileProcessor,
DataRegion.UpdateEndTimeCallBack updateLatestFlushTimeCallback,
boolean sequence)
throws IOException {
@@ -226,7 +223,7 @@ public class TsFileProcessor {
.applyForWALNode(WALManager.getApplicantUniqueId(storageGroupName, sequence));
flushListeners.add(FlushListener.DefaultMemTableFLushListener.INSTANCE);
flushListeners.add(this.walNode);
- closeFileListeners.add(closeTsFileCallback);
+ closeFileListeners.add(closeUnsealedTsFileProcessor);
logger.info("create a new tsfile processor {}", tsfile.getAbsolutePath());
}
@@ -1019,10 +1016,7 @@ public class TsFileProcessor {
*/
public void deleteDataInMemory(Deletion deletion, Set<PartialPath> devicePaths) {
flushQueryLock.writeLock().lock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- FLUSH_QUERY_WRITE_LOCKED, storageGroupName, tsFileResource.getTsFile().getName());
- }
+ logFlushQueryWriteLocked();
try {
if (workMemTable != null) {
logger.info(
@@ -1041,10 +1035,7 @@ public class TsFileProcessor {
}
} finally {
flushQueryLock.writeLock().unlock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- FLUSH_QUERY_WRITE_RELEASE, storageGroupName, tsFileResource.getTsFile().getName());
- }
+ logFlushQueryWriteUnlocked();
}
}
@@ -1071,12 +1062,6 @@ public class TsFileProcessor {
return false;
}
- public boolean shouldClose() {
- long fileSize = tsFileResource.getTsFileSize();
- long fileSizeThreshold = sequence ? config.getSeqTsFileSize() : config.getUnSeqTsFileSize();
- return fileSize >= fileSizeThreshold;
- }
-
public void syncClose() {
logger.info(
"Sync close file: {}, will firstly async close it",
@@ -1116,10 +1101,7 @@ public class TsFileProcessor {
/** async close one tsfile, register and close it by another thread */
public Future<?> asyncClose() {
flushQueryLock.writeLock().lock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- FLUSH_QUERY_WRITE_LOCKED, storageGroupName, tsFileResource.getTsFile().getName());
- }
+ logFlushQueryWriteLocked();
try {
if (logger.isDebugEnabled()) {
if (workMemTable != null) {
@@ -1179,10 +1161,7 @@ public class TsFileProcessor {
}
} finally {
flushQueryLock.writeLock().unlock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- FLUSH_QUERY_WRITE_RELEASE, storageGroupName, tsFileResource.getTsFile().getName());
- }
+ logFlushQueryWriteUnlocked();
}
return CompletableFuture.completedFuture(null);
}
@@ -1194,10 +1173,7 @@ public class TsFileProcessor {
public void syncFlush() throws IOException {
IMemTable tmpMemTable;
flushQueryLock.writeLock().lock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- FLUSH_QUERY_WRITE_LOCKED, storageGroupName, tsFileResource.getTsFile().getName());
- }
+ logFlushQueryWriteLocked();
try {
tmpMemTable = workMemTable == null ? new NotifyFlushMemTable() : workMemTable;
if (logger.isDebugEnabled() && tmpMemTable.isSignalMemTable()) {
@@ -1209,10 +1185,7 @@ public class TsFileProcessor {
addAMemtableIntoFlushingList(tmpMemTable);
} finally {
flushQueryLock.writeLock().unlock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- FLUSH_QUERY_WRITE_RELEASE, storageGroupName, tsFileResource.getTsFile().getName());
- }
+ logFlushQueryWriteUnlocked();
}
synchronized (flushingMemTables) {
@@ -1242,10 +1215,7 @@ public class TsFileProcessor {
/** Put the working memtable into flushing list and set the working memtable to null */
public void asyncFlush() {
flushQueryLock.writeLock().lock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- FLUSH_QUERY_WRITE_LOCKED, storageGroupName, tsFileResource.getTsFile().getName());
- }
+ logFlushQueryWriteLocked();
try {
if (workMemTable == null) {
return;
@@ -1261,10 +1231,7 @@ public class TsFileProcessor {
e);
} finally {
flushQueryLock.writeLock().unlock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- FLUSH_QUERY_WRITE_RELEASE, storageGroupName, tsFileResource.getTsFile().getName());
- }
+ logFlushQueryWriteUnlocked();
}
}
@@ -1292,7 +1259,7 @@ public class TsFileProcessor {
flushListener.onMemTableFlushStarted(tobeFlushed);
}
- lastWorkMemtableFlushTime = System.currentTimeMillis();
+ long lastWorkMemtableFlushTime = System.currentTimeMillis();
updateLatestFlushTimeCallback.call(this, lastTimeForEachDevice, lastWorkMemtableFlushTime);
SystemInfo.getInstance().addFlushingMemTableCost(tobeFlushed.getTVListsRamCost());
@@ -1320,10 +1287,7 @@ public class TsFileProcessor {
/** Put back the memtable to MemTablePool and make metadata in writer visible */
private void releaseFlushedMemTable(IMemTable memTable) {
flushQueryLock.writeLock().lock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- FLUSH_QUERY_WRITE_LOCKED, storageGroupName, tsFileResource.getTsFile().getName());
- }
+ logFlushQueryWriteLocked();
try {
writer.makeMetadataVisible();
if (!flushingMemTables.remove(memTable)) {
@@ -1346,7 +1310,7 @@ public class TsFileProcessor {
dataRegionInfo.releaseStorageGroupMemCost(memTable.getTVListsRamCost());
if (logger.isDebugEnabled()) {
logger.debug(
- "[mem control] {}: {} flush finished, try to reset system memcost, "
+ "[mem control] {}: {} flush finished, try to reset system mem cost, "
+ "flushing memtable list size: {}",
storageGroupName,
tsFileResource.getTsFile().getName(),
@@ -1367,10 +1331,7 @@ public class TsFileProcessor {
logger.error("{}: {}", storageGroupName, tsFileResource.getTsFile().getName(), e);
} finally {
flushQueryLock.writeLock().unlock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- FLUSH_QUERY_WRITE_RELEASE, storageGroupName, tsFileResource.getTsFile().getName());
- }
+ logFlushQueryWriteUnlocked();
}
}
@@ -1679,10 +1640,6 @@ public class TsFileProcessor {
return writer;
}
- public String getStorageGroupName() {
- return storageGroupName;
- }
-
private void processAlignedChunkMetaDataFromFlushedMemTable(
IDeviceID deviceID,
AlignedChunkMetadata alignedChunkMetadata,
@@ -1917,12 +1874,7 @@ public class TsFileProcessor {
WORKING_MEMTABLE, workMemTable != null ? 1 : 0);
flushQueryLock.readLock().unlock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{}: {} release flushQueryLock",
- storageGroupName,
- tsFileResource.getTsFile().getName());
- }
+ logFlushQueryReadUnlocked();
}
if (!deviceToMemChunkHandleMap.isEmpty() || !deviceToChunkMetadataListMap.isEmpty()) {
fileScanHandlesForQuery.add(
@@ -2006,12 +1958,7 @@ public class TsFileProcessor {
WORKING_MEMTABLE, workMemTable != null ? 1 : 0);
flushQueryLock.readLock().unlock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{}: {} release flushQueryLock",
- storageGroupName,
- tsFileResource.getTsFile().getName());
- }
+ logFlushQueryReadUnlocked();
}
if (!deviceToMemChunkHandleMap.isEmpty() || !deviceToChunkMetadataListMap.isEmpty()) {
@@ -2088,12 +2035,7 @@ public class TsFileProcessor {
WORKING_MEMTABLE, workMemTable != null ? 1 : 0);
flushQueryLock.readLock().unlock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{}: {} release flushQueryLock",
- storageGroupName,
- tsFileResource.getTsFile().getName());
- }
+ logFlushQueryReadUnlocked();
}
if (!pathToReadOnlyMemChunkMap.isEmpty() || !pathToChunkMetadataListMap.isEmpty()) {
@@ -2202,4 +2144,25 @@ public class TsFileProcessor {
public ConcurrentLinkedDeque<IMemTable> getFlushingMemTable() {
return flushingMemTables;
}
+
+ private void logFlushQueryWriteLocked() {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ FLUSH_QUERY_WRITE_LOCKED, storageGroupName, tsFileResource.getTsFile().getName());
+ }
+ }
+
+ private void logFlushQueryWriteUnlocked() {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ FLUSH_QUERY_WRITE_RELEASE, storageGroupName, tsFileResource.getTsFile().getName());
+ }
+ }
+
+ private void logFlushQueryReadUnlocked() {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "{}: {} release flushQueryLock", storageGroupName, tsFileResource.getTsFile().getName());
+ }
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/generator/SimpleFileVersionController.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/generator/SimpleFileVersionController.java
deleted file mode 100644
index 3ee0bd7aff7..00000000000
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/generator/SimpleFileVersionController.java
+++ /dev/null
@@ -1,160 +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.db.storageengine.dataregion.tsfile.generator;
-
-import org.apache.iotdb.commons.file.SystemFileFactory;
-
-import org.apache.commons.io.FileUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.io.IOException;
-import java.nio.file.Files;
-
-/** SimpleFileVersionController uses a local file and its file name to store the version. */
-public class SimpleFileVersionController implements VersionController {
-
- private static final Logger logger = LoggerFactory.getLogger(SimpleFileVersionController.class);
- public static final String FILE_PREFIX = "Version-";
- public static final String UPGRADE_DIR = "upgrade";
-
- /**
- * Every time currVersion - prevVersion >= saveInterval, currVersion is persisted and prevVersion
- * is set to currVersion. When recovering from file, the version number is automatically increased
- * by saveInterval to avoid conflicts.
- */
- private static long saveInterval = 100;
-
- /** time partition id to dividing time series into different database */
- private long timePartitionId;
-
- private long prevVersion;
- private long currVersion;
- private String directoryPath;
-
- public SimpleFileVersionController(String directoryPath, long timePartitionId)
- throws IOException {
- this.directoryPath = directoryPath + File.separator + timePartitionId;
- this.timePartitionId = timePartitionId;
- restore();
- }
-
- /** only used for upgrading */
- public SimpleFileVersionController(String directoryPath) throws IOException {
- this.directoryPath = directoryPath + File.separator + UPGRADE_DIR;
- restore();
- }
-
- public static long getSaveInterval() {
- return saveInterval;
- }
-
- // test only method
- public static void setSaveInterval(long saveInterval) {
- SimpleFileVersionController.saveInterval = saveInterval;
- }
-
- public long getTimePartitionId() {
- return timePartitionId;
- }
-
- public void setTimePartitionId(long timePartitionId) {
- this.timePartitionId = timePartitionId;
- }
-
- @Override
- public synchronized long nextVersion() {
- currVersion++;
- try {
- checkPersist();
- } catch (IOException e) {
- logger.error("Error occurred when getting next version.", e);
- }
- return currVersion;
- }
-
- /**
- * Test only method, no need for concurrency.
- *
- * @return the current version.
- */
- @Override
- public long currVersion() {
- return currVersion;
- }
-
- private void checkPersist() throws IOException {
- if ((currVersion - prevVersion) >= saveInterval) {
- persist();
- }
- }
-
- private void persist() throws IOException {
- File oldFile = SystemFileFactory.INSTANCE.getFile(directoryPath, FILE_PREFIX + prevVersion);
- File newFile = SystemFileFactory.INSTANCE.getFile(directoryPath, FILE_PREFIX + currVersion);
- if (oldFile.exists()) {
- FileUtils.moveFile(oldFile, newFile);
- }
- logger.info(
- "Version file updated, previous: {}, current: {}",
- oldFile.getAbsolutePath(),
- newFile.getAbsolutePath());
- prevVersion = currVersion;
- }
-
- /** recovery from disk */
- @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
- private void restore() throws IOException {
- File directory = SystemFileFactory.INSTANCE.getFile(directoryPath);
- if (!directory.exists()) {
- directory.mkdirs();
- }
- File[] versionFiles = directory.listFiles((dir, name) -> name.startsWith(FILE_PREFIX));
- File versionFile;
- if (versionFiles != null && versionFiles.length > 0) {
- long maxVersion = 0;
- int maxVersionIndex = 0;
- for (int i = 0; i < versionFiles.length; i++) {
- // extract version from "Version-123456"
- long fileVersion = Long.parseLong(versionFiles[i].getName().split("-")[1]);
- if (fileVersion > maxVersion) {
- maxVersion = fileVersion;
- maxVersionIndex = i;
- }
- }
- prevVersion = maxVersion;
- for (int i = 0; i < versionFiles.length; i++) {
- if (i != maxVersionIndex) {
- Files.delete(versionFiles[i].toPath());
- }
- }
- } else {
- versionFile = SystemFileFactory.INSTANCE.getFile(directory, FILE_PREFIX + "0");
- prevVersion = 0;
- if (!versionFile.createNewFile()) {
- logger.warn("Cannot create new version file {}", versionFile);
- }
- }
- // prevent overlapping in case of failure
- currVersion = prevVersion + saveInterval;
- persist();
- }
-}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/generator/VersionController.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/generator/VersionController.java
deleted file mode 100644
index fb30c57acbf..00000000000
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/generator/VersionController.java
+++ /dev/null
@@ -1,44 +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.db.storageengine.dataregion.tsfile.generator;
-
-/**
- * VersionController manages the version(a monotonically increasing long) of a database. We define
- * that each memtable flush, data deletion, or data update will generate a new version of dataset.
- * So upon the above actions are performed, a new version number is generated and assigned to such
- * actions. Additionally, we also assign versions to TsFiles in their file names, so hopefully we
- * will compare files directly across IoTDB replicas. NOTICE: Thread-safety should be guaranteed by
- * the caller.
- */
-public interface VersionController {
- /**
- * Get the next version number.
- *
- * @return the next version number.
- */
- long nextVersion();
-
- /**
- * Get the current version number.
- *
- * @return the current version number.
- */
- long currVersion();
-}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/timeindex/DeviceTimeIndex.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/timeindex/DeviceTimeIndex.java
index f233261d076..69fe462d0e3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/timeindex/DeviceTimeIndex.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/timeindex/DeviceTimeIndex.java
@@ -228,11 +228,11 @@ public class DeviceTimeIndex implements ITimeIndex {
index = deviceToIndex.get(deviceId);
} else {
index = deviceToIndex.size();
- deviceToIndex.put(deviceId, index);
if (startTimes.length <= index) {
startTimes = enLargeArray(startTimes, Long.MAX_VALUE);
endTimes = enLargeArray(endTimes, Long.MIN_VALUE);
}
+ deviceToIndex.put(deviceId, index);
}
return index;
}
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 628eaf0b56d..155db11082c 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
@@ -227,15 +227,9 @@ public class DataRegionTest {
TSRecord record = new TSRecord(j, deviceId);
record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(j)));
dataRegion.insert(buildInsertRowNodeByTSRecord(record));
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
}
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- dataRegion.syncCloseAllWorkingTsFileProcessors();
QueryDataSource queryDataSource =
dataRegion.query(
Collections.singletonList(new PartialPath(deviceId, measurementId)),
@@ -288,7 +282,7 @@ public class DataRegionTest {
times.length);
dataRegion.insertTablet(insertTabletNode1);
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
for (int r = 50; r < 149; r++) {
times[r - 50] = r;
@@ -310,7 +304,6 @@ public class DataRegionTest {
times.length);
dataRegion.insertTablet(insertTabletNode2);
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
dataRegion.syncCloseAllWorkingTsFileProcessors();
QueryDataSource queryDataSource =
@@ -449,7 +442,7 @@ public class DataRegionTest {
times.length);
dataRegion.insertTablet(insertTabletNode1);
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
for (int r = 50; r < 149; r++) {
times[r - 50] = r;
@@ -471,7 +464,6 @@ public class DataRegionTest {
times.length);
dataRegion.insertTablet(insertTabletNode2);
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
dataRegion.syncCloseAllWorkingTsFileProcessors();
QueryDataSource queryDataSource =
@@ -529,7 +521,7 @@ public class DataRegionTest {
insertTabletNode1.setFailedMeasurementNumber(2);
dataRegion.insertTablet(insertTabletNode1);
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
for (int r = 50; r < 149; r++) {
times[r - 50] = r;
@@ -552,7 +544,6 @@ public class DataRegionTest {
insertTabletNode2.setFailedMeasurementNumber(2);
dataRegion.insertTablet(insertTabletNode2);
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
dataRegion.syncCloseAllWorkingTsFileProcessors();
QueryDataSource queryDataSource =
@@ -577,19 +568,16 @@ public class DataRegionTest {
TSRecord record = new TSRecord(j, deviceId);
record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(j)));
dataRegion.insert(buildInsertRowNodeByTSRecord(record));
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
}
- dataRegion.syncCloseAllWorkingTsFileProcessors();
for (int j = 10; j >= 1; j--) {
TSRecord record = new TSRecord(j, deviceId);
record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(j)));
dataRegion.insert(buildInsertRowNodeByTSRecord(record));
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
}
- dataRegion.syncCloseAllWorkingTsFileProcessors();
-
QueryDataSource queryDataSource =
dataRegion.query(
Collections.singletonList(new PartialPath(deviceId, measurementId)),
@@ -616,9 +604,8 @@ public class DataRegionTest {
InsertRowNode rowNode = buildInsertRowNodeByTSRecord(record);
rowNode.setFailedMeasurementNumber(1);
dataRegion.insert(rowNode);
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
}
- dataRegion.syncCloseAllWorkingTsFileProcessors();
for (int j = 10; j >= 1; j--) {
TSRecord record = new TSRecord(j, deviceId);
@@ -626,11 +613,9 @@ public class DataRegionTest {
InsertRowNode rowNode = buildInsertRowNodeByTSRecord(record);
rowNode.setFailedMeasurementNumber(1);
dataRegion.insert(rowNode);
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
}
- dataRegion.syncCloseAllWorkingTsFileProcessors();
-
QueryDataSource queryDataSource =
dataRegion.query(
Collections.singletonList(new PartialPath(deviceId, measurementId)),
@@ -658,19 +643,16 @@ public class DataRegionTest {
TSRecord record = new TSRecord(j, deviceId);
record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(j)));
dataRegion.insert(buildInsertRowNodeByTSRecord(record));
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
}
- dataRegion.syncCloseAllWorkingTsFileProcessors();
for (int j = 10; j >= 1; j--) {
TSRecord record = new TSRecord(j, deviceId);
record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(j)));
dataRegion.insert(buildInsertRowNodeByTSRecord(record));
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
}
- dataRegion.syncCloseAllWorkingTsFileProcessors();
-
for (TsFileProcessor tsfileProcessor : dataRegion.getWorkUnsequenceTsFileProcessors()) {
tsfileProcessor.syncFlush();
}
@@ -737,7 +719,7 @@ public class DataRegionTest {
times.length);
dataRegion.insertTablet(insertTabletNode1);
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
for (int r = 149; r >= 50; r--) {
times[r - 50] = r;
@@ -758,7 +740,6 @@ public class DataRegionTest {
times.length);
dataRegion.insertTablet(insertTabletNode2);
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
dataRegion.syncCloseAllWorkingTsFileProcessors();
for (TsFileProcessor tsfileProcessor : dataRegion.getWorkUnsequenceTsFileProcessors()) {
@@ -826,7 +807,7 @@ public class DataRegionTest {
times.length);
dataRegion.insertTablet(insertTabletNode1);
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
for (int r = 1249; r >= 50; r--) {
times[r - 50] = r;
@@ -847,7 +828,6 @@ public class DataRegionTest {
times.length);
dataRegion.insertTablet(insertTabletNode2);
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
dataRegion.syncCloseAllWorkingTsFileProcessors();
for (TsFileProcessor tsfileProcessor : dataRegion.getWorkUnsequenceTsFileProcessors()) {
@@ -915,7 +895,7 @@ public class DataRegionTest {
times.length);
dataRegion.insertTablet(insertTabletNode1);
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
for (int r = 1249; r >= 50; r--) {
times[r - 50] = r;
@@ -936,7 +916,6 @@ public class DataRegionTest {
times.length);
dataRegion.insertTablet(insertTabletNode2);
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
dataRegion.syncCloseAllWorkingTsFileProcessors();
for (TsFileProcessor tsfileProcessor : dataRegion.getWorkUnsequenceTsFileProcessors()) {
@@ -1015,7 +994,7 @@ public class DataRegionTest {
TSRecord record = new TSRecord(j, "root.ln22");
record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(j)));
dataRegion1.insert(buildInsertRowNodeByTSRecord(record));
- dataRegion1.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion1.syncCloseAllWorkingTsFileProcessors();
}
dataRegion1.syncCloseAllWorkingTsFileProcessors();
@@ -1064,7 +1043,7 @@ public class DataRegionTest {
TSRecord record = new TSRecord(j, deviceId);
record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(j)));
dataRegion.insert(buildInsertRowNodeByTSRecord(record));
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
}
dataRegion.syncCloseAllWorkingTsFileProcessors();
@@ -1072,7 +1051,7 @@ public class DataRegionTest {
TSRecord record = new TSRecord(j, deviceId);
record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(j)));
dataRegion.insert(buildInsertRowNodeByTSRecord(record));
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
}
dataRegion.syncCloseAllWorkingTsFileProcessors();
@@ -1133,7 +1112,7 @@ public class DataRegionTest {
TSRecord record = new TSRecord(j, deviceId);
record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(j)));
dataRegion.insert(buildInsertRowNodeByTSRecord(record));
- dataRegion.asyncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
}
dataRegion.syncCloseAllWorkingTsFileProcessors();
ICompactionPerformer performer = new FastCompactionPerformer(false);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/generator/SimpleFileVersionControllerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/generator/SimpleFileVersionControllerTest.java
deleted file mode 100644
index 85fd1ebc696..00000000000
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/generator/SimpleFileVersionControllerTest.java
+++ /dev/null
@@ -1,56 +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.db.storageengine.dataregion.tsfile.generator;
-
-import org.apache.iotdb.db.utils.constant.TestConstant;
-
-import org.apache.commons.io.FileUtils;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.File;
-import java.io.IOException;
-
-import static org.junit.Assert.assertEquals;
-
-public class SimpleFileVersionControllerTest {
- @Test
- public void test() throws IOException {
- String tempFilePath = TestConstant.BASE_OUTPUT_PATH.concat("version.tmp");
-
- try {
- if (!new File(tempFilePath).mkdir()) {
- Assert.fail("can not create version.tmp folder");
- }
- VersionController versionController = new SimpleFileVersionController(tempFilePath, 1);
- assertEquals(SimpleFileVersionController.getSaveInterval(), versionController.currVersion());
- for (int i = 0; i < 150; i++) {
- versionController.nextVersion();
- }
- assertEquals(
- SimpleFileVersionController.getSaveInterval() + 150, versionController.currVersion());
- versionController = new SimpleFileVersionController(tempFilePath, 1);
- assertEquals(
- SimpleFileVersionController.getSaveInterval() + 200, versionController.currVersion());
- } finally {
- FileUtils.deleteDirectory(new File(tempFilePath));
- }
- }
-}