Pipe: merge batched aligned chunks in scan parser (apache#18010)

* Pipe: merge batched aligned chunks in scan parser

* Test pipe batched aligned chunk memory boundaries

* Pipe: fix batched aligned scan parser memory split

* Update TsFileInsertionEventParserTest.java

* Rename pending aligned chunk consumer

(cherry picked from commit f96fc58)

Author: Caideyipi

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

@@ -19,6 +19,7 @@
 
 package org.apache.iotdb.db.pipe.event.common.tsfile.container.scan;
 
+import org.apache.tsfile.common.constant.TsFileConstant;
 import org.apache.tsfile.compress.IUnCompressor;
 import org.apache.tsfile.encoding.decoder.Decoder;
 import org.apache.tsfile.enums.TSDataType;
@@ -159,22 +160,44 @@ static List<Long> toSuffixMaxList(final List<Long> pageEstimatedMemoryUsageInByt
   static long estimatePageMemoryUsageInBytesWithBatchData(
       final PageHeader timePageHeader,
       final Chunk timeChunk,
-      final List<TSDataType> valueDataTypeList) {
+      final List<TSDataType> valueDataTypeList)
+      throws IOException {
     return estimatePageMemoryUsageInBytesWithBatchData(
         timePageHeader.getUncompressedSize(),
         getPageRowCount(timePageHeader, timeChunk),
         valueDataTypeList);
   }
 
-  static int getPageRowCount(final PageHeader pageHeader, final Chunk chunk) {
+  static int getPageRowCount(final PageHeader pageHeader, final Chunk chunk) throws IOException {
     if (isSinglePageChunk(chunk.getHeader())) {
-      return Objects.isNull(chunk.getChunkStatistic())
-          ? 0
-          : saturateToInt(chunk.getChunkStatistic().getCount());
+      if (Objects.nonNull(chunk.getChunkStatistic())) {
+        return saturateToInt(chunk.getChunkStatistic().getCount());
+      }
+      return isTimeChunk(chunk.getHeader()) ? countSinglePageTimeValues(chunk) : 0;
     }
     return saturateToInt(pageHeader.getNumOfValues());
   }
 
+  private static int countSinglePageTimeValues(final Chunk chunk) throws IOException {
+    final ByteBuffer chunkDataBuffer = chunk.getData().duplicate();
+    final PageHeader pageHeader = deserializePageHeader(chunkDataBuffer, chunk.getHeader());
+    final ByteBuffer pageData = deserializePageData(pageHeader, chunkDataBuffer, chunk.getHeader());
+    final Decoder decoder =
+        Decoder.getDecoderByType(chunk.getHeader().getEncodingType(), TSDataType.INT64);
+
+    int rowCount = 0;
+    while (decoder.hasNext(pageData)) {
+      decoder.readLong(pageData);
+      ++rowCount;
+    }
+    return rowCount;
+  }
+
+  private static boolean isTimeChunk(final ChunkHeader chunkHeader) {
+    return (chunkHeader.getChunkType() & TsFileConstant.TIME_COLUMN_MASK)
+        == TsFileConstant.TIME_COLUMN_MASK;
+  }
+
   private static int saturateToInt(final long value) {
     return (int) Math.min(Integer.MAX_VALUE, value);
   }