jt2594838 commented on code in PR #12122: URL: https://github.com/apache/iotdb/pull/12122#discussion_r1580446812
########## iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/SettleCompactionTask.java: ########## @@ -0,0 +1,389 @@ +/* + * 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.compaction.execute.task; + +import org.apache.iotdb.db.service.metrics.FileMetrics; +import org.apache.iotdb.db.storageengine.dataregion.compaction.constant.CompactionTaskType; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionRecoverException; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ICompactionPerformer; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.log.CompactionLogAnalyzer; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.log.CompactionLogger; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.log.SimpleCompactionLogger; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.log.TsFileIdentifier; +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 java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +/** + * This settle task contains all_deleted files and partial_deleted files. The partial_deleted files + * are divided into several groups, each group may contain one or several files. This task will do + * the following two things respectively: 1. Settle all all_deleted files by deleting them directly. + * 2. Settle partial_deleted files: put the files of each partial_deleted group into an invisible + * innerCompactionTask, and then perform the cleanup work. The source files in a file group will be + * compacted into a target file. + */ +public class SettleCompactionTask extends InnerSpaceCompactionTask { + private List<TsFileResource> fullyDeletedFiles; + private double fullyDeletedFileSize = 0; + private double partialDeletedFileSize = 0; Review Comment: partial -> partially ########## iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleSummary.java: ########## @@ -20,12 +20,21 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.schedule; import org.apache.iotdb.db.storageengine.dataregion.compaction.constant.CompactionTaskType; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.AbstractCompactionTask; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.SettleCompactionTask; public class CompactionScheduleSummary { private int submitSeqInnerSpaceCompactionTaskNum = 0; private int submitUnseqInnerSpaceCompactionTaskNum = 0; private int submitCrossSpaceCompactionTaskNum = 0; private int submitInsertionCrossSpaceCompactionTaskNum = 0; + private int submitSettleCompactionTaskNum = 0; + + // region TTL info + private int allDeletedFileNum = 0; + + private int partialDeletedFileNum = 0; Review Comment: Previous issues reappear. ########## iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/settle/SettleCompactionRecoverTest.java: ########## @@ -0,0 +1,1137 @@ +/* + * 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.compaction.settle; + +import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.exception.StorageEngineException; +import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeTTLCache; +import org.apache.iotdb.db.storageengine.dataregion.compaction.AbstractCompactionTest; +import org.apache.iotdb.db.storageengine.dataregion.compaction.constant.CompactionTaskType; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.SettleCompactionTask; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.subtask.FastCompactionTaskSummary; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionUtils; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.log.CompactionLogger; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.log.SimpleCompactionLogger; +import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionFileGeneratorUtils; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator; +import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor; +import org.apache.iotdb.tsfile.exception.write.WriteProcessException; +import org.apache.iotdb.tsfile.utils.Pair; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR; + +public class SettleCompactionRecoverTest extends AbstractCompactionTest { + @Before + public void setUp() + throws IOException, WriteProcessException, MetadataException, InterruptedException { + super.setUp(); + IoTDBDescriptor.getInstance().getConfig().setTargetChunkSize(512); + IoTDBDescriptor.getInstance().getConfig().setTargetChunkPointNum(100); + TSFileDescriptor.getInstance().getConfig().setMaxNumberOfPointsInPage(10); Review Comment: Add a comment to avoid other people asking. ########## iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties: ########## @@ -450,6 +450,25 @@ data_replication_factor=1 # Unit: ms # default_ttl_in_ms=-1 +# The threshold for the number of TTL stored in the system, the default is 1000. Review Comment: "count" would better be "capacity". "count" usually means how many you have now instead of a restriction. ########## iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties: ########## @@ -450,6 +450,25 @@ data_replication_factor=1 # Unit: ms # default_ttl_in_ms=-1 +# The maximum number of TTL rules stored in the system, the default is 1000. +# Negative value means the threshold is unlimited. +# Datatype: int +# ttl_rule_count=1000 + +# The interval of TTL check task in each database. The TTL check task will inspect and select files with a higher volume of expired data for compaction. Default is 2 hours. +# Notice: It is not recommended to change it too small, as it will affect the read and write performance of the system. +# Unit: ms +# ttl_check_interval=7200000 + +# The maximum expiring time of devices that have a ttl. Default is 1 month. +# If the data elapsed time (current timestamp minus the maximum data timestamp of the device in the file) of such devices exceeds this value, then the file will be cleaned by compaction. +# Notice: It is not recommended to change it too small, as it will affect the read and write performance of the system. +# Unit: ms +# max_expired_time=2592000000 + +# The expired device ratio. If the ratio of expired devices in one file exceeds this value, then expired data of this file will be cleaned by compaction. +# expired_data_rate=0.3 Review Comment: rate -> ratio ########## iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java: ########## @@ -119,6 +119,9 @@ public class CommonConfig { */ private long[] tierTTLInMs = {Long.MAX_VALUE}; + /** The maximum number of TTL rules stored in the system, the default is 1000. */ + private int TTLRuleCount = 1000; Review Comment: Still not precise enough. Usually, "count" means the current number you have. In this case, it should probably be "capacity". ########## iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/impl/SettleSelectorImpl.java: ########## @@ -0,0 +1,335 @@ +/* + * 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.compaction.selector.impl; + +import org.apache.iotdb.commons.conf.IoTDBConstant; +import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.utils.CommonDateTimeUtils; +import org.apache.iotdb.db.conf.IoTDBConfig; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeTTLCache; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ICompactionPerformer; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.SettleCompactionTask; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionUtils; +import org.apache.iotdb.db.storageengine.dataregion.compaction.selector.ISettleSelector; +import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion; +import org.apache.iotdb.db.storageengine.dataregion.modification.Modification; +import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; +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.timeindex.DeviceTimeIndex; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndex; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ITimeIndex; + +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.PlainDeviceID; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import static org.apache.iotdb.db.storageengine.dataregion.compaction.selector.impl.SettleSelectorImpl.DirtyStatus.PARTIAL_DELETED; + +public class SettleSelectorImpl implements ISettleSelector { + private static final Logger LOGGER = + LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME); + private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + + // this parameter indicates whether to use the costly method for settle file selection. The + // high-cost selection has a lower triggering frequency, while the low-cost selection has a higher + // triggering frequency. + private final boolean heavySelect; + private final String storageGroupName; + private final String dataRegionId; + private final long timePartition; + private final TsFileManager tsFileManager; + private boolean isSeq; + + public SettleSelectorImpl( + boolean heavySelect, + String storageGroupName, + String dataRegionId, + long timePartition, + TsFileManager tsFileManager) { + this.heavySelect = heavySelect; + this.storageGroupName = storageGroupName; + this.dataRegionId = dataRegionId; + this.timePartition = timePartition; + this.tsFileManager = tsFileManager; + } + + static class FileDirtyInfo { + DirtyStatus status; + long dirtyDataSize = 0; + + public FileDirtyInfo(DirtyStatus status) { + this.status = status; + } + + public FileDirtyInfo(DirtyStatus status, long dirtyDataSize) { + this.status = status; + this.dirtyDataSize = dirtyDataSize; + } + } + + static class PartialDirtyResource { + List<TsFileResource> resources = new ArrayList<>(); + long totalFileSize = 0; + + public boolean add(TsFileResource resource, long dirtyDataSize) { + resources.add(resource); + totalFileSize += resource.getTsFileSize(); + totalFileSize -= dirtyDataSize; + return checkHasReachedThreshold(); + } + + public List<TsFileResource> getResources() { + return resources; + } + + public boolean checkHasReachedThreshold() { + return resources.size() >= config.getFileLimitPerInnerTask() + || totalFileSize >= config.getTargetCompactionFileSize(); + } + } + + @Override + public List<SettleCompactionTask> selectSettleTask(List<TsFileResource> tsFileResources) { + if (tsFileResources.isEmpty()) { + return Collections.emptyList(); + } + this.isSeq = tsFileResources.get(0).isSeq(); + return selectTasks(tsFileResources); + } + + private List<SettleCompactionTask> selectTasks(List<TsFileResource> resources) { + List<TsFileResource> fullyDirtyResource = new ArrayList<>(); + List<PartialDirtyResource> partialDirtyResourceList = new ArrayList<>(); + PartialDirtyResource partialDirtyResource = new PartialDirtyResource(); + try { + for (TsFileResource resource : resources) { + if (resource.getStatus() != TsFileResourceStatus.NORMAL) { + continue; + } + boolean shouldStop = false; + FileDirtyInfo fileDirtyInfo; + if (!heavySelect) { + fileDirtyInfo = selectFileBaseOnModSize(resource); + } else { + fileDirtyInfo = selectFileBaseOnDirtyData(resource); + } + + switch (fileDirtyInfo.status) { + case FULLY_DELETED: + fullyDirtyResource.add(resource); + break; + case PARTIAL_DELETED: + shouldStop = partialDirtyResource.add(resource, fileDirtyInfo.dirtyDataSize); + break; + case NOT_SATISFIED: + shouldStop = !partialDirtyResource.getResources().isEmpty(); + break; + default: + // do nothing + } + + if (shouldStop) { + partialDirtyResourceList.add(partialDirtyResource); + partialDirtyResource = new PartialDirtyResource(); + if (!heavySelect) { + // Non-heavy selection is triggered more frequently. In order to avoid selecting too + // many files containing mods for compaction when the disk is insufficient, the number + // and size of files are limited here. + break; + } + } + } + partialDirtyResourceList.add(partialDirtyResource); + return createTask(fullyDirtyResource, partialDirtyResourceList); + } catch (Exception e) { + LOGGER.error( + "{}-{} cannot select file for settle compaction", storageGroupName, dataRegionId, e); + } + return Collections.emptyList(); + } + + private FileDirtyInfo selectFileBaseOnModSize(TsFileResource resource) { + ModificationFile modFile = resource.getModFile(); + if (modFile == null || !modFile.exists()) { + return new FileDirtyInfo(DirtyStatus.NOT_SATISFIED); + } + return modFile.getSize() > config.getInnerCompactionTaskSelectionModsFileThreshold() + || !CompactionUtils.isDiskHasSpace( + config.getInnerCompactionTaskSelectionDiskRedundancy()) + ? new FileDirtyInfo(PARTIAL_DELETED) + : new FileDirtyInfo(DirtyStatus.NOT_SATISFIED); + } + + /** + * Only when all devices with ttl are deleted may they be selected. On the basic of the previous, + * only when the number of deleted devices exceeds the threshold or has expired for too long will + * they be selected. + * + * @return dirty status means the status of current resource. + */ + private FileDirtyInfo selectFileBaseOnDirtyData(TsFileResource resource) + throws IOException, IllegalPathException { + ModificationFile modFile = resource.getModFile(); + ITimeIndex timeIndex = resource.getTimeIndex(); + if (timeIndex instanceof FileTimeIndex) { + timeIndex = resource.buildDeviceTimeIndex(); + } + Set<IDeviceID> deletedDevices = new HashSet<>(); + boolean hasExpiredTooLong = false; + long currentTime = CommonDateTimeUtils.currentTime(); + + Collection<Modification> modifications = modFile.getModifications(); + for (IDeviceID device : ((DeviceTimeIndex) timeIndex).getDevices()) { + // check expired device by ttl + // TODO: remove deviceId conversion + long deviceTTL = DataNodeTTLCache.getInstance().getTTL(((PlainDeviceID) device).toStringID()); + boolean hasSetTTL = deviceTTL != Long.MAX_VALUE; + boolean isDeleted = + !timeIndex.isDeviceAlive(device, deviceTTL) + || isDeviceDeletedByMods( + modifications, + device, + timeIndex.getStartTime(device), + timeIndex.getEndTime(device)); + + if (hasSetTTL) { + if (!isDeleted) { + // For devices with TTL set, all data must expire in order to meet the conditions for + // being selected. + return new FileDirtyInfo(DirtyStatus.NOT_SATISFIED); + } + long outdatedTimeDiff = currentTime - timeIndex.getEndTime(device); + hasExpiredTooLong = + hasExpiredTooLong + || outdatedTimeDiff > Math.min(config.getMaxExpiredTime(), 3 * deviceTTL); + } + + if (isDeleted) { + deletedDevices.add(device); + } + } + + double deletedDeviceRatio = + ((double) deletedDevices.size()) / ((DeviceTimeIndex) timeIndex).getDevices().size(); + if (deletedDeviceRatio == 1d) { + // the whole file is completely dirty + return new FileDirtyInfo(DirtyStatus.FULLY_DELETED); + } + hasExpiredTooLong = config.getMaxExpiredTime() != Long.MAX_VALUE && hasExpiredTooLong; + if (hasExpiredTooLong || deletedDeviceRatio >= config.getExpiredDataRate()) { + // evaluate dirty data size in the tsfile + return new FileDirtyInfo( + PARTIAL_DELETED, (long) (deletedDeviceRatio * resource.getTsFileSize())); + } + return new FileDirtyInfo(DirtyStatus.NOT_SATISFIED); + } + + /** Check whether the device is completely deleted by mods or not. */ + private boolean isDeviceDeletedByMods( + Collection<Modification> modifications, IDeviceID device, long startTime, long endTime) + throws IllegalPathException { + for (Modification modification : modifications) { + PartialPath path = modification.getPath(); + if (path.endWithMultiLevelWildcard() + && path.getDevicePath().matchFullPath(new PartialPath(device)) + && ((Deletion) modification).getTimeRange().contains(startTime, endTime)) { + return true; + } + } + return false; + } + + private List<SettleCompactionTask> createTask( + List<TsFileResource> fullyDirtyResources, + List<PartialDirtyResource> partialDirtyResourceList) { + List<SettleCompactionTask> tasks = new ArrayList<>(); + for (int i = 0; i < partialDirtyResourceList.size(); i++) { + if (i == 0) { + if (fullyDirtyResources.isEmpty() + && partialDirtyResourceList.get(i).getResources().isEmpty()) { + continue; + } + tasks.add( + new SettleCompactionTask( + timePartition, + tsFileManager, + fullyDirtyResources, + partialDirtyResourceList.get(i).getResources(), + isSeq, + createCompactionPerformer(), + tsFileManager.getNextCompactionTaskId())); + } else { + if (partialDirtyResourceList.get(i).getResources().isEmpty()) { + continue; + } + tasks.add( + new SettleCompactionTask( + timePartition, + tsFileManager, + Collections.emptyList(), + partialDirtyResourceList.get(i).getResources(), + isSeq, + createCompactionPerformer(), + tsFileManager.getNextCompactionTaskId())); + } + } + return tasks; + } + + private ICompactionPerformer createCompactionPerformer() { + return isSeq + ? IoTDBDescriptor.getInstance() + .getConfig() + .getInnerSeqCompactionPerformer() + .createInstance() + : IoTDBDescriptor.getInstance() + .getConfig() + .getInnerUnseqCompactionPerformer() + .createInstance(); + } + + enum DirtyStatus { + FULLY_DELETED, // the whole file is deleted + PARTIAL_DELETED, // the file is partial deleted + NOT_SATISFIED; // do not satisfy settle condition, which does not mean there is no dirty data + + private long dirtyDataSize = 0; + + public void setDirtyDataSize(long dirtyDataSize) { + this.dirtyDataSize = dirtyDataSize; + } + + public long getDirtyDataSize() { + return dirtyDataSize; + } + } +} Review Comment: Is this still used? -- 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]
