jt2594838 · Oct 22, 2026 · Oct 22, 2026
diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -2403,6 +2403,10 @@ private DataNodePipeMessages() {}
   public static final String PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC =
       "forceResize: failed to allocate memory after %d retries, total memory size %d bytes, used "
           + "memory size %d bytes, requested memory size %d bytes";
+  public static final String EXCEPTION_FORCERESIZE_FAILED_AFTER_ARG_RETRIES_TOTAL_ARG_BYTES_USED_ARG_BYTES_USEDTABLETS_ARG_BYTES_USEDTSFILES_ARG_BYTES_TABLETLIMIT_ARG_BYTES_TSFILELIMIT_ARG_BYTES_TABLETANDTSFILELIMIT_ARG_BYTES_PROTECTEDPARSERLIMIT_ARG_BYTES_REQUESTED_ARG_BYTES_311C8C7E =
+      "forceResize failed after %d retries: total=%d bytes, used=%d bytes, "
+          + "usedTablets=%d bytes, usedTsFiles=%d bytes, tabletLimit=%d bytes, tsFileLimit=%d bytes, "
+          + "tabletAndTsFileLimit=%d bytes, protectedParserLimit=%d bytes, requested=%d bytes";
   public static final String
       EXCEPTION_UNSUPPORTED_BATCH_TYPE_ARG_WHEN_TRANSFERRING_TABLET_INSERTION_EVENT_66153E12 =
           "Unsupported batch type %s when transferring tablet insertion event.";

diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -2233,6 +2233,10 @@ private DataNodePipeMessages() {}
       "forceAllocate：重试 %d 次后仍无法分配内存，总内存大小 %d bytes，已用内存大小 %d bytes，请求内存大小 %d bytes";
   public static final String PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC =
       "forceResize：重试 %d 次后仍无法分配内存，总内存大小 %d bytes，已用内存大小 %d bytes，请求内存大小 %d bytes";
+  public static final String EXCEPTION_FORCERESIZE_FAILED_AFTER_ARG_RETRIES_TOTAL_ARG_BYTES_USED_ARG_BYTES_USEDTABLETS_ARG_BYTES_USEDTSFILES_ARG_BYTES_TABLETLIMIT_ARG_BYTES_TSFILELIMIT_ARG_BYTES_TABLETANDTSFILELIMIT_ARG_BYTES_PROTECTEDPARSERLIMIT_ARG_BYTES_REQUESTED_ARG_BYTES_311C8C7E =
+      "forceResize 重试 %d 次后仍无法分配内存：total=%d bytes，used=%d bytes，"
+          + "usedTablets=%d bytes，usedTsFiles=%d bytes，tabletLimit=%d bytes，tsFileLimit=%d bytes，"
+          + "tabletAndTsFileLimit=%d bytes，protectedParserLimit=%d bytes，requested=%d bytes";
   public static final String
       EXCEPTION_UNSUPPORTED_BATCH_TYPE_ARG_WHEN_TRANSFERRING_TABLET_INSERTION_EVENT_66153E12 =
           "传输 tablet insertion event 时不支持 batch 类型 %s。";

diff --git a/...che/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java b/...che/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java
@@ -20,10 +20,10 @@
 package org.apache.iotdb.db.pipe.event.common.tsfile.parser;
 
 import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
-import org.apache.iotdb.db.i18n.DataNodePipeMessages;
 import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
 import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
 import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager;
 
 /** Allocates parser working memory from the pool owned by the caller. */
 public interface TsFileInsertionEventParserMemoryManager {
@@ -63,11 +63,8 @@ public TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry(
               final String name, final long sizeInBytes) {
             return new PipeBlock(
                 PipeDataNodeResourceManager.memory()
-                    .forceAllocateForTabletWithRetry(
-                        name,
-                        sizeInBytes,
-                        PipeMemoryBlockCategory.TABLET,
-                        TsFileInsertionEventParser.class.getSimpleName()));
+                    .forceAllocateForTsFileParserWithRetry(
+                        name, sizeInBytes, TsFileInsertionEventParser.class.getSimpleName()));
           }
 
           @Override
@@ -90,8 +87,7 @@ public TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry(
               final Object assigner) {
             return new PipeBlock(
                 PipeDataNodeResourceManager.memory()
-                    .forceAllocateForTabletWithRetry(
-                        name, sizeInBytes, PipeMemoryBlockCategory.TABLET, assigner, parent));
+                    .forceAllocateForTsFileParserWithRetry(name, sizeInBytes, assigner, parent));
           }
 
           @Override
