Pipe: merge batched aligned chunks in scan parser (#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

Author: Caideyipi

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

@@ -21,6 +21,7 @@
 
 import org.apache.iotdb.db.i18n.DataNodePipeMessages;
 
+import org.apache.tsfile.common.constant.TsFileConstant;
 import org.apache.tsfile.compress.IUnCompressor;
 import org.apache.tsfile.encoding.decoder.Decoder;
 import org.apache.tsfile.encrypt.EncryptParameter;
@@ -169,22 +170,49 @@ 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(),
+            IDecryptor.getDecryptor(chunk.getEncryptParam()));
+    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);
   }