Adapt file-scoped TsFile contexts and resource lifecycles

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java

@@ -20,6 +20,10 @@
 package org.apache.iotdb.db.i18n;
 
 public final class DataNodeQueryMessages {
+  public static final String EXCEPTION_QUERY_CHUNK_RESOURCES_ARE_ALREADY_CLOSED_CA6A0559 =
+      "Query chunk resources are already closed";
+  public static final String LOG_FAILED_TO_RELEASE_QUERY_CHUNK_RESOURCES_91573FE5 =
+      "Failed to release query chunk resources";
   public static final String EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_STATUS_ARG_E7C9F477 =
       "Failed to convert TsFile %s to tablets. Status: %s";
   public static final String EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD =

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java

@@ -20,6 +20,16 @@
 package org.apache.iotdb.db.i18n;
 
 public final class StorageEngineMessages {
+  public static final String EXCEPTION_CANNOT_ENABLE_WAL_NODE_ARG_BECAUSE_ITS_PREVIOUS_RECOVERY_FAILED_RESTART_AFTER_RESOLVING_THE_RECOVERY_FAILURE_56DF9750 =
+      "Cannot enable WAL node %s because its previous recovery failed; restart after resolving the recovery failure.";
+  public static final String LOG_WAL_NODE_RECOVERY_IN_ARG_WAS_INCOMPLETE_RETAINING_WAL_AND_CHECKPOINT_FILES_FOR_RETRY_48C1F7BF =
+      "WAL node recovery in {} was incomplete; retaining WAL and checkpoint files for retry.";
+  public static final String EXCEPTION_FILE_ENCRYPTION_PARAMETER_PROVIDER_RETURNED_NULL_15C41723 =
+      "File encryption parameter provider returned null";
+  public static final String EXCEPTION_PRECALCULATED_CHUNK_STAGING_IS_NOT_SUPPORTED_BY_THE_FILE_ENCRYPTION_POLICY_13DA8C32 =
+      "Precalculated chunk staging is not supported by the file encryption policy";
+  public static final String EXCEPTION_MULTIPLE_FILE_ENCRYPTION_PARAMETER_PROVIDERS_ARE_CONFIGURED_5E7DDA5B =
+      "Multiple file encryption parameter providers are configured";
   public static final String EXCEPTION_UNSUPPORTED_OR_UNAVAILABLE_WAL_FORMAT_4629CFE4 =
       "Unsupported or unavailable WAL format";
   public static final String EXCEPTION_MULTIPLE_WAL_FORMAT_PROVIDERS_ARE_CONFIGURED_DD78877D =

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java

@@ -20,6 +20,10 @@
 package org.apache.iotdb.db.i18n;
 
 public final class DataNodeQueryMessages {
+  public static final String EXCEPTION_QUERY_CHUNK_RESOURCES_ARE_ALREADY_CLOSED_CA6A0559 =
+      "查询 Chunk 资源已关闭";
+  public static final String LOG_FAILED_TO_RELEASE_QUERY_CHUNK_RESOURCES_91573FE5 =
+      "释放查询 Chunk 资源失败";
   public static final String EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_STATUS_ARG_E7C9F477 =
       "将 TsFile %s 转换为 tablet 失败。状态：%s";
   public static final String EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD =

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java

@@ -20,6 +20,16 @@
 package org.apache.iotdb.db.i18n;
 
 public final class StorageEngineMessages {
+  public static final String EXCEPTION_CANNOT_ENABLE_WAL_NODE_ARG_BECAUSE_ITS_PREVIOUS_RECOVERY_FAILED_RESTART_AFTER_RESOLVING_THE_RECOVERY_FAILURE_56DF9750 =
+      "WAL 节点 %s 上次恢复失败，无法启用；请解决恢复失败原因后重启。";
+  public static final String LOG_WAL_NODE_RECOVERY_IN_ARG_WAS_INCOMPLETE_RETAINING_WAL_AND_CHECKPOINT_FILES_FOR_RETRY_48C1F7BF =
+      "目录 {} 中的 WAL 节点恢复未完成；保留 WAL 和 checkpoint 文件以便重试。";
+  public static final String EXCEPTION_FILE_ENCRYPTION_PARAMETER_PROVIDER_RETURNED_NULL_15C41723 =
+      "文件加密参数 Provider 返回 null";
+  public static final String EXCEPTION_PRECALCULATED_CHUNK_STAGING_IS_NOT_SUPPORTED_BY_THE_FILE_ENCRYPTION_POLICY_13DA8C32 =
+      "文件加密策略不支持预计算 Chunk 暂存";
+  public static final String EXCEPTION_MULTIPLE_FILE_ENCRYPTION_PARAMETER_PROVIDERS_ARE_CONFIGURED_5E7DDA5B =
+      "配置了多个文件加密参数 Provider";
   public static final String EXCEPTION_UNSUPPORTED_OR_UNAVAILABLE_WAL_FORMAT_4629CFE4 =
       "WAL 格式不支持或对应实现不可用";
   public static final String EXCEPTION_MULTIPLE_WAL_FORMAT_PROVIDERS_ARE_CONFIGURED_DD78877D =

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

@@ -63,37 +63,61 @@ public class AlignedSinglePageWholeChunkReader extends AbstractChunkReader
   // deleted intervals of all the sub sensors
   private final List<List<TimeRange>> valueDeleteIntervalsList = new ArrayList<>();
   private final long pageEstimatedMemoryUsageInBytes;
+  private final Chunk ownedTimeChunk;
+  private final List<Chunk> ownedValueChunks;
 
   public AlignedSinglePageWholeChunkReader(
       Chunk timeChunk, List<Chunk> valueChunkList, LongConsumer filteredRowsRecord)
       throws IOException {
     super(Long.MIN_VALUE, null, filteredRowsRecord);
+    ownedTimeChunk = timeChunk;
+    ownedValueChunks = new ArrayList<>(valueChunkList);
     this.timeChunkHeader = timeChunk.getHeader();
     this.timeChunkDataBuffer = timeChunk.getData();
     this.encryptParam = timeChunk.getEncryptParam();
-    this.pageEstimatedMemoryUsageInBytes =
-        calculateMaxPageEstimatedMemoryUsageInBytesWithBatchData(timeChunk, valueChunkList);
-
-    valueChunkList.forEach(
-        chunk -> {
-          this.valueChunkHeaderList.add(chunk == null ? null : chunk.getHeader());
-          this.valueChunkDataBufferList.add(chunk == null ? null : chunk.getData());
-          this.valueDeleteIntervalsList.add(chunk == null ? null : chunk.getDeleteIntervalList());
-        });
+    try {
+      this.pageEstimatedMemoryUsageInBytes =
+          calculateMaxPageEstimatedMemoryUsageInBytesWithBatchData(timeChunk, valueChunkList);
+
+      valueChunkList.forEach(
+          chunk -> {
+            this.valueChunkHeaderList.add(chunk == null ? null : chunk.getHeader());
+            this.valueChunkDataBufferList.add(chunk == null ? null : chunk.getData());
+            this.valueDeleteIntervalsList.add(chunk == null ? null : chunk.getDeleteIntervalList());
+          });
+
+      initAllPageReaders();
+    } catch (IOException | RuntimeException | Error e) {
+      try {
+        close();
+      } catch (RuntimeException failure) {
+        e.addSuppressed(failure);
+      }
+      throw e;
+    }
+  }
 
-    initAllPageReaders();
+  @Override
+  public void close() {
+    ownedTimeChunk.close();
+    for (Chunk chunk : ownedValueChunks) {
+      if (chunk != null) {
+        chunk.close();
+      }
+    }
   }
 
   private void initAllPageReaders() throws IOException {
+    int pageIndex = 0;
     while (timeChunkDataBuffer.remaining() > 0) {
-      AlignedPageReader alignedPageReader = deserializeFromSinglePageChunk();
+      AlignedPageReader alignedPageReader = deserializeFromSinglePageChunk(pageIndex++);
       if (alignedPageReader != null) {
         pageReaderList.add(alignedPageReader);
       }
     }
   }
 
-  private AlignedPageReader deserializeFromSinglePageChunk() throws IOException {
+  private AlignedPageReader deserializeFromSinglePageChunk(int pageIndex) throws IOException {
     PageHeader timePageHeader =
         PageHeader.deserializeFrom(timeChunkDataBuffer, (Statistics<? extends Serializable>) null);
     List<PageHeader> valuePageHeaderList = new ArrayList<>();
@@ -115,7 +139,7 @@ private AlignedPageReader deserializeFromSinglePageChunk() throws IOException {
       skipCurrentPage(timePageHeader, valuePageHeaderList);
       return null;
     }
-    return constructAlignedPageReader(timePageHeader, valuePageHeaderList);
+    return constructAlignedPageReader(timePageHeader, valuePageHeaderList, pageIndex);
   }
 
   private void skipCurrentPage(PageHeader timePageHeader, List<PageHeader> valuePageHeader) {
@@ -133,11 +157,17 @@ private void skipCurrentPage(PageHeader timePageHeader, List<PageHeader> valuePa
   }
 
   private AlignedPageReader constructAlignedPageReader(
-      PageHeader timePageHeader, List<PageHeader> rawValuePageHeaderList) throws IOException {
+      PageHeader timePageHeader, List<PageHeader> rawValuePageHeaderList, int pageIndex)
+      throws IOException {
     IDecryptor decryptor = IDecryptor.getDecryptor(encryptParam);
     ByteBuffer timePageData =
         ChunkReader.deserializePageData(
-            timePageHeader, timeChunkDataBuffer, timeChunkHeader, decryptor);
+            timePageHeader,
+            timeChunkDataBuffer,
+            timeChunkHeader,
+            decryptor,
+            encryptParam,
+            pageIndex);
 
     List<PageHeader> valuePageHeaderList = new ArrayList<>();
     LazyLoadPageData[] valuePageDataArray = new LazyLoadPageData[rawValuePageHeaderList.size()];
@@ -166,7 +196,9 @@ private AlignedPageReader constructAlignedPageReader(
                 valueChunkDataBufferList.get(i).array(),
                 currentPagePosition,
                 IUnCompressor.getUnCompressor(valueChunkHeader.getCompressionType()),
-                encryptParam);
+                encryptParam,
+                pageIndex,
+                valueChunkHeader.getChunkOrdinal());
         valueDataTypeList.add(valueChunkHeader.getDataType());
         valueDecoderList.add(
             Decoder.getDecoderByType(

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

@@ -35,6 +35,7 @@
 import org.apache.tsfile.read.common.Chunk;
 import org.apache.tsfile.read.common.type.Type;
 import org.apache.tsfile.read.reader.chunk.AbstractChunkReader;
+import org.apache.tsfile.read.reader.chunk.ChunkReader;
 import org.apache.tsfile.read.reader.page.LazyLoadPageData;
 import org.apache.tsfile.read.reader.page.PageReader;
 import org.apache.tsfile.utils.RamUsageEstimator;
@@ -55,29 +56,47 @@ public class SinglePageWholeChunkReader extends AbstractChunkReader
   private final ByteBuffer chunkDataBuffer;
   private final EncryptParameter encryptParam;
   private final long pageEstimatedMemoryUsageInBytes;
+  private final Chunk ownedChunk;
 
   public SinglePageWholeChunkReader(Chunk chunk) throws IOException {
     super(Long.MIN_VALUE, null, null);
+    ownedChunk = chunk;
 
     this.chunkHeader = chunk.getHeader();
     this.chunkDataBuffer = chunk.getData();
     this.encryptParam = chunk.getEncryptParam();
-    this.pageEstimatedMemoryUsageInBytes =
-        calculateMaxPageEstimatedMemoryUsageInBytesWithBatchData(chunk);
-    initAllPageReaders();
+    try {
+      this.pageEstimatedMemoryUsageInBytes =
+          calculateMaxPageEstimatedMemoryUsageInBytesWithBatchData(chunk);
+      initAllPageReaders();
+    } catch (IOException | RuntimeException | Error e) {
+      try {
+        chunk.close();
+      } catch (RuntimeException failure) {
+        e.addSuppressed(failure);
+      }
+      throw e;
+    }
+  }
+
+  @Override
+  public void close() {
+    ownedChunk.close();
   }
 
   private void initAllPageReaders() throws IOException {
+    int pageIndex = 0;
     // construct next satisfied page header
     while (chunkDataBuffer.remaining() > 0) {
       pageReaderList.add(
           constructPageReader(
               PageHeader.deserializeFrom(
-                  chunkDataBuffer, (Statistics<? extends Serializable>) null)));
+                  chunkDataBuffer, (Statistics<? extends Serializable>) null),
+              pageIndex++));
     }
   }
 
-  private PageReader constructPageReader(PageHeader pageHeader) throws IOException {
+  private PageReader constructPageReader(PageHeader pageHeader, int pageIndex) throws IOException {
     final int currentPagePosition = chunkDataBuffer.position();
     chunkDataBuffer.position(currentPagePosition + pageHeader.getCompressedSize());
     return new PageReader(
@@ -86,7 +105,9 @@ private PageReader constructPageReader(PageHeader pageHeader) throws IOException
             chunkDataBuffer.array(),
             currentPagePosition,
             IUnCompressor.getUnCompressor(chunkHeader.getCompressionType()),
-            encryptParam),
+            encryptParam,
+            pageIndex,
+            chunkHeader.getChunkOrdinal()),
         chunkHeader.getDataType(),
         Decoder.getDecoderByType(chunkHeader.getEncodingType(), chunkHeader.getDataType()),
         getConfiguredTimeDecoder(),
@@ -194,11 +215,13 @@ private static int countSinglePageTimeValues(final Chunk chunk) throws IOExcepti
     chunkDataBuffer.rewind();
     final PageHeader pageHeader = deserializePageHeader(chunkDataBuffer, chunk.getHeader());
     final ByteBuffer pageData =
-        deserializePageData(
+        ChunkReader.deserializePageData(
             pageHeader,
             chunkDataBuffer,
             chunk.getHeader(),
-            IDecryptor.getDecryptor(chunk.getEncryptParam()));
+            IDecryptor.getDecryptor(chunk.getEncryptParam()),
+            chunk.getEncryptParam(),
+            0);
     final Decoder decoder =
         Decoder.getDecoderByType(chunk.getHeader().getEncodingType(), TSDataType.INT64);