This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch add_flat_device in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3946193b0c5ccac3fb29b08ef633d55c43901de0 Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 16 15:42:59 2026 +0800 tmp --- .../impl/ReadChunkCompactionPerformer.java | 16 ++++++ .../impl/ReadPointCompactionPerformer.java | 49 ++++++++++++++-- .../utils/writer/AbstractCompactionWriter.java | 31 ++++++++++ .../writer/AbstractCrossCompactionWriter.java | 1 + .../writer/AbstractInnerCompactionWriter.java | 1 + .../writer/ReadPointCrossCompactionWriter.java | 11 +++- .../writer/ReadPointInnerCompactionWriter.java | 7 +++ .../writer/RepairUnsortedFileCompactionWriter.java | 25 +++++++- .../compaction/repair/RepairDataFileScanUtil.java | 66 +++++++++++++++++++++- 9 files changed, 196 insertions(+), 11 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java index 12772888e2c..f29a02556a5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java @@ -140,6 +140,22 @@ public class ReadChunkCompactionPerformer implements ISeqCompactionPerformer { InterruptedException, StorageEngineException, PageException { + if (FlatDeviceCompactionUtils.containsFlatDevice(seqFiles)) { + // Physical table devices can overlap in time even in sequence files. Chunk/page copying + // cannot merge their TAG row keys, so use the row-aware read-point path for this task. + ReadPointCompactionPerformer performer = + new ReadPointCompactionPerformer( + seqFiles, Collections.emptyList(), targetResources, firstEncryptParameter); + performer.setSummary(summary); + try { + performer.perform(); + } catch (IOException | MetadataException | InterruptedException | RuntimeException e) { + throw e; + } catch (Exception e) { + throw new IOException(e); + } + return; + } try (MultiTsFileDeviceIterator deviceIterator = new MultiTsFileDeviceIterator(seqFiles)) { schema = CompactionTableSchemaCollector.collectSchema( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadPointCompactionPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadPointCompactionPerformer.java index 8ba6cebb82d..0803f1f8aa0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadPointCompactionPerformer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadPointCompactionPerformer.java @@ -33,6 +33,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.Comp import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.subtask.ReadPointPerformerSubTask; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionTableSchemaCollector; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionUtils; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.FlatDeviceCompactionUtils; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.MultiTsFileDeviceIterator; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.reader.IDataBlockReader; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.reader.SeriesDataBlockReader; @@ -51,6 +52,7 @@ import org.apache.tsfile.common.constant.TsFileConstant; import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.encrypt.EncryptUtils; import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.TableSchema; import org.apache.tsfile.read.common.block.TsBlock; import org.apache.tsfile.read.reader.IPointReader; import org.apache.tsfile.utils.Pair; @@ -171,7 +173,12 @@ public class ReadPointCompactionPerformer if (isAligned) { compactAlignedSeries( - device, deviceIterator, compactionWriter, fragmentInstanceContext, queryDataSource); + device, + deviceIterator, + compactionWriter, + fragmentInstanceContext, + queryDataSource, + schemas.get(0).getTableSchemaMap().get(device.getTableName())); } else { compactNonAlignedSeries( device, deviceIterator, compactionWriter, fragmentInstanceContext, queryDataSource); @@ -210,7 +217,8 @@ public class ReadPointCompactionPerformer MultiTsFileDeviceIterator deviceIterator, AbstractCompactionWriter compactionWriter, FragmentInstanceContext fragmentInstanceContext, - QueryDataSource queryDataSource) + QueryDataSource queryDataSource, + TableSchema tableSchema) throws IOException, MetadataException { Map<String, MeasurementSchema> schemaMap = deviceIterator.getAllSchemasOfCurrentDevice(); IMeasurementSchema timeSchema = schemaMap.remove(TsFileConstant.TIME_COLUMN_ID); @@ -218,6 +226,8 @@ public class ReadPointCompactionPerformer if (measurementSchemas.isEmpty()) { return; } + List<Integer> rowKeyColumnIndices = + FlatDeviceCompactionUtils.getRowKeyColumnIndices(tableSchema, measurementSchemas); List<String> existedMeasurements = measurementSchemas.stream() .map(IMeasurementSchema::getMeasurementName) @@ -232,11 +242,15 @@ public class ReadPointCompactionPerformer new ArrayList<>(schemaMap.keySet()), fragmentInstanceContext, queryDataSource, - true); + true, + rowKeyColumnIndices); if (dataBlockReader.hasNextBatch()) { // chunkgroup is serialized only when at least one timeseries under this device has data - compactionWriter.startChunkGroup(device, true); + // The schema property distinguishes flat tables from ordinary devices with all-null TAGs. + compactionWriter.startChunkGroup( + device, true, FlatDeviceCompactionUtils.isFlatDevice(tableSchema)); + compactionWriter.setRowKeyColumnIndices(rowKeyColumnIndices); measurementSchemas.add(0, timeSchema); compactionWriter.startMeasurement( TsFileConstant.TIME_COLUMN_ID, @@ -314,6 +328,26 @@ public class ReadPointCompactionPerformer FragmentInstanceContext fragmentInstanceContext, QueryDataSource queryDataSource, boolean isAlign) { + return constructReader( + deviceId, + measurementIds, + measurementSchemas, + allSensors, + fragmentInstanceContext, + queryDataSource, + isAlign, + Collections.emptyList()); + } + + public static IDataBlockReader constructReader( + IDeviceID deviceId, + List<String> measurementIds, + List<IMeasurementSchema> measurementSchemas, + List<String> allSensors, + FragmentInstanceContext fragmentInstanceContext, + QueryDataSource queryDataSource, + boolean isAlign, + List<Integer> rowKeyColumnIndices) { IFullPath seriesPath; if (isAlign) { seriesPath = new AlignedFullPath(deviceId, measurementIds, measurementSchemas); @@ -322,7 +356,12 @@ public class ReadPointCompactionPerformer } return new SeriesDataBlockReader( - seriesPath, new HashSet<>(allSensors), fragmentInstanceContext, queryDataSource, true); + seriesPath, + new HashSet<>(allSensors), + fragmentInstanceContext, + queryDataSource, + true, + rowKeyColumnIndices); } @SuppressWarnings("squid:S1172") diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java index eff07d21260..d25f0264378 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java @@ -20,10 +20,12 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.writer; import org.apache.iotdb.commons.conf.IoTDBConstant; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionLastTimeCheckFailedException; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CompactionTaskSummary; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.FlatDeviceCompactionUtils; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.batch.utils.FollowingBatchCompactionAlignedChunkWriter; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.element.AlignedPageElement; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.element.ChunkMetadataElement; @@ -51,12 +53,17 @@ import org.apache.tsfile.write.chunk.ChunkWriterImpl; import org.apache.tsfile.write.chunk.IChunkWriter; import org.apache.tsfile.write.chunk.ValueChunkWriter; import org.apache.tsfile.write.schema.Schema; +import org.apache.tsfile.write.writer.TsFileIOWriter; import java.io.IOException; import java.nio.ByteBuffer; import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; +import java.util.Map; +import java.util.Set; import java.util.stream.Collectors; public abstract class AbstractCompactionWriter implements AutoCloseable { @@ -120,6 +127,30 @@ public abstract class AbstractCompactionWriter implements AutoCloseable { protected boolean isAlign; protected IDeviceID deviceId; + protected final Map<String, Set<String>> flatDeviceLogicalDevices = new HashMap<>(); + protected final Map<String, Long> flatDeviceRowCounts = new HashMap<>(); + + protected void recordFlatDeviceRows(TsBlock block, List<Integer> tagIndices) throws IOException { + if (tagIndices.isEmpty()) { + return; + } + Set<String> devices = + flatDeviceLogicalDevices.computeIfAbsent(deviceId.getTableName(), key -> new HashSet<>()); + IPointReader rows = block.getTsBlockAlignedRowIterator(); + while (rows.hasNextTimeValuePair()) { + devices.add( + FlatDeviceCompactionUtils.getLogicalDevice(deviceId, tagIndices, rows.nextTimeValuePair()) + .toString()); + } + flatDeviceRowCounts.merge(deviceId.getTableName(), (long) block.getPositionCount(), Long::sum); + } + + protected void writeFlatDeviceStatistics(TsFileIOWriter writer) { + flatDeviceLogicalDevices.forEach( + (table, devices) -> + TsFileTableSchemaUtil.addFlatDeviceStatistics( + writer, table, devices.size(), flatDeviceRowCounts.getOrDefault(table, 0L))); + } protected List<Integer> rowKeyColumnIndices = Collections.emptyList(); private TimeValuePair lastRowKey; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCrossCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCrossCompactionWriter.java index d9d5f0c07d7..880d688080a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCrossCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCrossCompactionWriter.java @@ -176,6 +176,7 @@ public abstract class AbstractCrossCompactionWriter extends AbstractCompactionWr @Override public void endFile() throws IOException { for (int i = 0; i < isEmptyFile.length; i++) { + writeFlatDeviceStatistics(targetFileWriters.get(i)); targetFileWriters.get(i).endFile(); // set empty target file to DELETED if (isEmptyFile[i]) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractInnerCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractInnerCompactionWriter.java index 13d8a9dac8b..c94ea1a7c27 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractInnerCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractInnerCompactionWriter.java @@ -174,6 +174,7 @@ public abstract class AbstractInnerCompactionWriter extends AbstractCompactionWr targetResources.get(currentFileIndex).forceMarkDeleted(); return; } + writeFlatDeviceStatistics(fileWriter); fileWriter.endFile(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointCrossCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointCrossCompactionWriter.java index 55a9e9ee4bf..8d447619b2e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointCrossCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointCrossCompactionWriter.java @@ -21,6 +21,7 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.wr import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.i18n.StorageEngineMessages; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.FlatDeviceCompactionUtils; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.element.AlignedPageElement; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.element.ChunkMetadataElement; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.writer.flushcontroller.AbstractCompactionFlushController; @@ -61,6 +62,8 @@ public class ReadPointCrossCompactionWriter extends AbstractCrossCompactionWrite @Override public void write(TsBlock tsBlock, int subTaskId) throws IOException { + checkFlatDeviceRowOrder(tsBlock); + recordFlatDeviceRows(tsBlock, rowKeyColumnIndices); TimeColumn timestamps = (TimeColumn) tsBlock.getTimeColumn(); Column[] columns = tsBlock.getValueColumns(); int batchSize = tsBlock.getPositionCount(); @@ -76,8 +79,12 @@ public class ReadPointCrossCompactionWriter extends AbstractCrossCompactionWrite synchronized (this) { // we need to synchronized here to avoid multi-thread competition in sub-task TsFileResource resource = targetResources.get(seqFileIndexArray[subTaskId]); - resource.updateStartTime(deviceId, timestamps.getStartTime()); - resource.updateEndTime(deviceId, timestamps.getEndTime()); + if (rowKeyColumnIndices.isEmpty()) { + resource.updateStartTime(deviceId, timestamps.getStartTime()); + resource.updateEndTime(deviceId, timestamps.getEndTime()); + } else { + FlatDeviceCompactionUtils.updateResource(resource, deviceId, rowKeyColumnIndices, tsBlock); + } } checkChunkSizeAndMayOpenANewChunk( targetFileWriters.get(seqFileIndexArray[subTaskId]), chunkWriter, subTaskId); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointInnerCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointInnerCompactionWriter.java index ec266f9cde8..82fad5bd3e8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointInnerCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointInnerCompactionWriter.java @@ -21,6 +21,7 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.wr import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.i18n.StorageEngineMessages; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.FlatDeviceCompactionUtils; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.element.AlignedPageElement; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.element.ChunkMetadataElement; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.writer.flushcontroller.AbstractCompactionFlushController; @@ -66,6 +67,12 @@ public class ReadPointInnerCompactionWriter extends AbstractInnerCompactionWrite @Override public void write(TsBlock tsBlock, int subTaskId) throws IOException { + checkFlatDeviceRowOrder(tsBlock); + recordFlatDeviceRows(tsBlock, rowKeyColumnIndices); + if (!rowKeyColumnIndices.isEmpty()) { + FlatDeviceCompactionUtils.updateResource( + targetResources.get(currentFileIndex), deviceId, rowKeyColumnIndices, tsBlock); + } TimeColumn timestamps = (TimeColumn) tsBlock.getTimeColumn(); Column[] columns = tsBlock.getValueColumns(); int batchSize = tsBlock.getPositionCount(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/RepairUnsortedFileCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/RepairUnsortedFileCompactionWriter.java index 63e8a58a440..08d1740e351 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/RepairUnsortedFileCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/RepairUnsortedFileCompactionWriter.java @@ -20,20 +20,27 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.writer; import org.apache.iotdb.commons.utils.TestOnly; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.FlatDeviceCompactionUtils; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.AlignedPriorityMergeReader; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.utils.EncryptDBUtils; import org.apache.tsfile.encrypt.EncryptParameter; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.read.TimeValuePair; import org.apache.tsfile.read.common.block.TsBlock; import org.apache.tsfile.read.reader.IPointReader; import org.apache.tsfile.utils.TsPrimitiveType; +import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; import org.apache.tsfile.write.chunk.IChunkWriter; +import org.apache.tsfile.write.chunk.ValueChunkWriter; import java.io.IOException; import java.util.ArrayList; import java.util.Comparator; import java.util.List; +import java.util.stream.Collectors; /** Used for fixing files which contains internal unsorted data */ public class RepairUnsortedFileCompactionWriter extends ReadPointInnerCompactionWriter { @@ -64,11 +71,23 @@ public class RepairUnsortedFileCompactionWriter extends ReadPointInnerCompaction super.endMeasurement(subTaskId); return; } - // remove duplicate timestamp and sort - dataOfCurrentSeries.sort(Comparator.comparingLong(TimeValuePair::getTimestamp)); + // Flat-device rows share timestamps; only identical time-and-TAG keys may be merged. + Comparator<TimeValuePair> comparator = Comparator.comparingLong(TimeValuePair::getTimestamp); + if (!rowKeyColumnIndices.isEmpty()) { + List<TSDataType> types = + ((AlignedChunkWriterImpl) chunkWriters[subTaskId]) + .getValueChunkWriterList().stream() + .map(ValueChunkWriter::getDataType) + .collect(Collectors.toList()); + comparator = + (left, right) -> + AlignedPriorityMergeReader.compareAlignedRowKey( + left, right, types, rowKeyColumnIndices); + } + dataOfCurrentSeries.sort(comparator); TimeValuePair previousTimeValuePair = dataOfCurrentSeries.get(0); for (TimeValuePair timeValuePair : dataOfCurrentSeries) { - if (timeValuePair.getTimestamp() == previousTimeValuePair.getTimestamp()) { + if (comparator.compare(timeValuePair, previousTimeValuePair) == 0) { // merge to previous TimeValuePair if is aligned series mergeTimeValuePair(timeValuePair, previousTimeValuePair); continue; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java index 97381eb29d0..003c600ba92 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.repair; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionLastTimeCheckFailedException; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionStatisticsCheckFailedException; @@ -26,6 +27,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.Com import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.reader.CompactionChunkReader; import org.apache.iotdb.db.storageengine.dataregion.compaction.io.CompactionTsFileReader; import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.constant.CompactionType; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.AlignedPriorityMergeReader; 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.ArrayDeviceTimeIndex; @@ -37,20 +39,26 @@ import org.apache.tsfile.common.constant.TsFileConstant; import org.apache.tsfile.compress.IUnCompressor; import org.apache.tsfile.encoding.decoder.Decoder; import org.apache.tsfile.encrypt.IDecryptor; +import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.MetaMarker; import org.apache.tsfile.file.header.ChunkHeader; import org.apache.tsfile.file.header.PageHeader; +import org.apache.tsfile.file.metadata.AbstractAlignedChunkMetadata; import org.apache.tsfile.file.metadata.ChunkMetadata; import org.apache.tsfile.file.metadata.IChunkMetadata; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.MetadataIndexNode; +import org.apache.tsfile.file.metadata.TableSchema; import org.apache.tsfile.file.metadata.TimeseriesMetadata; import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.TimeValuePair; import org.apache.tsfile.read.TsFileDeviceIterator; import org.apache.tsfile.read.TsFileSequenceReader; import org.apache.tsfile.read.common.Chunk; import org.apache.tsfile.read.common.TimeRange; +import org.apache.tsfile.read.reader.IPointReader; +import org.apache.tsfile.read.reader.chunk.TableChunkReader; import org.apache.tsfile.utils.Pair; import org.apache.tsfile.utils.ReadWriteForEncodingUtils; import org.slf4j.Logger; @@ -79,6 +87,7 @@ public class RepairDataFileScanUtil { private boolean previousTimeSet; private long previousTime; private boolean printLog; + private boolean currentDeviceIsFlat; public RepairDataFileScanUtil(TsFileResource resource) { this(resource, false); @@ -121,6 +130,9 @@ public class RepairDataFileScanUtil { while (deviceInFileIterator.hasNext()) { Pair<IDeviceID, Boolean> deviceIsAlignedPair = deviceInFileIterator.next(); IDeviceID deviceInfile = deviceIsAlignedPair.getLeft(); + currentDeviceIsFlat = + TsFileTableSchemaUtil.isFlatDevice( + reader.getTsFileProperties(), deviceInfile.getTableName()); if (checkTsFileResource) { if (!deviceIdsInTimeIndex.contains(deviceInfile)) { throw new CompactionStatisticsCheckFailedException( @@ -148,6 +160,9 @@ public class RepairDataFileScanUtil { metadataIndexNode, deviceTimeRangeInResource, checkTsFileResource); + if (currentDeviceIsFlat) { + checkFlatDeviceRows(reader, deviceInfile); + } } else { checkNonAlignedDeviceSeries( reader, @@ -240,6 +255,54 @@ public class RepairDataFileScanUtil { } } + private void checkFlatDeviceRows(TsFileSequenceReader reader, IDeviceID device) + throws IOException { + TableSchema schema = reader.getTableSchemaMap().get(device.getTableName()); + List<String> tags = new ArrayList<>(); + List<TSDataType> types = new ArrayList<>(); + List<Integer> indices = new ArrayList<>(); + for (int i = 0; i < schema.getColumnTypes().size(); i++) { + if (schema.getColumnTypes().get(i) == ColumnCategory.TAG) { + // Only TAG chunks are read below, so indices refer to the projected TAG vector. + indices.add(tags.size()); + tags.add(schema.getColumnSchemas().get(i).getMeasurementName()); + types.add(schema.getColumnSchemas().get(i).getType()); + } + } + TimeValuePair previous = null; + for (AbstractAlignedChunkMetadata metadata : reader.getAlignedChunkMetadata(device, false)) { + Map<String, IChunkMetadata> valueMetadata = new HashMap<>(); + for (IChunkMetadata value : metadata.getValueChunkMetadataList()) { + if (value != null) { + valueMetadata.put(value.getMeasurementUid(), value); + } + } + List<Chunk> tagChunks = new ArrayList<>(); + for (String tag : tags) { + IChunkMetadata value = valueMetadata.get(tag); + tagChunks.add(value == null ? null : reader.readMemChunk((ChunkMetadata) value)); + } + TableChunkReader chunkReader = + new TableChunkReader( + reader.readMemChunk((ChunkMetadata) metadata.getTimeChunkMetadata()), + tagChunks, + null); + while (chunkReader.hasNextSatisfiedPage()) { + IPointReader rows = chunkReader.nextPageData().getBatchDataIterator(); + while (rows.hasNextTimeValuePair()) { + TimeValuePair row = rows.nextTimeValuePair(); + if (previous != null + && AlignedPriorityMergeReader.compareAlignedRowKey(previous, row, types, indices) + >= 0) { + throw new CompactionLastTimeCheckFailedException( + device.toString(), row.getTimestamp(), previous.getTimestamp()); + } + previous = row; + } + } + } + } + private void checkTimeChunkInAlignedSeries( TsFileSequenceReader reader, IDeviceID device, ChunkMetadata timeChunkMetadata) throws IOException { @@ -439,7 +502,8 @@ public class RepairDataFileScanUtil { } private void checkPreviousTimeAndUpdate(IDeviceID deviceID, long time) { - if (previousTimeSet && previousTime >= time) { + if (previousTimeSet + && (previousTime > time || (previousTime == time && !currentDeviceIsFlat))) { throw new CompactionLastTimeCheckFailedException(deviceID.toString(), time, previousTime); } previousTime = time;
