@Test public void testMerge() throws WriteProcessException, QueryProcessException, IllegalPathException { mergeLock = new AtomicLong(0); for (int j = 21; j <= 30; j++) { TSRecord record = new TSRecord(j, deviceId); record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(j))); insertToStorageGroupProcessor(record); processor.asyncCloseAllWorkingTsFileProcessors(); } processor.syncCloseAllWorkingTsFileProcessors(); for (int j = 10; j >= 1; j--) { TSRecord record = new TSRecord(j, deviceId); record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(j))); insertToStorageGroupProcessor(record); processor.asyncCloseAllWorkingTsFileProcessors(); } processor.syncCloseAllWorkingTsFileProcessors(); processor.merge(true); while (mergeLock.get() == 0) { } QueryDataSource queryDataSource = processor.query(new PartialPath(deviceId), measurementId, context, null, null); Assert.assertEquals(10, queryDataSource.getSeqResources().size()); Assert.assertEquals(0, queryDataSource.getUnseqResources().size()); for (TsFileResource resource : queryDataSource.getSeqResources()) { Assert.assertTrue(resource.isClosed()); } for (TsFileResource resource : queryDataSource.getUnseqResources()) { Assert.assertTrue(resource.isClosed()); } }
public void merge(boolean fullMerge) { writeLock(); try { if (isMerging) { if (logger.isInfoEnabled()) { logger.info("{} Last merge is ongoing, currently consumed time: {}ms", storageGroupName, (System.currentTimeMillis() - mergeStartTime)); } return; } logger.info("{} will close all files for starting a merge (fullmerge = {})", storageGroupName, fullMerge); List<TsFileResource> seqMergeList = tsFileManagement.getStableTsFileList(true); List<TsFileResource> unSeqMergeList = tsFileManagement.getStableTsFileList(false); if (seqMergeList.isEmpty() || unSeqMergeList.isEmpty()) { logger.info("{} no files to be merged", storageGroupName); return; } long budget = IoTDBDescriptor.getInstance().getConfig().getMergeMemoryBudget(); long timeLowerBound = System.currentTimeMillis() - dataTTL; MergeResource mergeResource = new MergeResource(seqMergeList, unSeqMergeList, timeLowerBound); IMergeFileSelector fileSelector = getMergeFileSelector(budget, mergeResource); try { List[] mergeFiles = fileSelector.select(); if (mergeFiles.length == 0) { logger.info("{} cannot select merge candidates under the budget {}", storageGroupName, budget); return; } mergeResource.clear(); String taskName = storageGroupName + "-" + System.currentTimeMillis(); mergeResource.setCacheDeviceMeta(true); for (TsFileResource tsFileResource : mergeResource.getSeqFiles()) { tsFileResource.setMerging(true); } for (TsFileResource tsFileResource : mergeResource.getUnseqFiles()) { tsFileResource.setMerging(true); } MergeTask mergeTask = new MergeTask(mergeResource, storageGroupSysDir.getPath(), this::mergeEndAction, taskName, fullMerge, fileSelector.getConcurrentMergeNum(), storageGroupName); mergingModification = new ModificationFile( storageGroupSysDir + File.separator + MERGING_MODIFICATION_FILE_NAME); MergeManager.getINSTANCE().submitMainTask(mergeTask); if (logger.isInfoEnabled()) { logger.info("{} submits a merge task {}, merging {} seqFiles, {} unseqFiles", storageGroupName, taskName, mergeFiles[0].size(), mergeFiles[1].size()); } isMerging = true; mergeStartTime = System.currentTimeMillis(); } catch (MergeException | IOException e) { logger.error("{} cannot select file for merge", storageGroupName, e); } } finally { writeUnlock(); } }
StorageGroupProcessor { public void merge(boolean fullMerge) { writeLock(); try { if (isMerging) { if (logger.isInfoEnabled()) { logger.info("{} Last merge is ongoing, currently consumed time: {}ms", storageGroupName, (System.currentTimeMillis() - mergeStartTime)); } return; } logger.info("{} will close all files for starting a merge (fullmerge = {})", storageGroupName, fullMerge); List<TsFileResource> seqMergeList = tsFileManagement.getStableTsFileList(true); List<TsFileResource> unSeqMergeList = tsFileManagement.getStableTsFileList(false); if (seqMergeList.isEmpty() || unSeqMergeList.isEmpty()) { logger.info("{} no files to be merged", storageGroupName); return; } long budget = IoTDBDescriptor.getInstance().getConfig().getMergeMemoryBudget(); long timeLowerBound = System.currentTimeMillis() - dataTTL; MergeResource mergeResource = new MergeResource(seqMergeList, unSeqMergeList, timeLowerBound); IMergeFileSelector fileSelector = getMergeFileSelector(budget, mergeResource); try { List[] mergeFiles = fileSelector.select(); if (mergeFiles.length == 0) { logger.info("{} cannot select merge candidates under the budget {}", storageGroupName, budget); return; } mergeResource.clear(); String taskName = storageGroupName + "-" + System.currentTimeMillis(); mergeResource.setCacheDeviceMeta(true); for (TsFileResource tsFileResource : mergeResource.getSeqFiles()) { tsFileResource.setMerging(true); } for (TsFileResource tsFileResource : mergeResource.getUnseqFiles()) { tsFileResource.setMerging(true); } MergeTask mergeTask = new MergeTask(mergeResource, storageGroupSysDir.getPath(), this::mergeEndAction, taskName, fullMerge, fileSelector.getConcurrentMergeNum(), storageGroupName); mergingModification = new ModificationFile( storageGroupSysDir + File.separator + MERGING_MODIFICATION_FILE_NAME); MergeManager.getINSTANCE().submitMainTask(mergeTask); if (logger.isInfoEnabled()) { logger.info("{} submits a merge task {}, merging {} seqFiles, {} unseqFiles", storageGroupName, taskName, mergeFiles[0].size(), mergeFiles[1].size()); } isMerging = true; mergeStartTime = System.currentTimeMillis(); } catch (MergeException | IOException e) { logger.error("{} cannot select file for merge", storageGroupName, e); } } finally { writeUnlock(); } } }
StorageGroupProcessor { public void merge(boolean fullMerge) { writeLock(); try { if (isMerging) { if (logger.isInfoEnabled()) { logger.info("{} Last merge is ongoing, currently consumed time: {}ms", storageGroupName, (System.currentTimeMillis() - mergeStartTime)); } return; } logger.info("{} will close all files for starting a merge (fullmerge = {})", storageGroupName, fullMerge); List<TsFileResource> seqMergeList = tsFileManagement.getStableTsFileList(true); List<TsFileResource> unSeqMergeList = tsFileManagement.getStableTsFileList(false); if (seqMergeList.isEmpty() || unSeqMergeList.isEmpty()) { logger.info("{} no files to be merged", storageGroupName); return; } long budget = IoTDBDescriptor.getInstance().getConfig().getMergeMemoryBudget(); long timeLowerBound = System.currentTimeMillis() - dataTTL; MergeResource mergeResource = new MergeResource(seqMergeList, unSeqMergeList, timeLowerBound); IMergeFileSelector fileSelector = getMergeFileSelector(budget, mergeResource); try { List[] mergeFiles = fileSelector.select(); if (mergeFiles.length == 0) { logger.info("{} cannot select merge candidates under the budget {}", storageGroupName, budget); return; } mergeResource.clear(); String taskName = storageGroupName + "-" + System.currentTimeMillis(); mergeResource.setCacheDeviceMeta(true); for (TsFileResource tsFileResource : mergeResource.getSeqFiles()) { tsFileResource.setMerging(true); } for (TsFileResource tsFileResource : mergeResource.getUnseqFiles()) { tsFileResource.setMerging(true); } MergeTask mergeTask = new MergeTask(mergeResource, storageGroupSysDir.getPath(), this::mergeEndAction, taskName, fullMerge, fileSelector.getConcurrentMergeNum(), storageGroupName); mergingModification = new ModificationFile( storageGroupSysDir + File.separator + MERGING_MODIFICATION_FILE_NAME); MergeManager.getINSTANCE().submitMainTask(mergeTask); if (logger.isInfoEnabled()) { logger.info("{} submits a merge task {}, merging {} seqFiles, {} unseqFiles", storageGroupName, taskName, mergeFiles[0].size(), mergeFiles[1].size()); } isMerging = true; mergeStartTime = System.currentTimeMillis(); } catch (MergeException | IOException e) { logger.error("{} cannot select file for merge", storageGroupName, e); } } finally { writeUnlock(); } } StorageGroupProcessor(String systemDir, String storageGroupName,
      TsFileFlushPolicy fileFlushPolicy); }