@@ -123,15 +119,10 @@ public long getMemoryUsageInBytes() {
     @Override
     public void forceResize(final long newSizeInBytes) {
       final long oldSize = delegate.getMemoryUsageInBytes();
-      if (!PipeDataNodeResourceManager.memory().tryResize(delegate, newSizeInBytes)) {
+      final PipeMemoryManager memoryManager = PipeDataNodeResourceManager.memory();
+      if (!memoryManager.tryResize(delegate, newSizeInBytes)) {
         throw new PipeRuntimeOutOfMemoryCriticalException(
-            String.format(
-                DataNodePipeMessages
-                    .PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC,
-                0,
-                PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes(),
-                PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes(),
-                newSizeInBytes - oldSize));
+            memoryManager.getResizeFailureMessage(0, newSizeInBytes - oldSize));
       }
     }
 

diff --git a/...anode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlockCategory.java b/...anode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlockCategory.java
@@ -48,6 +48,7 @@ public static PipeMemoryBlockCategory fromType(final PipeMemoryBlockType type) {
     return switch (type) {
       case NORMAL -> OTHER;
       case TABLET -> TABLET;
+      case TS_FILE_PARSER -> PARSER;
       case TS_FILE -> TS_FILE;
       case BATCH -> BATCH;
       case WAL -> WAL;

diff --git a/.../datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlockType.java b/.../datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlockType.java
@@ -22,6 +22,7 @@
 public enum PipeMemoryBlockType {
   NORMAL,
   TABLET,
+  TS_FILE_PARSER,
   TS_FILE,
   BATCH,
   WAL

diff --git a/...re/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java b/...re/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
@@ -152,6 +152,35 @@ private double allowedMaxMemorySizeInBytesOfTsTiles(
         * totalNonFloatingMemorySizeInBytes;
   }
 
+  private boolean isEnoughForTsFileParserMemory(
+      final long extraMemoryInBytes, final long totalNonFloatingMemorySizeInBytes) {
+    return (double) usedMemorySizeInBytesOfTablets
+            + (double) usedMemorySizeInBytesOfTsFiles
+            + (double) extraMemoryInBytes
+        < EXCEED_PROTECT_THRESHOLD
+            * allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes);
+  }
+
+  /** Reports both global availability and the category limits that can reject a resize. */
+  public synchronized String getResizeFailureMessage(final int retries, final long requestedSize) {
+    final long totalNonFloatingMemorySizeInBytes = getTotalNonFloatingMemorySizeInBytes();
+    final double combinedLimit =
+        allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes);
+    return String.format(
+        DataNodePipeMessages
+            .EXCEPTION_FORCERESIZE_FAILED_AFTER_ARG_RETRIES_TOTAL_ARG_BYTES_USED_ARG_BYTES_USEDTABLETS_ARG_BYTES_USEDTSFILES_ARG_BYTES_TABLETLIMIT_ARG_BYTES_TSFILELIMIT_ARG_BYTES_TABLETANDTSFILELIMIT_ARG_BYTES_PROTECTEDPARSERLIMIT_ARG_BYTES_REQUESTED_ARG_BYTES_311C8C7E,
+        retries,
+        totalNonFloatingMemorySizeInBytes,
+        memoryBlock.getUsedMemoryInBytes(),
+        usedMemorySizeInBytesOfTablets,
+        usedMemorySizeInBytesOfTsFiles,
+        Math.round(allowedMaxMemorySizeInBytesOfTablets(totalNonFloatingMemorySizeInBytes)),
+        Math.round(allowedMaxMemorySizeInBytesOfTsTiles(totalNonFloatingMemorySizeInBytes)),
+        Math.round(combinedLimit),
+        Math.round(EXCEED_PROTECT_THRESHOLD * combinedLimit),
+        requestedSize);
+  }
+
   private static long getTsFileParserMemorySizeInBytes() {
     return Math.max(
         PIPE_CONFIG.getTsFileParserMemory(), PIPE_CONFIG.getPipeMemoryAllocateMinSizeInBytes());
@@ -556,6 +585,10 @@ < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInByt
 
   private boolean isHardEnoughForResizing(
       final PipeMemoryBlock block, final long extraMemoryInBytes) {
+    if (block instanceof PipeTsFileParserMemoryBlock) {
+      return isEnoughForTsFileParserMemory(
+          extraMemoryInBytes, getTotalNonFloatingMemorySizeInBytes());
+    }
     if (!(block instanceof PipeTabletMemoryBlock) && !(block instanceof PipeTsFileMemoryBlock)) {
       return true;
     }
@@ -757,6 +790,39 @@ public PipeTabletMemoryBlock forceAllocateForTabletWithRetry(
         name, tabletSizeInBytes, PipeMemoryBlockCategory.EVENT_CHILD, null, parent);
   }
 
+  /** Allocates TsFile parser memory against the protected tablet + TsFile combined quota. */
+  public PipeTsFileParserMemoryBlock forceAllocateForTsFileParserWithRetry(
+      final String name, final long parserMemorySizeInBytes, final Object assigner) {
+    return forceAllocateForTsFileParserWithRetry(name, parserMemorySizeInBytes, assigner, null);
+  }
+
+  /** Allocates TsFile parser memory as an event child against the combined quota. */
+  public synchronized PipeTsFileParserMemoryBlock forceAllocateForTsFileParserWithRetry(
+      final String name,
+      final long parserMemorySizeInBytes,
+      final Object assigner,
+      final PipeMemoryBlock parent) {
+    final PipeMemoryBlockCategory category =
+        parent == null ? PipeMemoryBlockCategory.PARSER : PipeMemoryBlockCategory.EVENT_CHILD;
+    final PipeTsFileParserMemoryBlock block =
+        (PipeTsFileParserMemoryBlock)
+            registerMemoryBlock(
+                name, 0, PipeMemoryBlockType.TS_FILE_PARSER, category, assigner, parent);
+    if (!PIPE_MEMORY_MANAGEMENT_ENABLED || parserMemorySizeInBytes == 0) {
+      return block;
+    }
+
+    try {
+      // Initial allocation and subsequent growth must check the same post-allocation usage
+      // under the manager lock, including after a wait releases that lock.
+      forceResize(block, parserMemorySizeInBytes);
+      return block;
+    } catch (final RuntimeException e) {
+      release(block);
+      throw e;
+    }
+  }
+
   public PipeTsFileMemoryBlock forceAllocateForTsFileWithRetry(
       final String name, final long tsFileSizeInBytes)
       throws PipeRuntimeOutOfMemoryCriticalException {
@@ -1001,17 +1067,14 @@ public synchronized void resize(
 
     if (force) {
       throw new PipeRuntimeOutOfMemoryCriticalException(
-          String.format(
-              DataNodePipeMessages
-                  .PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC,
-              memoryAllocateMaxRetries,
-              getTotalNonFloatingMemorySizeInBytes(),
-              memoryBlock.getUsedMemoryInBytes(),
-              sizeInBytes));
+          getResizeFailureMessage(memoryAllocateMaxRetries, sizeInBytes));
     }
   }
 
   private boolean tryResizeInternal(final PipeMemoryBlock block, final long targetSize) {
+    if (block.isReleased()) {
+      return false;
+    }
     // Parent blocks expose an aggregate usage. Do not let a direct resize reduce that aggregate
     // below the bytes still owned by live children.
     final long normalizedTargetSize =
@@ -1249,6 +1312,11 @@ private synchronized PipeMemoryBlock registerMemoryBlock(
         returnedMemoryBlock =
             new PipeTabletMemoryBlock(this, name, 0, normalizedCategory, null, normalizedParent);
         break;
+      case TS_FILE_PARSER:
+        returnedMemoryBlock =
+            new PipeTsFileParserMemoryBlock(
+                this, name, 0, normalizedCategory, null, normalizedParent);
+        break;
       case TS_FILE:
         returnedMemoryBlock =
             new PipeTsFileMemoryBlock(this, name, 0, normalizedCategory, null, normalizedParent);

diff --git a/...e/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTsFileParserMemoryBlock.java b/...e/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTsFileParserMemoryBlock.java
@@ -0,0 +1,33 @@
+/*
+ * 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.resource.memory;
+
+/** Parser working memory charged to TsFiles and admitted against the protected combined quota. */
+public class PipeTsFileParserMemoryBlock extends PipeTsFileMemoryBlock {
+
+  PipeTsFileParserMemoryBlock(
+      final PipeMemoryManager pipeMemoryManager,
+      final String name,
+      final long memoryUsageInBytes,
+      final PipeMemoryBlockCategory category,
+      final String assigner,
+      final PipeMemoryBlock parent) {
+    super(pipeMemoryManager, name, memoryUsageInBytes, category, assigner, parent);
+  }
+}
-Original file line number
+Diff line change
@@ Expand Up / @@ -22,6 +22,7 @@ @@
     public enum PipeMemoryBlockType {
       NORMAL,
       TABLET,
+      TS_FILE_PARSER,
       TS_FILE,
       BATCH,
       WAL
@@ Expand Down @@