You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2022/08/11 10:33:15 UTC
[GitHub] [iotdb] choubenson commented on a diff in pull request #6884: [IOTDB-3611] Support "Modify Time Series Encoding and Compression Type" interface/command
choubenson commented on code in PR #6884:
URL: https://github.com/apache/iotdb/pull/6884#discussion_r943247625
##########
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java:
##########
@@ -3662,6 +4071,38 @@ public void setAllowCompaction(boolean allowCompaction) {
this.tsFileManager.setAllowCompaction(allowCompaction);
}
+ /** delete tsfile, copy from CompactionUtils TODO We need to rename CompactionUtils */
+ public static boolean deleteTsFile(TsFileResource seqFile, String storageGroupName) {
+ try {
+ logger.info("{} Start to delete TsFile {}", storageGroupName, seqFile.getTsFilePath());
+ FileReaderManager.getInstance().closeFileAndRemoveReader(seqFile.getTsFilePath());
+ // seqFile.setStatus(TsFileResourceStatus.DELETED);
+ seqFile.delete();
Review Comment:
How about using `seqFile.remove()` to delete TsFiles, mods file and resource file, instead of using another method to delete mods file?
##########
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java:
##########
@@ -2069,6 +2249,235 @@ private List<TsFileResource> getFileResourceListForQuery(
return tsfileResourcesForQuery;
}
+ /**
+ * alter timeseries encoding & compressionType<br>
+ * 1、flush and close tsfile<br>
+ * 2、locks<br>
+ * 3、write temp tsfiles<br>
+ * 4、unregister old tsfiles and release locks<br>
+ * 5、rename temp tsfiles<br>
+ * 6、register tsfiles<br>
+ */
+ public void alter(
+ PartialPath fullPath, TSEncoding curEncoding, CompressionType curCompressionType)
+ throws IOException {
+
+ final String logKey =
+ logicalStorageGroupName + "-" + dataRegionId + "-" + fullPath.getFullPath();
+ // If there are still some old version tsfiles, the delete won't succeeded.
+ if (upgradeFileCount.get() != 0) {
+ throw new IOException(
+ "Alter failed. " + "Please do not delete until the old files upgraded.");
+ }
+ if (SettleService.getINSTANCE().getFilesToBeSettledCount().get() != 0) {
+ throw new IOException("Alter failed. " + "Please do not delete until the old files settled.");
+ }
+ logger.info("[alter timeseries] {} syncCloseAllWorkingTsFileProcessors", logKey);
+ // flush & close
+ syncCloseAllWorkingTsFileProcessors();
+ logger.info("[alter timeseries] writeLock");
+ // wait lock
+ if (!tsFileManager.rewriteLockWithTimeout(
+ IoTDBDescriptor.getInstance().getConfig().getRewriteLockWaitTimeoutInMS())) {
+ throw new IOException(
+ "Alter failed. "
+ + "Other file rewriting operations are in progress, please do it later.");
+ }
+ // recover log
+ File logFile =
+ SystemFileFactory.INSTANCE.getFile(storageGroupSysDir, AlteringLogger.ALTERING_LOG_NAME);
+ if (logFile.exists()) {
+ logger.info("[alter timeseries] {} rewriteUnlock", logKey);
+ tsFileManager.rewriteUnlock();
+ throw new IOException(
+ "Alter failed. "
+ + "alter.log detected, other alter operations may be running, please do it later.");
+ }
+ // rewrite target tsfiles
+ boolean done = false;
+ try (AlteringLogger alteringLogger = new AlteringLogger(logFile)) {
+ Set<Long> timePartitions = tsFileManager.getTimePartitions();
+ // Record the ALTER process for server restart recovery
+ alteringLogger.logHeader(fullPath, curEncoding, curCompressionType, timePartitions);
+ for (Long timePartition : timePartitions) {
+ logger.info("[alter timeseries] {} alterDataInTsFiles seq({})", logKey, timePartition);
+ try {
+ rewriteDataInTsFiles(
+ tsFileManager.getSequenceListByTimePartition(timePartition),
+ fullPath,
+ curEncoding,
+ curCompressionType,
+ timePartition,
+ true,
+ alteringLogger,
+ null,
+ logKey);
+ logger.info("[alter timeseries] {} alterDataInTsFiles unseq({})", logKey, timePartition);
+ rewriteDataInTsFiles(
+ tsFileManager.getUnsequenceListByTimePartition(timePartition),
+ fullPath,
+ curEncoding,
+ curCompressionType,
+ timePartition,
+ false,
+ alteringLogger,
+ null,
+ logKey);
+ } catch (IOException e) {
+ /**
+ * TODO If an exception occurs in the operation of a single tsfile, you need to terminate
+ * the operation and return 1. The modified schema and rewritten tsfile need to provide
+ * rollback commands 2. The tsfile that has not been rewritten needs to provide a recovery
+ * command
+ */
+ logger.error(
+ "[alter timeseries] " + logKey + " timePartition " + timePartition + " error", e);
+ throw e;
+ }
+ }
+ done = true;
+ } catch (Exception e) {
+ logger.error("[alter timeseries] " + logKey + " error", e);
+ throw e;
+ } finally {
+ logger.info("[alter timeseries] {} rewriteUnlock", logKey);
+ tsFileManager.rewriteUnlock();
+ // The process is complete and the logFile is deleted
+ if (done && logFile.exists()) {
+ FileUtils.delete(logFile);
+ }
+ }
+ }
+
+ private void rewriteDataInTsFiles(
+ List<TsFileResource> tsFileList,
+ PartialPath fullPath,
+ TSEncoding curEncoding,
+ CompressionType curCompressionType,
+ long timePartition,
+ boolean sequence,
+ AlteringLogger alteringLogger,
+ Set<TsFileIdentifier> undoneFiles,
+ String logKey)
+ throws IOException {
+
+ if (tsFileList == null || tsFileList.isEmpty()) {
+ return;
+ }
+ // log timePartition start
+ alteringLogger.startTimePartition(tsFileList, timePartition, sequence);
+ for (TsFileResource tsFileResource : tsFileList) {
+ if (!findUndoneResourcesAndRemove(tsFileResource, undoneFiles)) {
+ continue;
+ }
+ if (tsFileResource == null || !tsFileResource.isClosed()) {
+ return;
+ }
+ try {
+ logger.info(
+ "[alter timeseries] {} rewriteDataInTsFile:{}, fileSize:{} start",
+ logKey,
+ tsFileResource.getTsFilePath(),
+ tsFileResource.getTsFileSize());
+ // Generate the target tsFileResource
+ TsFileResource targetTsFileResource =
+ TsFileNameGenerator.generateNewAlterTsFileResource(tsFileResource);
+ // Data is read from the.tsfile file, re-encoded, compressed, and written to the .alter file
+ TsFileRewriteExcutor tsFileRewriteExcutor =
+ new TsFileRewriteExcutor(
+ tsFileResource,
+ targetTsFileResource,
+ fullPath,
+ curEncoding,
+ curCompressionType,
+ timePartition,
+ sequence);
+ tsFileRewriteExcutor.execute();
+ // .tsfile->.alter.old .alter->.tsfile
+ logger.debug("[alter timeseries] {} move tsfile", logKey);
+ tsFileResource.moveTsFile(TSFILE_SUFFIX, ALTER_OLD_TMP_FILE_SUFFIX);
+ targetTsFileResource.moveTsFile(IoTDBConstant.ALTER_TMP_FILE_SUFFIX, TSFILE_SUFFIX);
+ // replace
+ logger.debug("[alter timeseries] {} replace tsfile", logKey);
+ if (sequence) {
+ tsFileManager.replace(
+ new ArrayList<>(Collections.singletonList(tsFileResource)),
+ Collections.emptyList(),
+ new ArrayList<>(Collections.singletonList(targetTsFileResource)),
+ timePartition,
+ true);
+ } else {
+ tsFileManager.replace(
+ Collections.emptyList(),
+ new ArrayList<>(Collections.singletonList(tsFileResource)),
+ new ArrayList<>(Collections.singletonList(targetTsFileResource)),
Review Comment:
remove `new ArrayList<>()`
##########
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java:
##########
@@ -3662,6 +4071,38 @@ public void setAllowCompaction(boolean allowCompaction) {
this.tsFileManager.setAllowCompaction(allowCompaction);
}
+ /** delete tsfile, copy from CompactionUtils TODO We need to rename CompactionUtils */
+ public static boolean deleteTsFile(TsFileResource seqFile, String storageGroupName) {
+ try {
+ logger.info("{} Start to delete TsFile {}", storageGroupName, seqFile.getTsFilePath());
+ FileReaderManager.getInstance().closeFileAndRemoveReader(seqFile.getTsFilePath());
+ // seqFile.setStatus(TsFileResourceStatus.DELETED);
+ seqFile.delete();
+ } catch (IOException e) {
+ logger.error(e.getMessage(), e);
+ return false;
+ }
+ return true;
+ }
+
+ /** Delete all modification files for source files TODO We need to rename CompactionUtils */
+ public static void deleteModificationForSourceFile(
+ Collection<TsFileResource> sourceFiles, String storageGroupName) throws IOException {
+ logger.info("{} Start to delete modifications of source files", storageGroupName);
+ for (TsFileResource tsFileResource : sourceFiles) {
+ ModificationFile compactionModificationFile =
+ ModificationFile.getCompactionMods(tsFileResource);
+ if (compactionModificationFile.exists()) {
+ compactionModificationFile.remove();
+ }
Review Comment:
It's necessary to delete compaction mods file, because compaction mods file is only produced during compaction.
##########
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java:
##########
@@ -2069,6 +2249,235 @@ private List<TsFileResource> getFileResourceListForQuery(
return tsfileResourcesForQuery;
}
+ /**
+ * alter timeseries encoding & compressionType<br>
+ * 1、flush and close tsfile<br>
+ * 2、locks<br>
+ * 3、write temp tsfiles<br>
+ * 4、unregister old tsfiles and release locks<br>
+ * 5、rename temp tsfiles<br>
+ * 6、register tsfiles<br>
+ */
+ public void alter(
+ PartialPath fullPath, TSEncoding curEncoding, CompressionType curCompressionType)
+ throws IOException {
+
+ final String logKey =
+ logicalStorageGroupName + "-" + dataRegionId + "-" + fullPath.getFullPath();
+ // If there are still some old version tsfiles, the delete won't succeeded.
+ if (upgradeFileCount.get() != 0) {
+ throw new IOException(
+ "Alter failed. " + "Please do not delete until the old files upgraded.");
+ }
+ if (SettleService.getINSTANCE().getFilesToBeSettledCount().get() != 0) {
+ throw new IOException("Alter failed. " + "Please do not delete until the old files settled.");
Review Comment:
Change to `throw new IOException("Alter failed. Please do not alter until the old files settled.");`
##########
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java:
##########
@@ -2069,6 +2249,235 @@ private List<TsFileResource> getFileResourceListForQuery(
return tsfileResourcesForQuery;
}
+ /**
+ * alter timeseries encoding & compressionType<br>
+ * 1、flush and close tsfile<br>
+ * 2、locks<br>
+ * 3、write temp tsfiles<br>
+ * 4、unregister old tsfiles and release locks<br>
+ * 5、rename temp tsfiles<br>
+ * 6、register tsfiles<br>
+ */
+ public void alter(
+ PartialPath fullPath, TSEncoding curEncoding, CompressionType curCompressionType)
+ throws IOException {
+
+ final String logKey =
+ logicalStorageGroupName + "-" + dataRegionId + "-" + fullPath.getFullPath();
+ // If there are still some old version tsfiles, the delete won't succeeded.
+ if (upgradeFileCount.get() != 0) {
+ throw new IOException(
+ "Alter failed. " + "Please do not delete until the old files upgraded.");
Review Comment:
Change to `throw new IOException("Alter failed. Please do not alter until the old files upgraded.");`
##########
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java:
##########
@@ -2069,6 +2249,235 @@ private List<TsFileResource> getFileResourceListForQuery(
return tsfileResourcesForQuery;
}
+ /**
+ * alter timeseries encoding & compressionType<br>
+ * 1、flush and close tsfile<br>
+ * 2、locks<br>
+ * 3、write temp tsfiles<br>
+ * 4、unregister old tsfiles and release locks<br>
+ * 5、rename temp tsfiles<br>
+ * 6、register tsfiles<br>
+ */
+ public void alter(
+ PartialPath fullPath, TSEncoding curEncoding, CompressionType curCompressionType)
+ throws IOException {
+
+ final String logKey =
+ logicalStorageGroupName + "-" + dataRegionId + "-" + fullPath.getFullPath();
+ // If there are still some old version tsfiles, the delete won't succeeded.
+ if (upgradeFileCount.get() != 0) {
+ throw new IOException(
+ "Alter failed. " + "Please do not delete until the old files upgraded.");
+ }
+ if (SettleService.getINSTANCE().getFilesToBeSettledCount().get() != 0) {
+ throw new IOException("Alter failed. " + "Please do not delete until the old files settled.");
+ }
+ logger.info("[alter timeseries] {} syncCloseAllWorkingTsFileProcessors", logKey);
+ // flush & close
+ syncCloseAllWorkingTsFileProcessors();
+ logger.info("[alter timeseries] writeLock");
+ // wait lock
+ if (!tsFileManager.rewriteLockWithTimeout(
+ IoTDBDescriptor.getInstance().getConfig().getRewriteLockWaitTimeoutInMS())) {
+ throw new IOException(
+ "Alter failed. "
+ + "Other file rewriting operations are in progress, please do it later.");
+ }
+ // recover log
+ File logFile =
+ SystemFileFactory.INSTANCE.getFile(storageGroupSysDir, AlteringLogger.ALTERING_LOG_NAME);
+ if (logFile.exists()) {
+ logger.info("[alter timeseries] {} rewriteUnlock", logKey);
+ tsFileManager.rewriteUnlock();
+ throw new IOException(
+ "Alter failed. "
+ + "alter.log detected, other alter operations may be running, please do it later.");
+ }
+ // rewrite target tsfiles
+ boolean done = false;
+ try (AlteringLogger alteringLogger = new AlteringLogger(logFile)) {
+ Set<Long> timePartitions = tsFileManager.getTimePartitions();
+ // Record the ALTER process for server restart recovery
+ alteringLogger.logHeader(fullPath, curEncoding, curCompressionType, timePartitions);
+ for (Long timePartition : timePartitions) {
+ logger.info("[alter timeseries] {} alterDataInTsFiles seq({})", logKey, timePartition);
+ try {
+ rewriteDataInTsFiles(
+ tsFileManager.getSequenceListByTimePartition(timePartition),
+ fullPath,
+ curEncoding,
+ curCompressionType,
+ timePartition,
+ true,
+ alteringLogger,
+ null,
+ logKey);
+ logger.info("[alter timeseries] {} alterDataInTsFiles unseq({})", logKey, timePartition);
+ rewriteDataInTsFiles(
+ tsFileManager.getUnsequenceListByTimePartition(timePartition),
+ fullPath,
+ curEncoding,
+ curCompressionType,
+ timePartition,
+ false,
+ alteringLogger,
+ null,
+ logKey);
+ } catch (IOException e) {
+ /**
+ * TODO If an exception occurs in the operation of a single tsfile, you need to terminate
+ * the operation and return 1. The modified schema and rewritten tsfile need to provide
+ * rollback commands 2. The tsfile that has not been rewritten needs to provide a recovery
+ * command
+ */
+ logger.error(
+ "[alter timeseries] " + logKey + " timePartition " + timePartition + " error", e);
+ throw e;
+ }
+ }
+ done = true;
+ } catch (Exception e) {
+ logger.error("[alter timeseries] " + logKey + " error", e);
+ throw e;
+ } finally {
+ logger.info("[alter timeseries] {} rewriteUnlock", logKey);
+ tsFileManager.rewriteUnlock();
+ // The process is complete and the logFile is deleted
+ if (done && logFile.exists()) {
+ FileUtils.delete(logFile);
+ }
+ }
+ }
+
+ private void rewriteDataInTsFiles(
+ List<TsFileResource> tsFileList,
+ PartialPath fullPath,
+ TSEncoding curEncoding,
+ CompressionType curCompressionType,
+ long timePartition,
+ boolean sequence,
+ AlteringLogger alteringLogger,
+ Set<TsFileIdentifier> undoneFiles,
+ String logKey)
+ throws IOException {
+
+ if (tsFileList == null || tsFileList.isEmpty()) {
+ return;
+ }
+ // log timePartition start
+ alteringLogger.startTimePartition(tsFileList, timePartition, sequence);
+ for (TsFileResource tsFileResource : tsFileList) {
+ if (!findUndoneResourcesAndRemove(tsFileResource, undoneFiles)) {
+ continue;
+ }
+ if (tsFileResource == null || !tsFileResource.isClosed()) {
+ return;
+ }
+ try {
+ logger.info(
+ "[alter timeseries] {} rewriteDataInTsFile:{}, fileSize:{} start",
+ logKey,
+ tsFileResource.getTsFilePath(),
+ tsFileResource.getTsFileSize());
+ // Generate the target tsFileResource
+ TsFileResource targetTsFileResource =
+ TsFileNameGenerator.generateNewAlterTsFileResource(tsFileResource);
+ // Data is read from the.tsfile file, re-encoded, compressed, and written to the .alter file
+ TsFileRewriteExcutor tsFileRewriteExcutor =
+ new TsFileRewriteExcutor(
+ tsFileResource,
+ targetTsFileResource,
+ fullPath,
+ curEncoding,
+ curCompressionType,
+ timePartition,
+ sequence);
+ tsFileRewriteExcutor.execute();
+ // .tsfile->.alter.old .alter->.tsfile
+ logger.debug("[alter timeseries] {} move tsfile", logKey);
+ tsFileResource.moveTsFile(TSFILE_SUFFIX, ALTER_OLD_TMP_FILE_SUFFIX);
+ targetTsFileResource.moveTsFile(IoTDBConstant.ALTER_TMP_FILE_SUFFIX, TSFILE_SUFFIX);
+ // replace
+ logger.debug("[alter timeseries] {} replace tsfile", logKey);
+ if (sequence) {
+ tsFileManager.replace(
+ new ArrayList<>(Collections.singletonList(tsFileResource)),
+ Collections.emptyList(),
+ new ArrayList<>(Collections.singletonList(targetTsFileResource)),
Review Comment:
remove `new ArrayList<>()`
##########
server/src/main/java/org/apache/iotdb/db/engine/alter/TsFileRewriteExcutor.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.alter;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.CommonUtils;
+import org.apache.iotdb.tsfile.file.header.ChunkHeader;
+import org.apache.iotdb.tsfile.file.metadata.AlignedChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.IChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.read.TsFileAlignedSeriesReaderIterator;
+import org.apache.iotdb.tsfile.read.TsFileDeviceIterator;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.Chunk;
+import org.apache.iotdb.tsfile.read.common.IBatchDataIterator;
+import org.apache.iotdb.tsfile.read.reader.IChunkReader;
+import org.apache.iotdb.tsfile.read.reader.IPointReader;
+import org.apache.iotdb.tsfile.read.reader.chunk.AlignedChunkReader;
+import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReader;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
+import org.apache.iotdb.tsfile.write.chunk.AlignedChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.chunk.ChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** This class is used to rewrite one tsfile with current encoding & compressionType. */
+public class TsFileRewriteExcutor {
+
+ private static final Logger logger = LoggerFactory.getLogger(TsFileRewriteExcutor.class);
+
+ private final TsFileResource tsFileResource;
+ private final TsFileResource targetTsFileResource;
+ private final PartialPath fullPath;
+ private final TSEncoding curEncoding;
+ private final CompressionType curCompressionType;
+ private long timePartition;
+ private boolean sequence;
+ // record the min time and max time to update the target resource
+ private long minStartTimestamp = Long.MAX_VALUE;
+ private long maxEndTimestamp = Long.MIN_VALUE;
+
+ public TsFileRewriteExcutor(
+ TsFileResource tsFileResource,
+ TsFileResource targetTsFileResource,
+ PartialPath fullPath,
+ TSEncoding curEncoding,
+ CompressionType curCompressionType,
+ long timePartition,
+ boolean sequence) {
+ this.tsFileResource = tsFileResource;
+ this.targetTsFileResource = targetTsFileResource;
+ this.fullPath = fullPath;
+ this.curEncoding = curEncoding;
+ this.curCompressionType = curCompressionType;
+ this.timePartition = timePartition;
+ this.sequence = sequence;
+ }
+
+ /** This function execute the rewrite task */
+ public void execute() throws IOException {
+
+ tsFileResource.tryReadLock();
+ try (TsFileSequenceReader reader = new TsFileSequenceReader(tsFileResource.getTsFilePath());
+ TsFileIOWriter writer = new TsFileIOWriter(targetTsFileResource.getTsFile())) {
+ // read devices
+ TsFileDeviceIterator deviceIterator = reader.getAllDevicesIteratorWithIsAligned();
+ while (deviceIterator.hasNext()) {
+ minStartTimestamp = Long.MAX_VALUE;
+ maxEndTimestamp = Long.MIN_VALUE;
+ Pair<String, Boolean> deviceInfo = deviceIterator.next();
+ String device = deviceInfo.left;
+ boolean aligned = deviceInfo.right;
+ // write chunkGroup header
+ writer.startChunkGroup(device);
+ boolean isTargetDevice = fullPath.getDevice().equals(device);
+ String targetMeasurement = fullPath.getMeasurement();
+ // write chunk & page data
+ if (aligned) {
+ rewriteAlgined(reader, writer, device, isTargetDevice, targetMeasurement);
+ } else {
+ rewriteNotAligned(device, reader, writer, targetMeasurement, isTargetDevice);
+ }
+ // chunkGroup end
+ writer.endChunkGroup();
+
+ targetTsFileResource.updateStartTime(device, minStartTimestamp);
+ targetTsFileResource.updateEndTime(device, maxEndTimestamp);
+ }
+
+ targetTsFileResource.updatePlanIndexes(tsFileResource);
+ // write index,bloom,footer, end file
+ writer.endFile();
+ targetTsFileResource.close();
+ } finally {
+ tsFileResource.readUnlock();
+ }
+ }
+
+ private void rewriteAlgined(
+ TsFileSequenceReader reader,
+ TsFileIOWriter writer,
+ String device,
+ boolean isTargetDevice,
+ String targetMeasurement)
+ throws IOException {
+ List<AlignedChunkMetadata> alignedChunkMetadatas = reader.getAlignedChunkMetadata(device);
+ if (alignedChunkMetadatas == null || alignedChunkMetadatas.isEmpty()) {
+ logger.warn("[alter timeseries] device({}) alignedChunkMetadatas is null", device);
+ return;
+ }
+ // TODO To be optimized: Non-target modification measurements are directly written to data
+ List<IMeasurementSchema> schemaList =
+ collectSchemaList(alignedChunkMetadatas, reader, targetMeasurement, isTargetDevice);
+ List<IMeasurementSchema> schemaOldList =
+ collectSchemaList(alignedChunkMetadatas, reader, targetMeasurement, false);
Review Comment:
These two steps result in two more IO operations before starting to rewrite, which can be combined into one step.
##########
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java:
##########
@@ -2069,6 +2249,235 @@ private List<TsFileResource> getFileResourceListForQuery(
return tsfileResourcesForQuery;
}
+ /**
+ * alter timeseries encoding & compressionType<br>
+ * 1、flush and close tsfile<br>
+ * 2、locks<br>
+ * 3、write temp tsfiles<br>
+ * 4、unregister old tsfiles and release locks<br>
+ * 5、rename temp tsfiles<br>
+ * 6、register tsfiles<br>
+ */
+ public void alter(
+ PartialPath fullPath, TSEncoding curEncoding, CompressionType curCompressionType)
+ throws IOException {
+
+ final String logKey =
+ logicalStorageGroupName + "-" + dataRegionId + "-" + fullPath.getFullPath();
+ // If there are still some old version tsfiles, the delete won't succeeded.
+ if (upgradeFileCount.get() != 0) {
+ throw new IOException(
+ "Alter failed. " + "Please do not delete until the old files upgraded.");
+ }
+ if (SettleService.getINSTANCE().getFilesToBeSettledCount().get() != 0) {
+ throw new IOException("Alter failed. " + "Please do not delete until the old files settled.");
+ }
+ logger.info("[alter timeseries] {} syncCloseAllWorkingTsFileProcessors", logKey);
+ // flush & close
+ syncCloseAllWorkingTsFileProcessors();
Review Comment:
Add dataRegion write lock before closing all working processors to avoid writing operation from users later.
##########
server/src/main/java/org/apache/iotdb/db/engine/alter/TsFileRewriteExcutor.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.alter;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.CommonUtils;
+import org.apache.iotdb.tsfile.file.header.ChunkHeader;
+import org.apache.iotdb.tsfile.file.metadata.AlignedChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.IChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.read.TsFileAlignedSeriesReaderIterator;
+import org.apache.iotdb.tsfile.read.TsFileDeviceIterator;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.Chunk;
+import org.apache.iotdb.tsfile.read.common.IBatchDataIterator;
+import org.apache.iotdb.tsfile.read.reader.IChunkReader;
+import org.apache.iotdb.tsfile.read.reader.IPointReader;
+import org.apache.iotdb.tsfile.read.reader.chunk.AlignedChunkReader;
+import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReader;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
+import org.apache.iotdb.tsfile.write.chunk.AlignedChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.chunk.ChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** This class is used to rewrite one tsfile with current encoding & compressionType. */
+public class TsFileRewriteExcutor {
+
+ private static final Logger logger = LoggerFactory.getLogger(TsFileRewriteExcutor.class);
+
+ private final TsFileResource tsFileResource;
+ private final TsFileResource targetTsFileResource;
+ private final PartialPath fullPath;
+ private final TSEncoding curEncoding;
+ private final CompressionType curCompressionType;
+ private long timePartition;
+ private boolean sequence;
+ // record the min time and max time to update the target resource
+ private long minStartTimestamp = Long.MAX_VALUE;
+ private long maxEndTimestamp = Long.MIN_VALUE;
+
+ public TsFileRewriteExcutor(
+ TsFileResource tsFileResource,
+ TsFileResource targetTsFileResource,
+ PartialPath fullPath,
+ TSEncoding curEncoding,
+ CompressionType curCompressionType,
+ long timePartition,
+ boolean sequence) {
+ this.tsFileResource = tsFileResource;
+ this.targetTsFileResource = targetTsFileResource;
+ this.fullPath = fullPath;
+ this.curEncoding = curEncoding;
+ this.curCompressionType = curCompressionType;
+ this.timePartition = timePartition;
+ this.sequence = sequence;
+ }
+
+ /** This function execute the rewrite task */
+ public void execute() throws IOException {
+
+ tsFileResource.tryReadLock();
Review Comment:
If the resource is holding the write lock by another thread, this method will not get the read lock. Use `readLock()` instead.
##########
server/src/main/java/org/apache/iotdb/db/engine/alter/TsFileRewriteExcutor.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.alter;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.CommonUtils;
+import org.apache.iotdb.tsfile.file.header.ChunkHeader;
+import org.apache.iotdb.tsfile.file.metadata.AlignedChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.IChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.read.TsFileAlignedSeriesReaderIterator;
+import org.apache.iotdb.tsfile.read.TsFileDeviceIterator;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.Chunk;
+import org.apache.iotdb.tsfile.read.common.IBatchDataIterator;
+import org.apache.iotdb.tsfile.read.reader.IChunkReader;
+import org.apache.iotdb.tsfile.read.reader.IPointReader;
+import org.apache.iotdb.tsfile.read.reader.chunk.AlignedChunkReader;
+import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReader;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
+import org.apache.iotdb.tsfile.write.chunk.AlignedChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.chunk.ChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** This class is used to rewrite one tsfile with current encoding & compressionType. */
+public class TsFileRewriteExcutor {
+
+ private static final Logger logger = LoggerFactory.getLogger(TsFileRewriteExcutor.class);
+
+ private final TsFileResource tsFileResource;
+ private final TsFileResource targetTsFileResource;
+ private final PartialPath fullPath;
+ private final TSEncoding curEncoding;
+ private final CompressionType curCompressionType;
+ private long timePartition;
+ private boolean sequence;
+ // record the min time and max time to update the target resource
+ private long minStartTimestamp = Long.MAX_VALUE;
+ private long maxEndTimestamp = Long.MIN_VALUE;
+
+ public TsFileRewriteExcutor(
+ TsFileResource tsFileResource,
+ TsFileResource targetTsFileResource,
+ PartialPath fullPath,
+ TSEncoding curEncoding,
+ CompressionType curCompressionType,
+ long timePartition,
+ boolean sequence) {
+ this.tsFileResource = tsFileResource;
+ this.targetTsFileResource = targetTsFileResource;
+ this.fullPath = fullPath;
+ this.curEncoding = curEncoding;
+ this.curCompressionType = curCompressionType;
+ this.timePartition = timePartition;
+ this.sequence = sequence;
+ }
+
+ /** This function execute the rewrite task */
+ public void execute() throws IOException {
+
+ tsFileResource.tryReadLock();
+ try (TsFileSequenceReader reader = new TsFileSequenceReader(tsFileResource.getTsFilePath());
+ TsFileIOWriter writer = new TsFileIOWriter(targetTsFileResource.getTsFile())) {
+ // read devices
+ TsFileDeviceIterator deviceIterator = reader.getAllDevicesIteratorWithIsAligned();
+ while (deviceIterator.hasNext()) {
+ minStartTimestamp = Long.MAX_VALUE;
+ maxEndTimestamp = Long.MIN_VALUE;
+ Pair<String, Boolean> deviceInfo = deviceIterator.next();
+ String device = deviceInfo.left;
+ boolean aligned = deviceInfo.right;
+ // write chunkGroup header
+ writer.startChunkGroup(device);
+ boolean isTargetDevice = fullPath.getDevice().equals(device);
+ String targetMeasurement = fullPath.getMeasurement();
+ // write chunk & page data
+ if (aligned) {
+ rewriteAlgined(reader, writer, device, isTargetDevice, targetMeasurement);
+ } else {
+ rewriteNotAligned(device, reader, writer, targetMeasurement, isTargetDevice);
+ }
+ // chunkGroup end
+ writer.endChunkGroup();
+
+ targetTsFileResource.updateStartTime(device, minStartTimestamp);
+ targetTsFileResource.updateEndTime(device, maxEndTimestamp);
+ }
+
+ targetTsFileResource.updatePlanIndexes(tsFileResource);
+ // write index,bloom,footer, end file
+ writer.endFile();
+ targetTsFileResource.close();
+ } finally {
+ tsFileResource.readUnlock();
+ }
+ }
+
+ private void rewriteAlgined(
+ TsFileSequenceReader reader,
+ TsFileIOWriter writer,
+ String device,
+ boolean isTargetDevice,
+ String targetMeasurement)
+ throws IOException {
+ List<AlignedChunkMetadata> alignedChunkMetadatas = reader.getAlignedChunkMetadata(device);
+ if (alignedChunkMetadatas == null || alignedChunkMetadatas.isEmpty()) {
+ logger.warn("[alter timeseries] device({}) alignedChunkMetadatas is null", device);
+ return;
+ }
+ // TODO To be optimized: Non-target modification measurements are directly written to data
+ List<IMeasurementSchema> schemaList =
+ collectSchemaList(alignedChunkMetadatas, reader, targetMeasurement, isTargetDevice);
+ List<IMeasurementSchema> schemaOldList =
+ collectSchemaList(alignedChunkMetadatas, reader, targetMeasurement, false);
+ AlignedChunkWriterImpl chunkWriter = new AlignedChunkWriterImpl(schemaList);
+ TsFileAlignedSeriesReaderIterator readerIterator =
+ new TsFileAlignedSeriesReaderIterator(reader, alignedChunkMetadatas, schemaOldList);
+
+ while (readerIterator.hasNext()) {
+ Pair<AlignedChunkReader, Long> chunkReaderAndChunkSize = readerIterator.nextReader();
+ AlignedChunkReader chunkReader = chunkReaderAndChunkSize.left;
+ while (chunkReader.hasNextSatisfiedPage()) {
+ IBatchDataIterator batchDataIterator = chunkReader.nextPageData().getBatchDataIterator();
+ while (batchDataIterator.hasNext()) {
+ TsPrimitiveType[] pointsData = (TsPrimitiveType[]) batchDataIterator.currentValue();
+ long time = batchDataIterator.currentTime();
+ chunkWriter.write(time, pointsData);
+ targetTsFileResource.updateStartTime(device, time);
+ targetTsFileResource.updateEndTime(device, time);
+ batchDataIterator.next();
+ }
+ }
+ }
+ chunkWriter.writeToFileWriter(writer);
+ }
+
+ protected List<IMeasurementSchema> collectSchemaList(
+ List<AlignedChunkMetadata> alignedChunkMetadatas,
+ TsFileSequenceReader reader,
+ String targetMeasurement,
+ boolean isTargetDevice)
+ throws IOException {
+
+ Set<MeasurementSchema> schemaSet = new HashSet<>();
+ Set<String> measurementSet = new HashSet<>();
Review Comment:
Use ArrayList to avoid type conversion later.
##########
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java:
##########
@@ -2069,6 +2249,235 @@ private List<TsFileResource> getFileResourceListForQuery(
return tsfileResourcesForQuery;
}
+ /**
+ * alter timeseries encoding & compressionType<br>
+ * 1、flush and close tsfile<br>
+ * 2、locks<br>
+ * 3、write temp tsfiles<br>
+ * 4、unregister old tsfiles and release locks<br>
+ * 5、rename temp tsfiles<br>
+ * 6、register tsfiles<br>
+ */
+ public void alter(
+ PartialPath fullPath, TSEncoding curEncoding, CompressionType curCompressionType)
+ throws IOException {
+
+ final String logKey =
+ logicalStorageGroupName + "-" + dataRegionId + "-" + fullPath.getFullPath();
+ // If there are still some old version tsfiles, the delete won't succeeded.
+ if (upgradeFileCount.get() != 0) {
+ throw new IOException(
+ "Alter failed. " + "Please do not delete until the old files upgraded.");
+ }
+ if (SettleService.getINSTANCE().getFilesToBeSettledCount().get() != 0) {
+ throw new IOException("Alter failed. " + "Please do not delete until the old files settled.");
+ }
+ logger.info("[alter timeseries] {} syncCloseAllWorkingTsFileProcessors", logKey);
+ // flush & close
+ syncCloseAllWorkingTsFileProcessors();
+ logger.info("[alter timeseries] writeLock");
+ // wait lock
+ if (!tsFileManager.rewriteLockWithTimeout(
+ IoTDBDescriptor.getInstance().getConfig().getRewriteLockWaitTimeoutInMS())) {
+ throw new IOException(
+ "Alter failed. "
+ + "Other file rewriting operations are in progress, please do it later.");
+ }
+ // recover log
+ File logFile =
+ SystemFileFactory.INSTANCE.getFile(storageGroupSysDir, AlteringLogger.ALTERING_LOG_NAME);
+ if (logFile.exists()) {
+ logger.info("[alter timeseries] {} rewriteUnlock", logKey);
+ tsFileManager.rewriteUnlock();
+ throw new IOException(
+ "Alter failed. "
+ + "alter.log detected, other alter operations may be running, please do it later.");
+ }
+ // rewrite target tsfiles
+ boolean done = false;
+ try (AlteringLogger alteringLogger = new AlteringLogger(logFile)) {
+ Set<Long> timePartitions = tsFileManager.getTimePartitions();
+ // Record the ALTER process for server restart recovery
+ alteringLogger.logHeader(fullPath, curEncoding, curCompressionType, timePartitions);
+ for (Long timePartition : timePartitions) {
+ logger.info("[alter timeseries] {} alterDataInTsFiles seq({})", logKey, timePartition);
+ try {
+ rewriteDataInTsFiles(
+ tsFileManager.getSequenceListByTimePartition(timePartition),
+ fullPath,
+ curEncoding,
+ curCompressionType,
+ timePartition,
+ true,
+ alteringLogger,
+ null,
+ logKey);
+ logger.info("[alter timeseries] {} alterDataInTsFiles unseq({})", logKey, timePartition);
+ rewriteDataInTsFiles(
+ tsFileManager.getUnsequenceListByTimePartition(timePartition),
+ fullPath,
+ curEncoding,
+ curCompressionType,
+ timePartition,
+ false,
+ alteringLogger,
+ null,
+ logKey);
+ } catch (IOException e) {
+ /**
+ * TODO If an exception occurs in the operation of a single tsfile, you need to terminate
+ * the operation and return 1. The modified schema and rewritten tsfile need to provide
+ * rollback commands 2. The tsfile that has not been rewritten needs to provide a recovery
+ * command
+ */
+ logger.error(
+ "[alter timeseries] " + logKey + " timePartition " + timePartition + " error", e);
+ throw e;
+ }
+ }
+ done = true;
+ } catch (Exception e) {
+ logger.error("[alter timeseries] " + logKey + " error", e);
+ throw e;
+ } finally {
+ logger.info("[alter timeseries] {} rewriteUnlock", logKey);
+ tsFileManager.rewriteUnlock();
+ // The process is complete and the logFile is deleted
+ if (done && logFile.exists()) {
+ FileUtils.delete(logFile);
+ }
+ }
+ }
+
+ private void rewriteDataInTsFiles(
+ List<TsFileResource> tsFileList,
+ PartialPath fullPath,
+ TSEncoding curEncoding,
+ CompressionType curCompressionType,
+ long timePartition,
+ boolean sequence,
+ AlteringLogger alteringLogger,
+ Set<TsFileIdentifier> undoneFiles,
+ String logKey)
+ throws IOException {
+
+ if (tsFileList == null || tsFileList.isEmpty()) {
+ return;
+ }
+ // log timePartition start
+ alteringLogger.startTimePartition(tsFileList, timePartition, sequence);
+ for (TsFileResource tsFileResource : tsFileList) {
Review Comment:
It seems that all TsFiles will be rewritten no matter it has corresponding device or not. Add a judgment to filter out tsfiles that do not contain the corresponding device.
##########
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java:
##########
@@ -2069,6 +2249,235 @@ private List<TsFileResource> getFileResourceListForQuery(
return tsfileResourcesForQuery;
}
+ /**
+ * alter timeseries encoding & compressionType<br>
+ * 1、flush and close tsfile<br>
+ * 2、locks<br>
+ * 3、write temp tsfiles<br>
+ * 4、unregister old tsfiles and release locks<br>
+ * 5、rename temp tsfiles<br>
+ * 6、register tsfiles<br>
+ */
+ public void alter(
+ PartialPath fullPath, TSEncoding curEncoding, CompressionType curCompressionType)
+ throws IOException {
+
+ final String logKey =
+ logicalStorageGroupName + "-" + dataRegionId + "-" + fullPath.getFullPath();
+ // If there are still some old version tsfiles, the delete won't succeeded.
+ if (upgradeFileCount.get() != 0) {
+ throw new IOException(
+ "Alter failed. " + "Please do not delete until the old files upgraded.");
+ }
+ if (SettleService.getINSTANCE().getFilesToBeSettledCount().get() != 0) {
+ throw new IOException("Alter failed. " + "Please do not delete until the old files settled.");
+ }
+ logger.info("[alter timeseries] {} syncCloseAllWorkingTsFileProcessors", logKey);
+ // flush & close
+ syncCloseAllWorkingTsFileProcessors();
+ logger.info("[alter timeseries] writeLock");
+ // wait lock
+ if (!tsFileManager.rewriteLockWithTimeout(
+ IoTDBDescriptor.getInstance().getConfig().getRewriteLockWaitTimeoutInMS())) {
+ throw new IOException(
+ "Alter failed. "
+ + "Other file rewriting operations are in progress, please do it later.");
+ }
+ // recover log
+ File logFile =
+ SystemFileFactory.INSTANCE.getFile(storageGroupSysDir, AlteringLogger.ALTERING_LOG_NAME);
+ if (logFile.exists()) {
+ logger.info("[alter timeseries] {} rewriteUnlock", logKey);
+ tsFileManager.rewriteUnlock();
+ throw new IOException(
+ "Alter failed. "
+ + "alter.log detected, other alter operations may be running, please do it later.");
+ }
+ // rewrite target tsfiles
+ boolean done = false;
+ try (AlteringLogger alteringLogger = new AlteringLogger(logFile)) {
+ Set<Long> timePartitions = tsFileManager.getTimePartitions();
+ // Record the ALTER process for server restart recovery
+ alteringLogger.logHeader(fullPath, curEncoding, curCompressionType, timePartitions);
+ for (Long timePartition : timePartitions) {
+ logger.info("[alter timeseries] {} alterDataInTsFiles seq({})", logKey, timePartition);
+ try {
+ rewriteDataInTsFiles(
+ tsFileManager.getSequenceListByTimePartition(timePartition),
+ fullPath,
+ curEncoding,
+ curCompressionType,
+ timePartition,
+ true,
+ alteringLogger,
+ null,
+ logKey);
+ logger.info("[alter timeseries] {} alterDataInTsFiles unseq({})", logKey, timePartition);
+ rewriteDataInTsFiles(
+ tsFileManager.getUnsequenceListByTimePartition(timePartition),
+ fullPath,
+ curEncoding,
+ curCompressionType,
+ timePartition,
+ false,
+ alteringLogger,
+ null,
+ logKey);
+ } catch (IOException e) {
+ /**
+ * TODO If an exception occurs in the operation of a single tsfile, you need to terminate
+ * the operation and return 1. The modified schema and rewritten tsfile need to provide
+ * rollback commands 2. The tsfile that has not been rewritten needs to provide a recovery
+ * command
+ */
+ logger.error(
+ "[alter timeseries] " + logKey + " timePartition " + timePartition + " error", e);
+ throw e;
+ }
+ }
+ done = true;
+ } catch (Exception e) {
+ logger.error("[alter timeseries] " + logKey + " error", e);
+ throw e;
+ } finally {
+ logger.info("[alter timeseries] {} rewriteUnlock", logKey);
+ tsFileManager.rewriteUnlock();
+ // The process is complete and the logFile is deleted
+ if (done && logFile.exists()) {
+ FileUtils.delete(logFile);
+ }
+ }
+ }
+
+ private void rewriteDataInTsFiles(
+ List<TsFileResource> tsFileList,
+ PartialPath fullPath,
+ TSEncoding curEncoding,
+ CompressionType curCompressionType,
+ long timePartition,
+ boolean sequence,
+ AlteringLogger alteringLogger,
+ Set<TsFileIdentifier> undoneFiles,
+ String logKey)
+ throws IOException {
+
+ if (tsFileList == null || tsFileList.isEmpty()) {
+ return;
+ }
+ // log timePartition start
+ alteringLogger.startTimePartition(tsFileList, timePartition, sequence);
+ for (TsFileResource tsFileResource : tsFileList) {
+ if (!findUndoneResourcesAndRemove(tsFileResource, undoneFiles)) {
+ continue;
+ }
+ if (tsFileResource == null || !tsFileResource.isClosed()) {
+ return;
+ }
+ try {
+ logger.info(
+ "[alter timeseries] {} rewriteDataInTsFile:{}, fileSize:{} start",
+ logKey,
+ tsFileResource.getTsFilePath(),
+ tsFileResource.getTsFileSize());
+ // Generate the target tsFileResource
+ TsFileResource targetTsFileResource =
+ TsFileNameGenerator.generateNewAlterTsFileResource(tsFileResource);
+ // Data is read from the.tsfile file, re-encoded, compressed, and written to the .alter file
+ TsFileRewriteExcutor tsFileRewriteExcutor =
+ new TsFileRewriteExcutor(
+ tsFileResource,
+ targetTsFileResource,
+ fullPath,
+ curEncoding,
+ curCompressionType,
+ timePartition,
+ sequence);
+ tsFileRewriteExcutor.execute();
+ // .tsfile->.alter.old .alter->.tsfile
+ logger.debug("[alter timeseries] {} move tsfile", logKey);
+ tsFileResource.moveTsFile(TSFILE_SUFFIX, ALTER_OLD_TMP_FILE_SUFFIX);
Review Comment:
add `resource.writeLock()` before moving.
##########
server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java:
##########
@@ -2069,6 +2249,235 @@ private List<TsFileResource> getFileResourceListForQuery(
return tsfileResourcesForQuery;
}
+ /**
+ * alter timeseries encoding & compressionType<br>
+ * 1、flush and close tsfile<br>
+ * 2、locks<br>
+ * 3、write temp tsfiles<br>
+ * 4、unregister old tsfiles and release locks<br>
+ * 5、rename temp tsfiles<br>
+ * 6、register tsfiles<br>
+ */
+ public void alter(
+ PartialPath fullPath, TSEncoding curEncoding, CompressionType curCompressionType)
Review Comment:
What if fullPath is in the following format: root.sg.d1.**? Missing path judgment with wildcard characters.
##########
server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/CrossSpaceCompactionTask.java:
##########
@@ -299,6 +299,11 @@ private boolean addReadLock(List<TsFileResource> tsFileResourceList) {
return false;
}
try {
+ // Merge is not allowed when rewrite operation is running
+ if (tsFileManager.isRewriteLocked()) {
+ releaseAllLock();
+ return false;
+ }
Review Comment:
Rewrite operations may get stuck on compaction for a long time, resulting in the accumulation of unseq data in the system.
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org