StorageGroupProcessor { public void merge(boolean fullMerge) { writeLock(); try { if (isMerging) { if (logger.isInfoEnabled()) { logger.info("{} Last merge is ongoing, currently consumed time: {}ms", storageGroupName, (System.currentTimeMillis() - mergeStartTime)); } return; } logger.info("{} will close all files for starting a merge (fullmerge = {})", storageGroupName, fullMerge); List<TsFileResource> seqMergeList = tsFileManagement.getStableTsFileList(true); List<TsFileResource> unSeqMergeList = tsFileManagement.getStableTsFileList(false); if (seqMergeList.isEmpty() || unSeqMergeList.isEmpty()) { logger.info("{} no files to be merged", storageGroupName); return; } long budget = IoTDBDescriptor.getInstance().getConfig().getMergeMemoryBudget(); long timeLowerBound = System.currentTimeMillis() - dataTTL; MergeResource mergeResource = new MergeResource(seqMergeList, unSeqMergeList, timeLowerBound); IMergeFileSelector fileSelector = getMergeFileSelector(budget, mergeResource); try { List[] mergeFiles = fileSelector.select(); if (mergeFiles.length == 0) { logger.info("{} cannot select merge candidates under the budget {}", storageGroupName, budget); return; } mergeResource.clear(); String taskName = storageGroupName + "-" + System.currentTimeMillis(); mergeResource.setCacheDeviceMeta(true); for (TsFileResource tsFileResource : mergeResource.getSeqFiles()) { tsFileResource.setMerging(true); } for (TsFileResource tsFileResource : mergeResource.getUnseqFiles()) { tsFileResource.setMerging(true); } MergeTask mergeTask = new MergeTask(mergeResource, storageGroupSysDir.getPath(), this::mergeEndAction, taskName, fullMerge, fileSelector.getConcurrentMergeNum(), storageGroupName); mergingModification = new ModificationFile( storageGroupSysDir + File.separator + MERGING_MODIFICATION_FILE_NAME); MergeManager.getINSTANCE().submitMainTask(mergeTask); if (logger.isInfoEnabled()) { logger.info("{} submits a merge task {}, merging {} seqFiles, {} unseqFiles", storageGroupName, taskName, mergeFiles[0].size(), mergeFiles[1].size()); } isMerging = true; mergeStartTime = System.currentTimeMillis(); } catch (MergeException | IOException e) { logger.error("{} cannot select file for merge", storageGroupName, e); } } finally { writeUnlock(); } } StorageGroupProcessor(String systemDir, String storageGroupName,
      TsFileFlushPolicy fileFlushPolicy); void insert(InsertRowPlan insertRowPlan); @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning void insertTablet(InsertTabletPlan insertTabletPlan); void asyncCloseOneTsFileProcessor(boolean sequence, TsFileProcessor tsFileProcessor); void deleteFolder(String systemDir); void closeAllResources(); void syncDeleteDataFiles(); synchronized void checkFilesTTL(); void syncCloseAllWorkingTsFileProcessors(); void asyncCloseAllWorkingTsFileProcessors(); void forceCloseAllWorkingTsFileProcessors(); QueryDataSource query(PartialPath deviceId, String measurementId, QueryContext context,
      QueryFileManager filePathsManager, Filter timeFilter); void writeLock(); void writeUnlock(); void delete(PartialPath deviceId, String measurementId, long startTime, long endTime); void updateNewlyFlushedPartitionLatestFlushedTimeForEachDevice(long partitionId,
      String deviceId, long time); int countUpgradeFiles(); void upgrade(); void merge(boolean fullMerge); void loadNewTsFileForSync(TsFileResource newTsFileResource); void loadNewTsFile(TsFileResource newTsFileResource); void setPartitionFileVersionToMax(long partition, long version); @SuppressWarnings("unused") void removeFullyOverlapFiles(TsFileResource resource); boolean deleteTsfile(File tsfieToBeDeleted); Collection<TsFileProcessor> getWorkSequenceTsFileProcessors(); boolean moveTsfile(File fileToBeMoved, File targetDir); Collection<TsFileProcessor> getWorkUnsequenceTsFileProcessor(); void setDataTTL(long dataTTL); List<TsFileResource> getSequenceFileTreeSet(); List<TsFileResource> getUnSequenceFileList(); String getStorageGroupName(); boolean isFileAlreadyExist(TsFileResource tsFileResource, long partitionNum); void removePartitions(TimePartitionFilter filter); boolean isHotCompactionMergeWorking(); }
