(iotdb) 22/38: tmp

[ji...@apache.org]

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 <jt...@163.com>
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;