You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by xi...@apache.org on 2022/12/16 03:59:43 UTC
[iotdb] branch master updated: skip validating resource with file time index after compaction (#8473)
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 8f10abdb8b skip validating resource with file time index after compaction (#8473)
8f10abdb8b is described below
commit 8f10abdb8b23a7d8acb3bf5f90a6b40eddfb0177
Author: 周沛辰 <45...@users.noreply.github.com>
AuthorDate: Fri Dec 16 11:59:38 2022 +0800
skip validating resource with file time index after compaction (#8473)
---
.../db/engine/compaction/CompactionUtils.java | 22 +++--
...eCompactionWithFastPerformerValidationTest.java | 100 +++++++++++++++++++++
2 files changed, 115 insertions(+), 7 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionUtils.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionUtils.java
index d4b5a33e60..7787a7611a 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionUtils.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
import org.slf4j.Logger;
@@ -269,30 +270,37 @@ public class CompactionUtils {
Long.compareUnsigned(
Long.parseLong(f1.getTsFile().getName().split("-")[0]),
Long.parseLong(f2.getTsFile().getName().split("-")[0])));
- Map<String, Long> lastEndTimeMap = new HashMap<>();
- TsFileResource prevTsFileResource = null;
+ // deviceID -> <TsFileResource, last end time>
+ Map<String, Pair<TsFileResource, Long>> lastEndTimeMap = new HashMap<>();
for (TsFileResource resource : resources) {
+ if (resource.getTimeIndexType() != 1) {
+ // if time index is not device time index, then skip it
+ continue;
+ }
Set<String> devices = resource.getDevices();
for (String device : devices) {
long currentStartTime = resource.getStartTime(device);
long currentEndTime = resource.getEndTime(device);
- long lastEndTime = lastEndTimeMap.computeIfAbsent(device, x -> Long.MIN_VALUE);
+ Pair<TsFileResource, Long> lastDeviceInfo =
+ lastEndTimeMap.computeIfAbsent(device, x -> new Pair<>(null, Long.MIN_VALUE));
+ long lastEndTime = lastDeviceInfo.right;
if (lastEndTime >= currentStartTime) {
logger.error(
"{} Device {} is overlapped between {} and {}, end time in {} is {}, start time in {} is {}",
storageGroupName,
device,
- prevTsFileResource,
+ lastDeviceInfo.left,
resource,
- prevTsFileResource,
+ lastDeviceInfo.left,
lastEndTime,
resource,
currentStartTime);
return false;
}
- lastEndTimeMap.put(device, currentEndTime);
+ lastDeviceInfo.left = resource;
+ lastDeviceInfo.right = currentEndTime;
+ lastEndTimeMap.put(device, lastDeviceInfo);
}
- prevTsFileResource = resource;
}
return true;
}
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/CrossSpaceCompactionWithFastPerformerValidationTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/CrossSpaceCompactionWithFastPerformerValidationTest.java
index e19bd0a11f..4a939da189 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/CrossSpaceCompactionWithFastPerformerValidationTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/CrossSpaceCompactionWithFastPerformerValidationTest.java
@@ -22,10 +22,14 @@ package org.apache.iotdb.db.engine.compaction.cross;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.compaction.AbstractCompactionTest;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
import org.apache.iotdb.db.engine.compaction.cross.rewrite.CrossSpaceCompactionResource;
import org.apache.iotdb.db.engine.compaction.cross.rewrite.RewriteCrossSpaceCompactionSelector;
+import org.apache.iotdb.db.engine.compaction.inner.InnerSpaceCompactionTask;
import org.apache.iotdb.db.engine.compaction.performer.ICrossCompactionPerformer;
import org.apache.iotdb.db.engine.compaction.performer.impl.FastCompactionPerformer;
+import org.apache.iotdb.db.engine.compaction.task.CompactionTaskSummary;
+import org.apache.iotdb.db.engine.compaction.utils.CompactionFileGeneratorUtils;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.engine.storagegroup.TsFileResourceStatus;
@@ -43,9 +47,13 @@ import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
+import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_SEPARATOR;
+
public class CrossSpaceCompactionWithFastPerformerValidationTest extends AbstractCompactionTest {
TsFileManager tsFileManager =
new TsFileManager(COMPACTION_TEST_SG, "0", STORAGE_GROUP_DIR.getPath());
@@ -2092,4 +2100,96 @@ public class CrossSpaceCompactionWithFastPerformerValidationTest extends Abstrac
validateSeqFiles(true);
}
+
+ /**
+ * Cross space compaction select 1, 2, 3, 4, 5 seq file, but file 3 and 4 are being compacted and
+ * being deleted by other inner compaction task. Cross space compaction selector should abort this
+ * task.
+ */
+ @Test
+ public void testSelectingFilesWhenSomeFilesBeingDeleted()
+ throws MetadataException, IOException, WriteProcessException, StorageEngineException,
+ InterruptedException, MergeException {
+ registerTimeseriesInMManger(5, 10, true);
+ createFiles(5, 10, 5, 1000, 0, 0, 100, 100, false, true);
+ createFiles(1, 5, 10, 4500, 500, 500, 0, 100, false, false);
+ tsFileManager.addAll(seqResources, true);
+ tsFileManager.addAll(unseqResources, false);
+
+ // seq file 3 and 4 are being compacted by inner space compaction
+ List<TsFileResource> sourceFiles = new ArrayList<>();
+ sourceFiles.add(seqResources.get(2));
+ sourceFiles.add(seqResources.get(3));
+ List<TsFileResource> targetResources =
+ CompactionFileGeneratorUtils.getInnerCompactionTargetTsFileResources(sourceFiles, true);
+ FastCompactionPerformer performer = new FastCompactionPerformer(false);
+ performer.setSourceFiles(sourceFiles);
+ performer.setTargetFiles(targetResources);
+ performer.setSummary(new CompactionTaskSummary());
+ performer.perform();
+
+ CompactionUtils.moveTargetFile(targetResources, true, COMPACTION_TEST_SG + "-" + "0");
+ CompactionUtils.combineModsInInnerCompaction(sourceFiles, targetResources.get(0));
+ tsFileManager.replace(sourceFiles, Collections.emptyList(), targetResources, 0, true);
+ CompactionUtils.deleteTsFilesInDisk(sourceFiles, COMPACTION_TEST_SG + "-" + "0");
+
+ // start selecting files and then start a cross space compaction task
+ ICrossSpaceSelector selector =
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getCrossCompactionSelector()
+ .createInstance(COMPACTION_TEST_SG, "0", 0, tsFileManager);
+ // In the process of getting the file list and starting to select files, the file list is
+ // updated (the file is deleted or the status is updated)
+ List<Pair<List<TsFileResource>, List<TsFileResource>>> selected =
+ selector.selectCrossSpaceTask(seqResources, unseqResources);
+
+ Assert.assertEquals(0, selected.size());
+ }
+
+ /**
+ * Validate seq files with file time index and device time index under this time partition after
+ * compaction.
+ */
+ @Test
+ public void testTsFileValidationWithFileTimeIndex()
+ throws MetadataException, IOException, WriteProcessException {
+ registerTimeseriesInMManger(5, 10, true);
+ createFiles(10, 10, 5, 1000, 0, 0, 100, 100, false, true);
+ createFiles(1, 5, 10, 4500, 500, 500, 0, 100, false, false);
+ tsFileManager.addAll(seqResources, true);
+ tsFileManager.addAll(unseqResources, false);
+
+ // set the end time of d1 in the first seq file to 1100
+ tsFileManager
+ .getTsFileList(true)
+ .get(0)
+ .updateEndTime(COMPACTION_TEST_SG + PATH_SEPARATOR + "d1", 1100L);
+
+ // set the end time of d1 in the second seq file to 1200
+ tsFileManager
+ .getTsFileList(true)
+ .get(1)
+ .updateStartTime(COMPACTION_TEST_SG + PATH_SEPARATOR + "d1", 1200L);
+
+ for (int i = 1; i < seqResources.size(); i++) {
+ tsFileManager.getTsFileList(true).get(i).degradeTimeIndex();
+ }
+
+ // seq file 4,5 and 6 are being compacted by inner space compaction
+ List<TsFileResource> sourceFiles = new ArrayList<>();
+ sourceFiles.add(seqResources.get(4));
+ sourceFiles.add(seqResources.get(5));
+ sourceFiles.add(seqResources.get(6));
+ FastCompactionPerformer performer = new FastCompactionPerformer(false);
+ performer.setSourceFiles(sourceFiles);
+ InnerSpaceCompactionTask innerSpaceCompactionTask =
+ new InnerSpaceCompactionTask(
+ 0, tsFileManager, sourceFiles, true, performer, new AtomicInteger(0), 0);
+ if (!CompactionUtils.validateTsFileResources(tsFileManager, COMPACTION_TEST_SG, 0)) {
+ Assert.fail("meet overlap seq files");
+ }
+ innerSpaceCompactionTask.start();
+ validateSeqFiles(true);
+ }
}