StorageGroupProcessor { public void merge(boolean fullMerge) { writeLock(); try { if (isMerging) { if (logger.isInfoEnabled()) { logger.info("{} Last merge is ongoing, currently consumed time: {}ms", storageGroupName, (System.currentTimeMillis() - mergeStartTime)); } return; } logger.info("{} will close all files for starting a merge (fullmerge = {})", storageGroupName, fullMerge); List<TsFileResource> seqMergeList = tsFileManagement.getStableTsFileList(true); List<TsFileResource> unSeqMergeList = tsFileManagement.getStableTsFileList(false); if (seqMergeList.isEmpty() || unSeqMergeList.isEmpty()) { logger.info("{} no files to be merged", storageGroupName); return; } long budget = IoTDBDescriptor.getInstance().getConfig().getMergeMemoryBudget(); long timeLowerBound = System.currentTimeMillis() - dataTTL; MergeResource mergeResource = new MergeResource(seqMergeList, unSeqMergeList, timeLowerBound); IMergeFileSelector fileSelector = getMergeFileSelector(budget, mergeResource); try { List[] mergeFiles = fileSelector.select(); if (mergeFiles.length == 0) { logger.info("{} cannot select merge candidates under the budget {}", storageGroupName, budget); return; } mergeResource.clear(); String taskName = storageGroupName + "-" + System.currentTimeMillis(); mergeResource.setCacheDeviceMeta(true); for (TsFileResource tsFileResource : mergeResource.getSeqFiles()) { tsFileResource.setMerging(true); } for (TsFileResource tsFileResource : mergeResource.getUnseqFiles()) { tsFileResource.setMerging(true); } MergeTask mergeTask = new MergeTask(mergeResource, storageGroupSysDir.getPath(), this::mergeEndAction, taskName, fullMerge, fileSelector.getConcurrentMergeNum(), storageGroupName); mergingModification = new ModificationFile( storageGroupSysDir + File.separator + MERGING_MODIFICATION_FILE_NAME); MergeManager.getINSTANCE().submitMainTask(mergeTask); if (logger.isInfoEnabled()) { logger.info("{} submits a merge task {}, merging {} seqFiles, {} unseqFiles", storageGroupName, taskName, mergeFiles[0].size(), mergeFiles[1].size()); } isMerging = true; mergeStartTime = System.currentTimeMillis(); } catch (MergeException | IOException e) { logger.error("{} cannot select file for merge", storageGroupName, e); } } finally { writeUnlock(); } } StorageGroupProcessor(String systemDir, String storageGroupName,
      TsFileFlushPolicy fileFlushPolicy); void insert(InsertRowPlan insertRowPlan); @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning void insertTablet(InsertTabletPlan insertTabletPlan); void asyncCloseOneTsFileProcessor(boolean sequence, TsFileProcessor tsFileProcessor); void deleteFolder(String systemDir); void closeAllResources(); void syncDeleteDataFiles(); synchronized void checkFilesTTL(); void syncCloseAllWorkingTsFileProcessors(); void asyncCloseAllWorkingTsFileProcessors(); void forceCloseAllWorkingTsFileProcessors(); QueryDataSource query(PartialPath deviceId, String measurementId, QueryContext context,
      QueryFileManager filePathsManager, Filter timeFilter); void writeLock(); void writeUnlock(); void delete(PartialPath deviceId, String measurementId, long startTime, long endTime); void updateNewlyFlushedPartitionLatestFlushedTimeForEachDevice(long partitionId,
      String deviceId, long time); int countUpgradeFiles(); void upgrade(); void merge(boolean fullMerge); void loadNewTsFileForSync(TsFileResource newTsFileResource); void loadNewTsFile(TsFileResource newTsFileResource); void setPartitionFileVersionToMax(long partition, long version); @SuppressWarnings("unused") void removeFullyOverlapFiles(TsFileResource resource); boolean deleteTsfile(File tsfieToBeDeleted); Collection<TsFileProcessor> getWorkSequenceTsFileProcessors(); boolean moveTsfile(File fileToBeMoved, File targetDir); Collection<TsFileProcessor> getWorkUnsequenceTsFileProcessor(); void setDataTTL(long dataTTL); List<TsFileResource> getSequenceFileTreeSet(); List<TsFileResource> getUnSequenceFileList(); String getStorageGroupName(); boolean isFileAlreadyExist(TsFileResource tsFileResource, long partitionNum); void removePartitions(TimePartitionFilter filter); boolean isHotCompactionMergeWorking();  }