cigarl commented on a change in pull request #4537:
URL: https://github.com/apache/iotdb/pull/4537#discussion_r771139526
##########
File path:
server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/inplace/task/MergeMultiChunkTask.java
##########
@@ -112,44 +116,60 @@ public MergeMultiChunkTask(
InplaceCompactionLogger inplaceCompactionLogger,
CrossSpaceMergeResource mergeResource,
boolean fullMerge,
- List<PartialPath> unmergedSeries,
+ List<PartialPath> unmergedDevice,
int concurrentMergeSeriesNum,
String storageGroupName) {
this.mergeContext = context;
this.taskName = taskName;
this.inplaceCompactionLogger = inplaceCompactionLogger;
this.resource = mergeResource;
this.fullMerge = fullMerge;
- this.unmergedSeries = unmergedSeries;
+ this.unmergedDevice = unmergedDevice;
this.concurrentMergeSeriesNum = concurrentMergeSeriesNum;
this.storageGroupName = storageGroupName;
}
void mergeSeries() throws IOException {
- if (logger.isInfoEnabled()) {
- logger.info("{} starts to merge {} series", taskName,
unmergedSeries.size());
- }
long startTime = System.currentTimeMillis();
for (TsFileResource seqFile : resource.getSeqFiles()) {
// record the unmergeChunkStartTime for each sensor in each file
mergeContext.getUnmergedChunkStartTimes().put(seqFile, new HashMap<>());
}
+ mergedDeviceCnt = 0;
// merge each series and write data into each seqFile's corresponding temp
merge file
- List<List<PartialPath>> devicePaths =
MergeUtils.splitPathsByDevice(unmergedSeries);
- for (List<PartialPath> pathList : devicePaths) {
+ for (PartialPath device : unmergedDevice) {
// TODO: use statistics of queries to better rearrange series
- IMergePathSelector pathSelector = new NaivePathSelector(pathList,
concurrentMergeSeriesNum);
- while (pathSelector.hasNext()) {
- currMergingPaths = pathSelector.next();
- mergePaths();
- resource.clearChunkWriterCache();
- if (Thread.interrupted()) {
- logger.info("MergeMultiChunkTask {} aborted", taskName);
- Thread.currentThread().interrupt();
- return;
+ List<PartialPath> measurementPathListByDevice;
+ try {
+ measurementPathListByDevice =
+ new
ArrayList<>(IoTDB.metaManager.getAllMeasurementByDevicePath(device));
+ } catch (PathNotExistException e) {
+ logger.error("Can not get all measurements of {}.",
device.getDevice());
+ continue;
+ }
+ // just for unit tests, we need to consider whether there is a need to
exist
+ Collections.sort(measurementPathListByDevice);
+
+ IMergePathSelector pathSelector =
+ new NaivePathSelector(measurementPathListByDevice,
concurrentMergeSeriesNum);
+ try {
+ while (pathSelector.hasNext()) {
+ currMergingPaths = pathSelector.next();
+ mergePaths();
+ resource.clearChunkWriterCache();
+ if (Thread.interrupted()) {
+ logger.info("MergeMultiChunkTask {} aborted", taskName);
+ Thread.currentThread().interrupt();
+ return;
+ }
}
- mergedSeriesCnt += currMergingPaths.size();
+ mergedDeviceCnt++;
logMergeProgress();
+ } catch (MetadataException e) {
+ logger.warn(
Review comment:
yes，you are right~
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
