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 e5f58907e7d2a95ebf03f7ddbfd6fe5737f7855a Author: Tian Jiang <[email protected]> AuthorDate: Tue Sep 22 14:47:44 2026 +0800 tmp --- .../writer/ReadPointCrossCompactionWriter.java | 68 ++++++++++++++++++++++ .../writer/ReadPointInnerCompactionWriter.java | 2 +- .../compaction/repair/RepairDataFileScanUtil.java | 3 +- 3 files changed, 71 insertions(+), 2 deletions(-) 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 5051c1d68b9..b7e2175461d 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 @@ -102,6 +102,74 @@ public class ReadPointCrossCompactionWriter extends AbstractCrossCompactionWrite lastTimeSet[subTaskId] = true; } + private int flatTargetIndex(long timestamp) { + int index = 0; + while (index < seqTsFileResources.size() - 1 + && (timestamp > currentDeviceEndTime[index] + || !isCurrentDeviceExistedInSourceSeqFiles[index])) { + index++; + } + return index; + } + + private void writeFlatDeviceBlock(TsBlock block, int subTaskId) throws IOException { + int start = 0; + while (start < block.getPositionCount()) { + int targetIndex = flatTargetIndex(block.getTimeByIndex(start)); + int end = start + 1; + while (end < block.getPositionCount() + && flatTargetIndex(block.getTimeByIndex(end)) == targetIndex) { + end++; + } + AlignedChunkWriterImpl chunkWriter = (AlignedChunkWriterImpl) chunkWriters[subTaskId]; + if (seqFileIndexArray[subTaskId] != targetIndex) { + // A new TAG may restart time and return to an earlier target. Flush the old target's + // chunk before switching, including backwards switches; each target remains TAG/time + // sorted. + sealChunk(targetFileWriters.get(seqFileIndexArray[subTaskId]), chunkWriter, subTaskId); + seqFileIndexArray[subTaskId] = targetIndex; + } + FlatDeviceAlignedChunkWriter flatChunkWriter = (FlatDeviceAlignedChunkWriter) chunkWriter; + flatChunkWriter.write( + (TimeColumn) block.getTimeColumn(), block.getValueColumns(), end - start, start); + chunkPointNumArray[subTaskId] += end - start; + if (hasVariableLengthTypeArray[subTaskId]) { + writtenPointTotalSizeArray[subTaskId] += + estimateWrittenPointTotalSize(block, start, end - start); + } + TsFileIOWriter writer = targetFileWriters.get(targetIndex); + Set<String> devices = + flatDevicesByFile + .computeIfAbsent(writer, key -> new HashMap<>()) + .computeIfAbsent(deviceId.getTableName(), key -> new HashSet<>()); + recordFlatDeviceLogicalDevices(block, start, end - start, rowKeyColumnIndices, devices); + flatRowsByFile + .computeIfAbsent(writer, key -> new HashMap<>()) + .merge(deviceId.getTableName(), (long) (end - start), Long::sum); + FlatDeviceCompactionUtils.updateResource( + targetResources.get(targetIndex), + deviceId, + block, + start, + end - start); + checkChunkSizeAndMayOpenANewChunk(targetFileWriters.get(targetIndex), chunkWriter, subTaskId); + isDeviceExistedInTargetFiles[targetIndex] = true; + isEmptyFile[targetIndex] = false; + start = end; + } + } + + @Override + protected void writeFlatDeviceStatistics(TsFileIOWriter writer) { + Map<String, Set<String>> tables = flatDevicesByFile.get(writer); + if (tables != null) { + tables.forEach( + (table, devices) -> + TsFileTableSchemaUtil.addFlatDeviceStatistics( + writer, table, devices.size(), flatRowsByFile.get(writer).get(table))); + } + } + @Override protected TsFileSequenceReader getFileReader(TsFileResource resource) throws IOException { return FileReaderManager.getInstance() 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 82fad5bd3e8..07c88725962 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 @@ -71,7 +71,7 @@ public class ReadPointInnerCompactionWriter extends AbstractInnerCompactionWrite recordFlatDeviceRows(tsBlock, rowKeyColumnIndices); if (!rowKeyColumnIndices.isEmpty()) { FlatDeviceCompactionUtils.updateResource( - targetResources.get(currentFileIndex), deviceId, rowKeyColumnIndices, tsBlock); + targetResources.get(currentFileIndex), deviceId, tsBlock); } TimeColumn timestamps = (TimeColumn) tsBlock.getTimeColumn(); Column[] columns = tsBlock.getValueColumns(); 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 003c600ba92..2b94bb5c42c 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 @@ -27,7 +27,6 @@ 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; @@ -61,6 +60,7 @@ 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.apache.tsfile.utils.TsPrimitiveType; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -68,6 +68,7 @@ import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.HashSet;
