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 24cebc31f7d87e88eaf23501bf9edb7eaec0bbf3 Author: Tian Jiang <[email protected]> AuthorDate: Tue Sep 15 18:34:22 2026 +0800 tmp --- .../execute/performer/impl/FastCompactionPerformer.java | 11 +++++++++++ .../execute/performer/impl/ReadChunkCompactionPerformer.java | 1 + .../execute/utils/writer/AbstractCrossCompactionWriter.java | 8 +++----- 3 files changed, 15 insertions(+), 5 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/FastCompactionPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/FastCompactionPerformer.java index 9b57f56499a..b8e0c3ed8fd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/FastCompactionPerformer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/FastCompactionPerformer.java @@ -38,6 +38,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.subt import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionSeriesContext; 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.writer.AbstractCompactionWriter; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.writer.FastCrossCompactionWriter; @@ -157,6 +158,16 @@ public class FastCompactionPerformer @Override public void perform() throws Exception { + if (FlatDeviceCompactionUtils.containsFlatDevice(seqFiles) + || FlatDeviceCompactionUtils.containsFlatDevice(unseqFiles)) { + // Fast compaction deduplicates by timestamp and can split TAG columns across batches. + // Keep the complete flat-device row key together in the read-point path. + ReadPointCompactionPerformer performer = + new ReadPointCompactionPerformer(seqFiles, unseqFiles, targetFiles, encryptParameter); + performer.setSummary(subTaskSummary); + performer.perform(); + return; + } this.subTaskSummary.setTemporalFileNum(targetFiles.size()); try (MultiTsFileDeviceIterator deviceIterator = new MultiTsFileDeviceIterator(seqFiles, unseqFiles, readerCacheMap); 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 9f08759c996..12772888e2c 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 @@ -32,6 +32,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.Comp import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionTableSchema; 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.executor.batch.BatchedReadChunkAlignedSeriesCompactionExecutor; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.readchunk.SingleSeriesCompactionExecutor; 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 f2e518021ce..d9d5f0c07d7 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 @@ -37,7 +37,6 @@ import org.apache.tsfile.file.metadata.TimeseriesMetadata; import org.apache.tsfile.read.TimeValuePair; import org.apache.tsfile.read.TsFileSequenceReader; import org.apache.tsfile.read.common.block.TsBlock; -import org.apache.tsfile.utils.Pair; import org.apache.tsfile.utils.TsPrimitiveType; import org.apache.tsfile.write.schema.Schema; @@ -242,10 +241,9 @@ public abstract class AbstractCrossCompactionWriter extends AbstractCompactionWr while (fileIndex < seqTsFileResources.size()) { ITimeIndex timeIndex = seqTsFileResources.get(fileIndex).getTimeIndex(); if (isFlatDevice) { - Optional<Pair<Long, Long>> tableTime = - seqTsFileResources.get(fileIndex).getTableTime(deviceId.getTableName()); - currentDeviceEndTime[fileIndex] = tableTime.map(time -> time.right).orElse(Long.MIN_VALUE); - isCurrentDeviceExistedInSourceSeqFiles[fileIndex] = tableTime.isPresent(); + Optional<Long> endTime = timeIndex.getEndTime(deviceId); + currentDeviceEndTime[fileIndex] = endTime.orElse(Long.MIN_VALUE); + isCurrentDeviceExistedInSourceSeqFiles[fileIndex] = endTime.isPresent(); } else if (timeIndex.getTimeIndexType() != ITimeIndex.FILE_TIME_INDEX_TYPE) { // the timeIndexType of resource is deviceTimeIndex Optional<Long> endTime = timeIndex.getEndTime(deviceId);
