You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2020/10/20 02:25:58 UTC
[GitHub] [iotdb] samperson1997 commented on a change in pull request #1758: add device chunk point cache
samperson1997 commented on a change in pull request #1758:
URL: https://github.com/apache/iotdb/pull/1758#discussion_r507688530
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/tsfilemanagement/level/LevelTsFileManagement.java
##########
@@ -31,29 +31,36 @@
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CopyOnWriteArrayList;
+
+import com.clearspring.analytics.stream.cardinality.HyperLogLog;
+import com.clearspring.analytics.stream.cardinality.ICardinality;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.cache.ChunkMetadataCache;
+import org.apache.iotdb.db.engine.merge.manage.MergeManager;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.merge.selector.IMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.task.MergeTask;
+import org.apache.iotdb.db.engine.modification.ModificationFile;
Review comment:
Remove the useless import
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/tsfilemanagement/TsFileManagement.java
##########
@@ -146,4 +178,207 @@ public void run() {
closeHotCompactionMergeCallBack.call();
}
}
+
+ public void merge(boolean fullMerge, List<TsFileResource> seqMergeList,
+ List<TsFileResource> unSeqMergeList, long dataTTL) {
+ if (isUnseqMerging) {
+ 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);
+
+ if (seqMergeList.isEmpty()) {
+ logger.info("{} no seq files to be merged", storageGroupName);
+ return;
+ }
+
+ if (unSeqMergeList.isEmpty()) {
+ logger.info("{} no unseq 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;
+ }
+ // avoid pending tasks holds the metadata and streams
+ mergeResource.clear();
+ String taskName = storageGroupName + "-" + System.currentTimeMillis();
+ // do not cache metadata until true candidates are chosen, or too much metadata will be
+ // cached during selection
+ mergeResource.setCacheDeviceMeta(true);
+
+ for (TsFileResource tsFileResource : mergeResource.getSeqFiles()) {
+ tsFileResource.setMerging(true);
+ }
+ for (TsFileResource tsFileResource : mergeResource.getUnseqFiles()) {
+ tsFileResource.setMerging(true);
+ }
+
+ MergeTask mergeTask = new MergeTask(mergeResource, storageGroupDir,
+ this::mergeEndAction, taskName, fullMerge, fileSelector.getConcurrentMergeNum(),
+ storageGroupName);
+ mergingModification = new ModificationFile(
+ storageGroupDir + 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());
+ }
+ isUnseqMerging = true;
+ mergeStartTime = System.currentTimeMillis();
+
+ } catch (MergeException | IOException e) {
+ logger.error("{} cannot select file for merge", storageGroupName, e);
+ }
+ }
+
+ private IMergeFileSelector getMergeFileSelector(long budget, MergeResource resource) {
+ MergeFileStrategy strategy = IoTDBDescriptor.getInstance().getConfig().getMergeFileStrategy();
+ switch (strategy) {
+ case MAX_FILE_NUM:
+ return new MaxFileMergeFileSelector(resource, budget);
+ case MAX_SERIES_NUM:
+ return new MaxSeriesMergeFileSelector(resource, budget);
+ default:
+ throw new UnsupportedOperationException("Unknown MergeFileStrategy " + strategy);
+ }
+ }
+
+ /**
+ * acquire the write locks of the resource , the merge lock and the hot compaction lock
+ */
+ private void doubleWriteLock(TsFileResource seqFile) {
+ boolean fileLockGot;
+ boolean mergeLockGot;
+ boolean hotCompactionLockGot;
+ while (true) {
+ fileLockGot = seqFile.tryWriteLock();
+ mergeLockGot = mergeLock.writeLock().tryLock();
+ hotCompactionLockGot = tryWriteLock();
+
+ if (fileLockGot && mergeLockGot && hotCompactionLockGot) {
+ break;
+ } else {
+ // did not get all of them, release the gotten one and retry
+ if (hotCompactionLockGot) {
+ writeUnlock();
+ }
+ if (mergeLockGot) {
+ mergeLock.writeLock().unlock();
+ }
+ if (fileLockGot) {
+ seqFile.writeUnlock();
+ }
+ }
+ }
+ }
+
+ /**
+ * release the write locks of the resource , the merge lock and the hot compaction lock
+ */
+ private void doubleWriteUnlock(TsFileResource seqFile) {
+ writeUnlock();
+ mergeLock.writeLock().unlock();
+ seqFile.writeUnlock();
+ }
+
+ private void removeUnseqFiles(List<TsFileResource> unseqFiles) {
+ mergeLock.writeLock().lock();
+ writeLock();
+ try {
+ removeAll(unseqFiles, false);
+ } finally {
+ writeUnlock();
+ mergeLock.writeLock().unlock();
+ }
+
+ for (TsFileResource unseqFile : unseqFiles) {
+ unseqFile.writeLock();
+ try {
+ unseqFile.remove();
+ } finally {
+ unseqFile.writeUnlock();
+ }
+ }
+ }
+
+ @SuppressWarnings("squid:S1141")
+ private void updateMergeModification(TsFileResource seqFile) {
+ try {
+ // remove old modifications and write modifications generated during merge
+ seqFile.removeModFile();
+ if (mergingModification != null) {
+ for (Modification modification : mergingModification.getModifications()) {
+ seqFile.getModFile().write(modification);
+ }
+ try {
+ seqFile.getModFile().close();
+ } catch (IOException e) {
+ logger
+ .error("Cannot close the ModificationFile {}", seqFile.getModFile().getFilePath(), e);
+ }
+ }
+ } catch (IOException e) {
+ logger.error("{} cannot clean the ModificationFile of {} after merge", storageGroupName,
+ seqFile.getTsFile(), e);
+ }
+ }
+
+ private void removeMergingModification() {
+ try {
+ if (mergingModification != null) {
+ mergingModification.remove();
+ mergingModification = null;
+ }
+ } catch (IOException e) {
+ logger.error("{} cannot remove merging modification ", storageGroupName, e);
+ }
+ }
+
+ public void mergeEndAction(List<TsFileResource> seqFiles, List<TsFileResource> unseqFiles,
+ File mergeLog) {
+ logger.info("{} a merge task is ending...", storageGroupName);
+
+ if (unseqFiles.isEmpty()) {
+ // merge runtime exception arose, just end this merge
+ isUnseqMerging = false;
+ logger.info("{} a merge task abnormally ends", storageGroupName);
+ return;
+ }
+
+ removeUnseqFiles(unseqFiles);
+
+ for (int i = 0; i < seqFiles.size(); i++) {
+ TsFileResource seqFile = seqFiles.get(i);
+ // get both seqFile lock and merge lock
+ doubleWriteLock(seqFile);
+
+ try {
+ updateMergeModification(seqFile);
+ if (i == seqFiles.size() - 1) {
+ //FIXME if there is an exception, the the modification file will be not closed.
+ removeMergingModification();
+ isUnseqMerging = false;
+ mergeLog.delete();
Review comment:
```suggestion
Files.delete(mergeLog.toPath());
```
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/tsfilemanagement/level/LevelTsFileManagement.java
##########
@@ -407,135 +430,152 @@ public void recover() {
}
@Override
- public void forkCurrentFileList(long timePartition) throws IOException {
- Pair<Long, Map<Path, MeasurementSchema>> seqResult = forkTsFileList(
+ public void forkCurrentFileList(long timePartition) {
+ Pair<Double, Double> seqStatisticsPair = forkTsFileList(
forkedSequenceTsFileResources,
- sequenceTsFileResources.computeIfAbsent(timePartition, this::newSequenceTsFileResources));
- forkedSeqListPointNum = seqResult.left;
- forkedSeqListPathMeasurementSchemaMap = seqResult.right;
- Pair<Long, Map<Path, MeasurementSchema>> unSeqResult = forkTsFileList(
+ sequenceTsFileResources.computeIfAbsent(timePartition, this::newSequenceTsFileResources),
+ maxLevelNum);
+ forkedSeqListPointNum = seqStatisticsPair.left;
+ forkedSeqListMeasurementSize = seqStatisticsPair.right;
+ Pair<Double, Double> unSeqStatisticsPair = forkTsFileList(
forkedUnSequenceTsFileResources,
unSequenceTsFileResources
- .computeIfAbsent(timePartition, this::newUnSequenceTsFileResources));
- forkedUnSeqListPointNum = unSeqResult.left;
- forkedUnSeqListPathMeasurementSchemaMap = unSeqResult.right;
+ .computeIfAbsent(timePartition, this::newUnSequenceTsFileResources), maxUnseqLevelNum);
+ forkedUnSeqListPointNum = unSeqStatisticsPair.left;
+ forkedUnSeqListMeasurementSize = unSeqStatisticsPair.right;
}
- private Pair<Long, Map<Path, MeasurementSchema>> forkTsFileList(
+ private Pair<Double, Double> forkTsFileList(
List<List<TsFileResource>> forkedTsFileResources,
- List rawTsFileResources) throws IOException {
+ List rawTsFileResources, int currMaxLevel) {
forkedTsFileResources.clear();
// just fork part of the TsFile list, controlled by max_merge_chunk_point
long pointNum = 0;
// all flush to target file
- Map<Path, MeasurementSchema> pathMeasurementSchemaMap = new HashMap<>();
- for (int i = 0; i < maxLevelNum - 1; i++) {
+ ICardinality measurementSet = new HyperLogLog(13);
+ for (int i = 0; i < currMaxLevel - 1; i++) {
List<TsFileResource> forkedLevelTsFileResources = new ArrayList<>();
Collection<TsFileResource> levelRawTsFileResources = (Collection<TsFileResource>) rawTsFileResources
.get(i);
- for (TsFileResource tsFileResource : levelRawTsFileResources) {
- if (tsFileResource.isClosed()) {
- RestorableTsFileIOWriter writer;
- try {
- writer = new RestorableTsFileIOWriter(
- tsFileResource.getTsFile());
- } catch (Exception e) {
- logger.error("[Hot Compaction] {} open writer failed",
- tsFileResource.getTsFile().getPath(), e);
- continue;
- }
- Map<String, Map<String, List<ChunkMetadata>>> schemaMap = writer
- .getMetadatasForQuery();
- for (Entry<String, Map<String, List<ChunkMetadata>>> schemaMapEntry : schemaMap
- .entrySet()) {
- String device = schemaMapEntry.getKey();
- for (Entry<String, List<ChunkMetadata>> entry : schemaMapEntry.getValue()
- .entrySet()) {
- String measurement = entry.getKey();
- List<ChunkMetadata> chunkMetadataList = entry.getValue();
- for (ChunkMetadata chunkMetadata : chunkMetadataList) {
- pointNum += chunkMetadata.getNumOfPoints();
+ synchronized (levelRawTsFileResources) {
+ for (TsFileResource tsFileResource : levelRawTsFileResources) {
+ if (tsFileResource.isClosed()) {
+ String path = tsFileResource.getTsFile().getAbsolutePath();
+ try {
+ if (tsFileResource.getTsFile().exists()) {
+ TsFileSequenceReader reader = new TsFileSequenceReader(path);
Review comment:
reader should be closed
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/tsfilemanagement/utils/HotCompactionUtils.java
##########
@@ -91,130 +84,204 @@ private HotCompactionUtils() {
return new Pair<>(newChunkMetadata, newChunk);
}
- private static long readUnseqChunk(String storageGroup,
- Map<String, TsFileSequenceReader> tsFileSequenceReaderMap, String deviceId, long maxVersion,
- String measurementId,
- Map<Long, TimeValuePair> timeValuePairMap, List<TsFileResource> levelResources)
+ private static long readByDeserializeMerge(RateLimiter compactionReadRateLimiter,
+ Map<TsFileSequenceReader, List<ChunkMetadata>> readerChunkMetadataMap, long maxVersion,
+ Map<Long, TimeValuePair> timeValuePairMap)
throws IOException {
- for (TsFileResource levelResource : levelResources) {
- TsFileSequenceReader reader = buildReaderFromTsFileResource(levelResource,
- tsFileSequenceReaderMap,
- storageGroup);
- if (reader == null) {
- continue;
- }
- List<ChunkMetadata> chunkMetadataList = reader
- .getChunkMetadataList(new Path(deviceId, measurementId));
+ for (Entry<TsFileSequenceReader, List<ChunkMetadata>> entry : readerChunkMetadataMap
+ .entrySet()) {
+ TsFileSequenceReader reader = entry.getKey();
+ List<ChunkMetadata> chunkMetadataList = entry.getValue();
for (ChunkMetadata chunkMetadata : chunkMetadataList) {
maxVersion = Math.max(chunkMetadata.getVersion(), maxVersion);
IChunkReader chunkReader = new ChunkReaderByTimestamp(
reader.readMemChunk(chunkMetadata));
+ long chunkSize = 0;
while (chunkReader.hasNextSatisfiedPage()) {
IPointReader iPointReader = new BatchDataIterator(
chunkReader.nextPageData());
while (iPointReader.hasNextTimeValuePair()) {
TimeValuePair timeValuePair = iPointReader.nextTimeValuePair();
+ chunkSize += timeValuePair.getSize();
timeValuePairMap.put(timeValuePair.getTimestamp(), timeValuePair);
}
}
+ MergeManager
+ .mergeRateLimiterAcquire(compactionReadRateLimiter, chunkSize);
}
}
return maxVersion;
}
- private static void fillDeviceMeasurementMap(Set<String> devices,
- Map<String, Map<String, MeasurementSchema>> deviceMeasurementMap,
- List<TsFileResource> subLevelResources,
+ private static long writeByAppendMerge(long maxVersion, String device,
+ RateLimiter compactionWriteRateLimiter, RateLimiter compactionReadRateLimiter,
+ Map<TsFileSequenceReader, List<ChunkMetadata>> readerChunkMetadatasMap,
+ TsFileResource targetResource, RestorableTsFileIOWriter writer) throws IOException {
+ Pair<ChunkMetadata, Chunk> chunkPair = readByAppendMerge(compactionReadRateLimiter,
+ readerChunkMetadatasMap);
+ ChunkMetadata newChunkMetadata = chunkPair.left;
+ Chunk newChunk = chunkPair.right;
+ if (newChunkMetadata != null && newChunk != null) {
+ maxVersion = Math.max(newChunkMetadata.getVersion(), maxVersion);
+ // wait for limit write
+ MergeManager.mergeRateLimiterAcquire(compactionWriteRateLimiter,
+ newChunk.getHeader().getDataSize() + newChunk.getData().position());
Review comment:
```suggestion
(long) newChunk.getHeader().getDataSize() + newChunk.getData().position());
```
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -208,26 +200,15 @@
private File storageGroupSysDir;
// manage seqFileList and unSeqFileList
- private TsFileManagement tsFileManagement;
+ public TsFileManagement tsFileManagement;
Review comment:
Why public?
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/tsfilemanagement/HotCompactionMergeTaskPoolManager.java
##########
@@ -100,7 +104,8 @@ public ServiceType getID() {
return ServiceType.HOT_COMPACTION_SERVICE;
}
- public void submitTask(HotCompactionMergeTask hotCompactionMergeTask) {
+ public void submitTask(HotCompactionMergeTask hotCompactionMergeTask)
+ throws RejectedExecutionException {
Review comment:
`RejectedExecutionException` is a runtime exception. No need to throw it.
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/tsfilemanagement/TsFileManagement.java
##########
@@ -19,24 +19,56 @@
package org.apache.iotdb.db.engine.tsfilemanagement;
+import static org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.MERGING_MODIFICATION_FILE_NAME;
+
+import java.io.File;
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.merge.manage.MergeManager;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.merge.selector.IMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.selector.MaxFileMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.selector.MaxSeriesMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.selector.MergeFileStrategy;
+import org.apache.iotdb.db.engine.merge.task.MergeTask;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.CloseHotCompactionMergeCallBack;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.engine.tsfilemanagement.level.LevelTsFileManagement;
+import org.apache.iotdb.db.exception.MergeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public abstract class TsFileManagement {
+ private static final Logger logger = LoggerFactory.getLogger(TsFileManagement.class);
protected String storageGroupName;
protected String storageGroupDir;
+
+ /**
+ * mergeLock is to be used in the merge process. Concurrent queries, deletions and merges may
+ * result in losing some deletion in the merged new file, so a lock is necessary.
+ */
+ public ReentrantReadWriteLock mergeLock = new ReentrantReadWriteLock();
/**
* hotCompactionMergeLock is used to wait for TsFile list change in hot compaction merge
* processor.
*/
private final ReadWriteLock hotCompactionMergeLock = new ReentrantReadWriteLock();
+ public volatile boolean isUnseqMerging = false;
+ /**
+ * This is the modification file of the result of the current merge. Because the merged file may
+ * be invisible at this moment, without this, deletion/update during merge could be lost.
+ */
+ public ModificationFile mergingModification;
+ public long mergeStartTime;
Review comment:
Please check the accessibility of **all new** fields ... Class variable fields should not have public accessibility
```suggestion
private long mergeStartTime;
```
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/tsfilemanagement/TsFileManagement.java
##########
@@ -19,24 +19,56 @@
package org.apache.iotdb.db.engine.tsfilemanagement;
+import static org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.MERGING_MODIFICATION_FILE_NAME;
+
+import java.io.File;
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.merge.manage.MergeManager;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.merge.selector.IMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.selector.MaxFileMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.selector.MaxSeriesMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.selector.MergeFileStrategy;
+import org.apache.iotdb.db.engine.merge.task.MergeTask;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.CloseHotCompactionMergeCallBack;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.engine.tsfilemanagement.level.LevelTsFileManagement;
+import org.apache.iotdb.db.exception.MergeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public abstract class TsFileManagement {
+ private static final Logger logger = LoggerFactory.getLogger(TsFileManagement.class);
protected String storageGroupName;
protected String storageGroupDir;
+
+ /**
+ * mergeLock is to be used in the merge process. Concurrent queries, deletions and merges may
+ * result in losing some deletion in the merged new file, so a lock is necessary.
+ */
+ public ReentrantReadWriteLock mergeLock = new ReentrantReadWriteLock();
Review comment:
```suggestion
private final ReentrantReadWriteLock mergeLock = new ReentrantReadWriteLock();
```
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/tsfilemanagement/utils/HotCompactionUtils.java
##########
@@ -53,32 +55,23 @@
public class HotCompactionUtils {
private static final Logger logger = LoggerFactory.getLogger(HotCompactionUtils.class);
+ private static final int mergePagePointNum = IoTDBDescriptor.getInstance().getConfig()
Review comment:
```suggestion
private static final int MERGE_PAGE_POINT_NUM = IoTDBDescriptor.getInstance().getConfig()
```
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/tsfilemanagement/TsFileManagement.java
##########
@@ -19,24 +19,56 @@
package org.apache.iotdb.db.engine.tsfilemanagement;
+import static org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.MERGING_MODIFICATION_FILE_NAME;
+
+import java.io.File;
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.merge.manage.MergeManager;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.merge.selector.IMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.selector.MaxFileMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.selector.MaxSeriesMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.selector.MergeFileStrategy;
+import org.apache.iotdb.db.engine.merge.task.MergeTask;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.CloseHotCompactionMergeCallBack;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.engine.tsfilemanagement.level.LevelTsFileManagement;
Review comment:
Remove unused import
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/tsfilemanagement/TsFileManagement.java
##########
@@ -19,24 +19,56 @@
package org.apache.iotdb.db.engine.tsfilemanagement;
+import static org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.MERGING_MODIFICATION_FILE_NAME;
+
+import java.io.File;
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.merge.manage.MergeManager;
+import org.apache.iotdb.db.engine.merge.manage.MergeResource;
+import org.apache.iotdb.db.engine.merge.selector.IMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.selector.MaxFileMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.selector.MaxSeriesMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.selector.MergeFileStrategy;
+import org.apache.iotdb.db.engine.merge.task.MergeTask;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.CloseHotCompactionMergeCallBack;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.engine.tsfilemanagement.level.LevelTsFileManagement;
+import org.apache.iotdb.db.exception.MergeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public abstract class TsFileManagement {
+ private static final Logger logger = LoggerFactory.getLogger(TsFileManagement.class);
protected String storageGroupName;
protected String storageGroupDir;
+
+ /**
+ * mergeLock is to be used in the merge process. Concurrent queries, deletions and merges may
+ * result in losing some deletion in the merged new file, so a lock is necessary.
+ */
+ public ReentrantReadWriteLock mergeLock = new ReentrantReadWriteLock();
/**
* hotCompactionMergeLock is used to wait for TsFile list change in hot compaction merge
* processor.
*/
private final ReadWriteLock hotCompactionMergeLock = new ReentrantReadWriteLock();
+ public volatile boolean isUnseqMerging = false;
Review comment:
```suggestion
private volatile boolean isUnseqMerging = false;
```
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/tsfilemanagement/utils/HotCompactionUtils.java
##########
@@ -91,130 +84,204 @@ private HotCompactionUtils() {
return new Pair<>(newChunkMetadata, newChunk);
}
- private static long readUnseqChunk(String storageGroup,
- Map<String, TsFileSequenceReader> tsFileSequenceReaderMap, String deviceId, long maxVersion,
- String measurementId,
- Map<Long, TimeValuePair> timeValuePairMap, List<TsFileResource> levelResources)
+ private static long readByDeserializeMerge(RateLimiter compactionReadRateLimiter,
+ Map<TsFileSequenceReader, List<ChunkMetadata>> readerChunkMetadataMap, long maxVersion,
+ Map<Long, TimeValuePair> timeValuePairMap)
throws IOException {
- for (TsFileResource levelResource : levelResources) {
- TsFileSequenceReader reader = buildReaderFromTsFileResource(levelResource,
- tsFileSequenceReaderMap,
- storageGroup);
- if (reader == null) {
- continue;
- }
- List<ChunkMetadata> chunkMetadataList = reader
- .getChunkMetadataList(new Path(deviceId, measurementId));
+ for (Entry<TsFileSequenceReader, List<ChunkMetadata>> entry : readerChunkMetadataMap
+ .entrySet()) {
+ TsFileSequenceReader reader = entry.getKey();
+ List<ChunkMetadata> chunkMetadataList = entry.getValue();
for (ChunkMetadata chunkMetadata : chunkMetadataList) {
maxVersion = Math.max(chunkMetadata.getVersion(), maxVersion);
IChunkReader chunkReader = new ChunkReaderByTimestamp(
reader.readMemChunk(chunkMetadata));
+ long chunkSize = 0;
while (chunkReader.hasNextSatisfiedPage()) {
IPointReader iPointReader = new BatchDataIterator(
chunkReader.nextPageData());
while (iPointReader.hasNextTimeValuePair()) {
TimeValuePair timeValuePair = iPointReader.nextTimeValuePair();
+ chunkSize += timeValuePair.getSize();
timeValuePairMap.put(timeValuePair.getTimestamp(), timeValuePair);
}
}
+ MergeManager
+ .mergeRateLimiterAcquire(compactionReadRateLimiter, chunkSize);
}
}
return maxVersion;
}
- private static void fillDeviceMeasurementMap(Set<String> devices,
- Map<String, Map<String, MeasurementSchema>> deviceMeasurementMap,
- List<TsFileResource> subLevelResources,
+ private static long writeByAppendMerge(long maxVersion, String device,
+ RateLimiter compactionWriteRateLimiter, RateLimiter compactionReadRateLimiter,
+ Map<TsFileSequenceReader, List<ChunkMetadata>> readerChunkMetadatasMap,
+ TsFileResource targetResource, RestorableTsFileIOWriter writer) throws IOException {
+ Pair<ChunkMetadata, Chunk> chunkPair = readByAppendMerge(compactionReadRateLimiter,
+ readerChunkMetadatasMap);
+ ChunkMetadata newChunkMetadata = chunkPair.left;
+ Chunk newChunk = chunkPair.right;
+ if (newChunkMetadata != null && newChunk != null) {
+ maxVersion = Math.max(newChunkMetadata.getVersion(), maxVersion);
+ // wait for limit write
+ MergeManager.mergeRateLimiterAcquire(compactionWriteRateLimiter,
+ newChunk.getHeader().getDataSize() + newChunk.getData().position());
+ writer.writeChunk(newChunk, newChunkMetadata);
+ targetResource.updateStartTime(device, newChunkMetadata.getStartTime());
+ targetResource.updateEndTime(device, newChunkMetadata.getEndTime());
+ }
+ return maxVersion;
+ }
+
+ private static long writeByDeserializeMerge(long maxVersion, String device,
+ RateLimiter compactionRateLimiter, RateLimiter compactionReadRateLimiter,
+ Entry<String, Map<TsFileSequenceReader, List<ChunkMetadata>>> entry,
+ TsFileResource targetResource, RestorableTsFileIOWriter writer) throws IOException {
+ Map<Long, TimeValuePair> timeValuePairMap = new TreeMap<>();
+ maxVersion = readByDeserializeMerge(compactionReadRateLimiter, entry.getValue(), maxVersion,
+ timeValuePairMap);
+ Iterator<List<ChunkMetadata>> chunkMetadataListIterator = entry.getValue().values()
+ .iterator();
+ if (!chunkMetadataListIterator.hasNext()) {
+ return maxVersion;
+ }
+ List<ChunkMetadata> chunkMetadataList = chunkMetadataListIterator.next();
+ if (chunkMetadataList.size() <= 0) {
Review comment:
```suggestion
if (chunkMetadataList.isEmpty()) {
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org