(Internet of Things Database) 是一款时序数据库管理系统，可以为用户提供数据收集、存储和分析等服务

/* * 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.engine.compaction; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.engine.compaction.execute.performer.impl.ReadPointCompactionPerformer; import org.apache.iotdb.db.engine.compaction.execute.task.CrossSpaceCompactionTask; import org.apache.iotdb.db.engine.compaction.execute.task.InnerSpaceCompactionTask; import org.apache.iotdb.db.engine.storagegroup.TsFileResource; import org.apache.iotdb.db.exception.StorageEngineException; import org.apache.iotdb.db.query.control.FileReaderManager; import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor; import org.apache.iotdb.tsfile.exception.write.WriteProcessException; import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding; import org.apache.iotdb.tsfile.read.TimeValuePair; import org.apache.iotdb.tsfile.read.common.TimeRange; import org.apache.iotdb.tsfile.write.chunk.AlignedChunkWriterImpl; import org.apache.iotdb.tsfile.write.chunk.IChunkWriter; import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter; import org.junit.After; import org.junit.Assert; 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.Map; import java.util.concurrent.atomic.AtomicInteger; import static org.apache.iotdb.db.engine.compaction.utils.TsFileGeneratorUtils.createChunkWriter; import static org.apache.iotdb.db.engine.compaction.utils.TsFileGeneratorUtils.createCompressionType; import static org.apache.iotdb.db.engine.compaction.utils.TsFileGeneratorUtils.createDataType; import static org.apache.iotdb.db.engine.compaction.utils.TsFileGeneratorUtils.createEncodingType; import static org.apache.iotdb.db.engine.compaction.utils.TsFileGeneratorUtils.createTimeseries; import static org.apache.iotdb.db.engine.compaction.utils.TsFileGeneratorUtils.testStorageGroup; import static org.apache.iotdb.db.engine.compaction.utils.TsFileGeneratorUtils.writeAlignedChunk; import static org.apache.iotdb.db.engine.compaction.utils.TsFileGeneratorUtils.writeOneAlignedPage; import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR; public class ReadPointAlignedCrossCompactionTest extends AbstractCompactionTest { private final String oldThreadName = Thread.currentThread().getName(); @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(30); TSFileDescriptor.getInstance().getConfig().setMaxDegreeOfIndexNode(3); Thread.currentThread().setName("pool-1-IoTDB-Compaction-Worker-1"); } @After public void tearDown() throws IOException, StorageEngineException { super.tearDown(); Thread.currentThread().setName(oldThreadName); for (TsFileResource tsFileResource : seqResources) { FileReaderManager.getInstance().closeFileAndRemoveReader(tsFileResource.getTsFilePath()); } for (TsFileResource tsFileResource : unseqResources) { FileReaderManager.getInstance().closeFileAndRemoveReader(tsFileResource.getTsFilePath()); } } @Test public void test1() throws MetadataException, IOException { List<PartialPath> timeserisPathList = new ArrayList<>(); List<TSDataType> tsDataTypes = new ArrayList<>(); // seq file 1 int deviceNum = 10; int measurementNum = 10; TsFileResource resource = createEmptyFileAndResource(true); try (TsFileIOWriter tsFileIOWriter = new TsFileIOWriter(resource.getTsFile())) { // write the data in device for (int deviceIndex = 0; deviceIndex < deviceNum; deviceIndex++) { tsFileIOWriter.startChunkGroup(COMPACTION_TEST_SG + PATH_SEPARATOR + "d" + deviceIndex); List<TSDataType> dataTypes = createDataType(measurementNum); List<TSEncoding> encodings = createEncodingType(measurementNum); List<CompressionType> compressionTypes = createCompressionType(measurementNum); List<Integer> measurementIndexes = new ArrayList<>(); for (int i = 0; i < measurementNum; i++) { measurementIndexes.add(i); } List<PartialPath> timeseriesPath = createTimeseries(deviceIndex, measurementIndexes, dataTypes, true); List<TimeRange> pages = new ArrayList<>(); pages.add(new TimeRange(0L, 300L)); pages.add(new TimeRange(500L, 600L)); for (IChunkWriter iChunkWriter : createChunkWriter(timeseriesPath, dataTypes, encodings, compressionTypes, true)) { writeAlignedChunk((AlignedChunkWriterImpl) iChunkWriter, tsFileIOWriter, pages, true); } tsFileIOWriter.endChunkGroup(); resource.updateStartTime(COMPACTION_TEST_SG + PATH_SEPARATOR + "d" + deviceIndex, 0); resource.updateEndTime(COMPACTION_TEST_SG + PATH_SEPARATOR + "d" + deviceIndex, 599); timeserisPathList.addAll(timeseriesPath); tsDataTypes.addAll(dataTypes); } tsFileIOWriter.endFile(); } resource.serialize(); seqResources.add(resource); // unseq file 1 deviceNum = 15; measurementNum = 5; resource = createEmptyFileAndResource(false); try (TsFileIOWriter tsFileIOWriter = new TsFileIOWriter(resource.getTsFile())) { // write the data in device for (int deviceIndex = 0; deviceIndex < deviceNum; deviceIndex++) { tsFileIOWriter.startChunkGroup(COMPACTION_TEST_SG + PATH_SEPARATOR + "d" + deviceIndex); List<TSDataType> dataTypes = createDataType(measurementNum); List<TSEncoding> encodings = createEncodingType(measurementNum); List<CompressionType> compressionTypes = createCompressionType(measurementNum); List<Integer> measurementIndexes = new ArrayList<>(); for (int i = 0; i < measurementNum; i++) { measurementIndexes.add(i); } List<PartialPath> timeseriesPath = createTimeseries(deviceIndex, measurementIndexes, dataTypes, true); List<TimeRange> pages = new ArrayList<>(); pages.add(new TimeRange(200L, 2200L)); for (IChunkWriter iChunkWriter : createChunkWriter(timeseriesPath, dataTypes, encodings, compressionTypes, true)) { writeAlignedChunk((AlignedChunkWriterImpl) iChunkWriter, tsFileIOWriter, pages, false); } tsFileIOWriter.endChunkGroup(); resource.updateStartTime(COMPACTION_TEST_SG + PATH_SEPARATOR + "d" + deviceIndex, 200); resource.updateEndTime(COMPACTION_TEST_SG + PATH_SEPARATOR + "d" + deviceIndex, 2199); timeserisPathList.addAll(timeseriesPath); tsDataTypes.addAll(dataTypes); } tsFileIOWriter.endFil