(iotdb) branch master updated: Pipe / Subscription : Fixed the bug that TsFileInsertionScanDataContainer may use wrong isMultiPage flag for aligned chunk when value chunk is not continuous (#13554)

[rong]

This is an automated email from the ASF dual-hosted git repository.

rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new e48b8b925b5 Pipe / Subscription : Fixed the bug that 
TsFileInsertionScanDataContainer may use wrong isMultiPage flag for aligned 
chunk when value chunk is not continuous (#13554)
e48b8b925b5 is described below

commit e48b8b925b50eec8cb3a77ace821f539ced88af1
Author: Caideyipi <87789683+caidey...@users.noreply.github.com>
AuthorDate: Fri Sep 20 18:43:43 2024 +0800

    Pipe / Subscription : Fixed the bug that TsFileInsertionScanDataContainer 
may use wrong isMultiPage flag for aligned chunk when value chunk is not 
continuous (#13554)
---
 .../container/scan/TsFileInsertionScanDataContainer.java | 16 +++++++++++-----
 1 file changed, 11 insertions(+), 5 deletions(-)

diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/scan/TsFileInsertionScanDataContainer.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/scan/TsFileInsertionScanDataContainer.java
index 7004c6ab1dc..0a83cd40193 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/scan/TsFileInsertionScanDataContainer.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/scan/TsFileInsertionScanDataContainer.java
@@ -71,13 +71,15 @@ public class TsFileInsertionScanDataContainer extends 
TsFileInsertionDataContain
   private IChunkReader chunkReader;
   private BatchData data;
 
-  private boolean isMultiPage;
+  private boolean currentIsMultiPage;
   private IDeviceID currentDevice;
   private boolean currentIsAligned;
   private final List<IMeasurementSchema> currentMeasurements = new 
ArrayList<>();
 
   // Cached time chunk
   private final List<Chunk> timeChunkList = new ArrayList<>();
+  private final List<Boolean> isMultiPageList = new ArrayList<>();
+
   private final Map<String, Integer> measurementIndexMap = new HashMap<>();
   private int lastIndex = -1;
   private ChunkHeader firstChunkHeader4NextSequentialValueChunks;
@@ -188,7 +190,8 @@ public class TsFileInsertionScanDataContainer extends 
TsFileInsertionDataContain
 
       boolean isFirstRow = true;
       while (data.hasCurrent()) {
-        if (isMultiPage || data.currentTime() >= startTime && 
data.currentTime() <= endTime) {
+        if (currentIsMultiPage
+            || data.currentTime() >= startTime && data.currentTime() <= 
endTime) {
           if (isFirstRow) {
             // Calculate row count and memory size of the tablet based on the 
first row
             Pair<Integer, Integer> rowCountAndMemorySize =
@@ -348,7 +351,7 @@ public class TsFileInsertionScanDataContainer extends 
TsFileInsertionDataContain
         case MetaMarker.ONLY_ONE_PAGE_TIME_CHUNK_HEADER:
           // Notice that the data in one chunk group is either aligned or 
non-aligned
           // There is no need to consider non-aligned chunks when there are 
value chunks
-          isMultiPage = marker == MetaMarker.CHUNK_HEADER || marker == 
MetaMarker.TIME_CHUNK_HEADER;
+          currentIsMultiPage = marker == MetaMarker.CHUNK_HEADER;
 
           chunkHeader = tsFileSequenceReader.readChunkHeader(marker);
 
@@ -363,6 +366,7 @@ public class TsFileInsertionScanDataContainer extends 
TsFileInsertionDataContain
             timeChunkList.add(
                 new Chunk(
                     chunkHeader, tsFileSequenceReader.readChunk(-1, 
chunkHeader.getDataSize())));
+            isMultiPageList.add(marker == MetaMarker.TIME_CHUNK_HEADER);
             break;
           }
 
@@ -373,7 +377,7 @@ public class TsFileInsertionScanDataContainer extends 
TsFileInsertionDataContain
           }
 
           chunkReader =
-              isMultiPage
+              currentIsMultiPage
                   ? new ChunkReader(
                       new Chunk(
                           chunkHeader,
@@ -438,6 +442,7 @@ public class TsFileInsertionScanDataContainer extends 
TsFileInsertionDataContain
           // Clear because the cached data will never be used in the next 
chunk group
           lastIndex = -1;
           timeChunkList.clear();
+          isMultiPageList.clear();
           measurementIndexMap.clear();
           final IDeviceID deviceID = 
tsFileSequenceReader.readChunkGroupHeader().getDeviceID();
           currentDevice = pattern.mayOverlapWithDevice(deviceID) ? deviceID : 
null;
@@ -459,8 +464,9 @@ public class TsFileInsertionScanDataContainer extends 
TsFileInsertionDataContain
   private boolean recordAlignedChunk(final List<Chunk> valueChunkList, final 
byte marker)
       throws IOException {
     if (!valueChunkList.isEmpty()) {
+      currentIsMultiPage = isMultiPageList.get(lastIndex);
       chunkReader =
-          isMultiPage
+          currentIsMultiPage
               ? new AlignedChunkReader(timeChunkList.get(lastIndex), 
valueChunkList, filter)
               : new 
AlignedSinglePageWholeChunkReader(timeChunkList.get(lastIndex), valueChunkList);
       currentIsAligned = true;