Pipe: Fixed the OOM bug of parser for large aligned pages (#17639)

* chunk

* shop

* Update TsFileInsertionEventScanParser.java

* Fix

Author: Caideyipi

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

@@ -19,6 +19,7 @@
 
 package org.apache.iotdb.db.pipe.event.common.tsfile.parser.scan;
 
+import org.apache.tsfile.compress.IUnCompressor;
 import org.apache.tsfile.encoding.decoder.Decoder;
 import org.apache.tsfile.encrypt.EncryptParameter;
 import org.apache.tsfile.encrypt.IDecryptor;
@@ -31,6 +32,7 @@
 import org.apache.tsfile.read.reader.chunk.AbstractChunkReader;
 import org.apache.tsfile.read.reader.chunk.ChunkReader;
 import org.apache.tsfile.read.reader.page.AlignedPageReader;
+import org.apache.tsfile.read.reader.page.LazyLoadPageData;
 
 import java.io.IOException;
 import java.io.Serializable;
@@ -43,7 +45,8 @@
  * The {@link AlignedSinglePageWholeChunkReader} is used to read a whole single page aligned chunk
  * with need to pass in the statistics.
  */
-public class AlignedSinglePageWholeChunkReader extends AbstractChunkReader {
+public class AlignedSinglePageWholeChunkReader extends AbstractChunkReader
+    implements EstimatedMemoryChunkReader {
 
   // chunk header of the time column
   private final ChunkHeader timeChunkHeader;
@@ -58,6 +61,7 @@ public class AlignedSinglePageWholeChunkReader extends AbstractChunkReader {
   private final List<ByteBuffer> valueChunkDataBufferList = new ArrayList<>();
   // deleted intervals of all the sub sensors
   private final List<List<TimeRange>> valueDeleteIntervalsList = new ArrayList<>();
+  private final long pageEstimatedMemoryUsageInBytes;
 
   public AlignedSinglePageWholeChunkReader(
       Chunk timeChunk, List<Chunk> valueChunkList, LongConsumer filteredRowsRecord)
@@ -66,6 +70,8 @@ public AlignedSinglePageWholeChunkReader(
     this.timeChunkHeader = timeChunk.getHeader();
     this.timeChunkDataBuffer = timeChunk.getData();
     this.encryptParam = timeChunk.getEncryptParam();
+    this.pageEstimatedMemoryUsageInBytes =
+        calculatePageEstimatedMemoryUsageInBytes(timeChunk, valueChunkList);
 
     valueChunkList.forEach(
         chunk -> {
@@ -133,7 +139,7 @@ private AlignedPageReader constructAlignedPageReader(
             timePageHeader, timeChunkDataBuffer, timeChunkHeader, decryptor);
 
     List<PageHeader> valuePageHeaderList = new ArrayList<>();
-    List<ByteBuffer> valuePageDataList = new ArrayList<>();
+    LazyLoadPageData[] valuePageDataArray = new LazyLoadPageData[rawValuePageHeaderList.size()];
     List<TSDataType> valueDataTypeList = new ArrayList<>();
     List<Decoder> valueDecoderList = new ArrayList<>();
 
@@ -144,15 +150,22 @@ private AlignedPageReader constructAlignedPageReader(
       if (valuePageHeader == null || valuePageHeader.getUncompressedSize() == 0) {
         // Empty Page
         valuePageHeaderList.add(null);
-        valuePageDataList.add(null);
+        valuePageDataArray[i] = null;
         valueDataTypeList.add(null);
         valueDecoderList.add(null);
       } else {
         ChunkHeader valueChunkHeader = valueChunkHeaderList.get(i);
+        int currentPagePosition = valueChunkDataBufferList.get(i).position();
+        valueChunkDataBufferList
+            .get(i)
+            .position(currentPagePosition + valuePageHeader.getCompressedSize());
         valuePageHeaderList.add(valuePageHeader);
-        valuePageDataList.add(
-            ChunkReader.deserializePageData(
-                valuePageHeader, valueChunkDataBufferList.get(i), valueChunkHeader, decryptor));
+        valuePageDataArray[i] =
+            new LazyLoadPageData(
+                valueChunkDataBufferList.get(i).array(),
+                currentPagePosition,
+                IUnCompressor.getUnCompressor(valueChunkHeader.getCompressionType()),
+                encryptParam);
         valueDataTypeList.add(valueChunkHeader.getDataType());
         valueDecoderList.add(
             Decoder.getDecoderByType(
@@ -169,11 +182,38 @@ private AlignedPageReader constructAlignedPageReader(
             timePageData,
             defaultTimeDecoder,
             valuePageHeaderList,
-            valuePageDataList,
+            valuePageDataArray,
             valueDataTypeList,
             valueDecoderList,
             queryFilter);
     alignedPageReader.setDeleteIntervalList(valueDeleteIntervalsList);
     return alignedPageReader;
   }
+
+  @Override
+  public long getCurrentPageEstimatedMemoryUsageInBytes() {
+    return pageEstimatedMemoryUsageInBytes;
+  }
+
+  public static long calculatePageEstimatedMemoryUsageInBytes(
+      final Chunk timeChunk, final List<Chunk> valueChunkList) throws IOException {
+    final ByteBuffer timeChunkDataBuffer = timeChunk.getData().duplicate();
+    long estimatedMemoryUsageInBytes =
+        PageHeader.deserializeFrom(timeChunkDataBuffer, (Statistics<? extends Serializable>) null)
+            .getUncompressedSize();
+
+    for (final Chunk valueChunk : valueChunkList) {
+      if (valueChunk == null) {
+        continue;
+      }
+
+      final ByteBuffer valueChunkDataBuffer = valueChunk.getData().duplicate();
+      estimatedMemoryUsageInBytes +=
+          PageHeader.deserializeFrom(
+                  valueChunkDataBuffer, (Statistics<? extends Serializable>) null)
+              .getUncompressedSize();
+    }
+
+    return estimatedMemoryUsageInBytes;
+  }
 }

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

@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.event.common.tsfile.parser.scan;
+
+interface EstimatedMemoryChunkReader {
+
+  long getCurrentPageEstimatedMemoryUsageInBytes();
+}

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

@@ -29,6 +29,7 @@
 import org.apache.tsfile.file.metadata.statistics.Statistics;
 import org.apache.tsfile.read.common.Chunk;
 import org.apache.tsfile.read.reader.chunk.AbstractChunkReader;
+import org.apache.tsfile.read.reader.page.LazyLoadPageData;
 import org.apache.tsfile.read.reader.page.PageReader;
 
 import java.io.IOException;
@@ -37,17 +38,20 @@
 
 import static org.apache.tsfile.file.metadata.enums.CompressionType.UNCOMPRESSED;
 
-public class SinglePageWholeChunkReader extends AbstractChunkReader {
+public class SinglePageWholeChunkReader extends AbstractChunkReader
+    implements EstimatedMemoryChunkReader {
   private final ChunkHeader chunkHeader;
   private final ByteBuffer chunkDataBuffer;
   private final EncryptParameter encryptParam;
+  private final long pageEstimatedMemoryUsageInBytes;
 
   public SinglePageWholeChunkReader(Chunk chunk) throws IOException {
     super(Long.MIN_VALUE, null, null);
 
     this.chunkHeader = chunk.getHeader();
     this.chunkDataBuffer = chunk.getData();
     this.encryptParam = chunk.getEncryptParam();
+    this.pageEstimatedMemoryUsageInBytes = calculatePageEstimatedMemoryUsageInBytes(chunk);
     initAllPageReaders();
   }
 
@@ -62,16 +66,34 @@ private void initAllPageReaders() throws IOException {
   }
 
   private PageReader constructPageReader(PageHeader pageHeader) throws IOException {
-    IDecryptor decryptor = IDecryptor.getDecryptor(encryptParam);
+    final int currentPagePosition = chunkDataBuffer.position();
+    chunkDataBuffer.position(currentPagePosition + pageHeader.getCompressedSize());
     return new PageReader(
         pageHeader,
-        deserializePageData(pageHeader, chunkDataBuffer, chunkHeader, decryptor),
+        new LazyLoadPageData(
+            chunkDataBuffer.array(),
+            currentPagePosition,
+            IUnCompressor.getUnCompressor(chunkHeader.getCompressionType()),
+            encryptParam),
         chunkHeader.getDataType(),
         Decoder.getDecoderByType(chunkHeader.getEncodingType(), chunkHeader.getDataType()),
         defaultTimeDecoder,
         null);
   }
 
+  @Override
+  public long getCurrentPageEstimatedMemoryUsageInBytes() {
+    return pageEstimatedMemoryUsageInBytes;
+  }
+
+  public static long calculatePageEstimatedMemoryUsageInBytes(final Chunk chunk)
+      throws IOException {
+    final ByteBuffer chunkDataBuffer = chunk.getData().duplicate();
+    final PageHeader pageHeader =
+        PageHeader.deserializeFrom(chunkDataBuffer, (Statistics<? extends Serializable>) null);
+    return pageHeader.getUncompressedSize();
+  }
+
   /////////////////////////////////////////////////////////////////////////////////////////////////
   // util methods
   /////////////////////////////////////////////////////////////////////////////////////////////////