[To dev/1.3] Pipe: Fixed the first-chunk calculation bug for scan parser (#17597) (#17622)

* Pipe: Fixed the first-chunk calculation bug for scan parser (#17597)

* Update TsFileInsertionDataContainerTest.java

* Update TsFileInsertionDataContainerTest.java

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/scan/TsFileInsertionScanDataContainer.java

@@ -605,11 +605,6 @@ private void moveToNextChunkReader() throws IOException, IllegalStateException {
                   final long chunkSize = timeChunkSize + valueChunkSize;
                   if (chunkSize + chunkHeader.getDataSize()
                       > allocatedMemoryBlockForChunk.getMemoryUsageInBytes()) {
-                    if (valueChunkList.size() == 1
-                        && chunkSize > allocatedMemoryBlockForChunk.getMemoryUsageInBytes()) {
-                      PipeDataNodeResourceManager.memory()
-                          .forceResize(allocatedMemoryBlockForChunk, chunkSize);
-                    }
                     needReturn = recordAlignedChunk(valueChunkList, marker);
                   }
                 }
@@ -619,9 +614,11 @@ private void moveToNextChunkReader() throws IOException, IllegalStateException {
                 firstChunkHeader4NextSequentialValueChunks = chunkHeader;
                 return;
               }
+              resizeChunkMemoryBlockIfFirstValueChunkExceedsLimit(valueChunkList, chunkHeader);
             } else {
               chunkHeader = firstChunkHeader4NextSequentialValueChunks;
               firstChunkHeader4NextSequentialValueChunks = null;
+              resizeChunkMemoryBlockIfFirstValueChunkExceedsLimit(valueChunkList, chunkHeader);
             }
 
             Chunk chunk =
@@ -690,6 +687,20 @@ private boolean recordAlignedChunk(final List<Chunk> valueChunkList, final byte
     return false;
   }
 
+  private void resizeChunkMemoryBlockIfFirstValueChunkExceedsLimit(
+      final List<Chunk> valueChunkList, final ChunkHeader valueChunkHeader) {
+    if (!valueChunkList.isEmpty() || lastIndex < 0) {
+      return;
+    }
+
+    final long chunkSize =
+        PipeMemoryWeightUtil.calculateChunkRamBytesUsed(timeChunkList.get(lastIndex))
+            + valueChunkHeader.getDataSize();
+    if (chunkSize > allocatedMemoryBlockForChunk.getMemoryUsageInBytes()) {
+      PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlockForChunk, chunkSize);
+    }
+  }
+
   @Override
   public void close() {
     super.close();

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java

@@ -27,6 +27,7 @@
 import org.apache.iotdb.db.pipe.event.common.tsfile.container.TsFileInsertionDataContainer;
 import org.apache.iotdb.db.pipe.event.common.tsfile.container.query.TsFileInsertionQueryDataContainer;
 import org.apache.iotdb.db.pipe.event.common.tsfile.container.scan.TsFileInsertionScanDataContainer;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
 import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionTestFileWriter;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
@@ -56,6 +57,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.lang.reflect.Field;
 import java.time.LocalDate;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -119,6 +121,47 @@ public void testScanContainer() throws Exception {
     System.out.println(System.currentTimeMillis() - startTime);
   }
 
+  @Test
+  public void testScanParserResizesChunkMemoryForFirstAlignedValueChunk() throws Exception {
+    final long originalPipeMaxReaderChunkSize =
+        PipeConfig.getInstance().getPipeMaxReaderChunkSize();
+    CommonDescriptor.getInstance().getConfig().setPipeMaxReaderChunkSize(0);
+
+    alignedTsFile = new File("single-aligned-value-chunk.tsfile");
+    final List<MeasurementSchema> schemaList = new ArrayList<>();
+    schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
+
+    final Tablet tablet = new Tablet("root.sg.d", schemaList, 2);
+    tablet.rowSize = 2;
+    tablet.addTimestamp(0, 1);
+    tablet.addValue("s1", 0, 1L);
+    tablet.addTimestamp(1, 2);
+    tablet.addValue("s1", 1, 2L);
+
+    try {
+      try (final TsFileWriter writer = new TsFileWriter(alignedTsFile)) {
+        writer.registerAlignedTimeseries(new Path("root.sg.d"), schemaList);
+        writer.writeAligned(tablet);
+      }
+
+      try (final TsFileInsertionScanDataContainer parser =
+          new TsFileInsertionScanDataContainer(
+              alignedTsFile,
+              new PrefixPipePattern("root"),
+              Long.MIN_VALUE,
+              Long.MAX_VALUE,
+              null,
+              null,
+              false)) {
+        Assert.assertTrue(getAllocatedChunkMemory(parser).getMemoryUsageInBytes() > 0);
+      }
+    } finally {
+      CommonDescriptor.getInstance()
+          .getConfig()
+          .setPipeMaxReaderChunkSize(originalPipeMaxReaderChunkSize);
+    }
+  }
+
   public void testToTabletInsertionEvents(final boolean isQuery) throws Exception {
     // Test empty chunk
     testMixedTsFileWithEmptyChunk(isQuery);
@@ -645,4 +688,12 @@ private int getNonNullSize(final Tablet tablet) {
     }
     return count;
   }
+
+  private PipeMemoryBlock getAllocatedChunkMemory(final TsFileInsertionScanDataContainer parser)
+      throws NoSuchFieldException, IllegalAccessException {
+    final Field field =
+        TsFileInsertionScanDataContainer.class.getDeclaredField("allocatedMemoryBlockForChunk");
+    field.setAccessible(true);
+    return (PipeMemoryBlock) field.get(parser);
+  }
 }