You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/10/12 06:50:50 UTC
(iotdb) 15/38: tmp
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 <jt...@163.com>
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);