apache · Caideyipi · Sep 3, 2026
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java
@@ -535,6 +535,7 @@ public void testInformationSchema() throws SQLException {
               "functions,INF,",
               "keywords,INF,",
               "nodes,INF,",
+              "pipe_memory,INF,",
               "pipe_plugins,INF,",
               "pipes,INF,",
               "queries,INF,",
@@ -616,6 +617,11 @@ public void testInformationSchema() throws SQLException {
                   "estimated_remaining_seconds,DOUBLE,ATTRIBUTE,",
                   "is_degraded,BOOLEAN,ATTRIBUTE,",
                   "recent_failures,STRING,ATTRIBUTE,")));
+      TestUtils.assertResultSetEqual(
+          statement.executeQuery("desc pipe_memory"),
+          "ColumnName,DataType,Category,",
+          new HashSet<>(
+              Arrays.asList("name,STRING,TAG,", "memory_usage_in_bytes,INT64,ATTRIBUTE,")));
       TestUtils.assertResultSetEqual(
           statement.executeQuery("desc pipe_plugins"),
           "ColumnName,DataType,Category,",
@@ -731,6 +737,9 @@ public void testInformationSchema() throws SQLException {
       Assert.assertThrows(SQLException.class, () -> statement.execute("select * from data_nodes"));
       Assert.assertThrows(
           SQLException.class, () -> statement.executeQuery("select * from pipe_plugins"));
+      Assert.assertThrows(
+          SQLException.class, () -> statement.executeQuery("select * from pipe_memory"));
+      Assert.assertThrows(SQLException.class, () -> statement.executeQuery("SHOW PIPE MEMORY"));
       Assert.assertThrows(
           SQLException.class, () -> statement.executeQuery("select * from table_disk_usage"));
 
@@ -765,6 +774,32 @@ public void testInformationSchema() throws SQLException {
       // Test table query
       statement.execute("use information_schema");
 
+      try (final ResultSet resultSet = statement.executeQuery("SHOW PIPE MEMORY")) {
+        final ResultSetMetaData metaData = resultSet.getMetaData();
+        assertEquals(2, metaData.getColumnCount());
+        assertEquals("name", metaData.getColumnName(1));
+        assertEquals("memory_usage_in_bytes", metaData.getColumnName(2));
+        boolean hasFloatingMemory = false;
+        while (resultSet.next()) {
+          if ("FloatingMemory".equals(resultSet.getString(1))) {
+            assertTrue(resultSet.getLong(2) >= 0);
+            hasFloatingMemory = true;
+          }
+        }
+        assertTrue(hasFloatingMemory);
+      }
+      try (final ResultSet resultSet =
+          statement.executeQuery("select * from information_schema.pipe_memory")) {
+        boolean hasFloatingMemory = false;
+        while (resultSet.next()) {
+          if ("FloatingMemory".equals(resultSet.getString(1))) {
+            assertTrue(resultSet.getLong(2) >= 0);
+            hasFloatingMemory = true;
+          }
+        }
+        assertTrue(hasFloatingMemory);
+      }
+
       statement.execute("create database test");
       statement.execute(
           "create table test.test (a tag, b attribute, c int32 comment 'turbine') comment 'test'");
@@ -813,6 +848,7 @@ public void testInformationSchema() throws SQLException {
                   "information_schema,columns,INF,USING,null,SYSTEM VIEW,false,",
                   "information_schema,queries,INF,USING,null,SYSTEM VIEW,false,",
                   "information_schema,regions,INF,USING,null,SYSTEM VIEW,false,",
+                  "information_schema,pipe_memory,INF,USING,null,SYSTEM VIEW,false,",
                   "information_schema,topics,INF,USING,null,SYSTEM VIEW,false,",
                   "information_schema,pipe_plugins,INF,USING,null,SYSTEM VIEW,false,",
                   "information_schema,pipes,INF,USING,null,SYSTEM VIEW,false,",
@@ -834,7 +870,7 @@ public void testInformationSchema() throws SQLException {
       TestUtils.assertResultSetEqual(
           statement.executeQuery("count devices from tables where status = 'USING'"),
           "count(devices),",
-          Collections.singleton("23,"));
+          Collections.singleton("24,"));
       TestUtils.assertResultSetEqual(
           statement.executeQuery(
               "select * from columns where table_name = 'queries' or database = 'test'"),

diff --git a/...tanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java b/...tanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java
@@ -102,7 +102,9 @@ private void initLoggerPeriodicalLogReducer() {
     if (pipeLogReducerMemoryBlock == null) {
       pipeLogReducerMemoryBlock =
           PipeDataNodeResourceManager.memory()
-              .tryAllocate(PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes());
+              .tryAllocate(
+                  PipeDataNodeRuntimeAgent.class.getSimpleName() + "#logger",
+                  PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes());
     }
 
     LoggerPeriodicalLogReducer.setMemoryResizeFunction(

diff --git a/...ain/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java b/...ain/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java
@@ -86,7 +86,8 @@ public PipeStatementInsertionEvent(
     this.statement = statement;
     // Allocate empty memory block, will be resized later.
     this.allocatedMemoryBlock =
-        PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
+        PipeDataNodeResourceManager.memory()
+            .forceAllocateForTabletWithRetry(PipeStatementInsertionEvent.class.getSimpleName(), 0);
   }
 
   @Override

diff --git a/...java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java b/...java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
@@ -498,7 +498,9 @@ public synchronized List<Tablet> convertToTablets() {
       allocatedMemoryBlock.compareAndSet(
           null,
           PipeDataNodeResourceManager.memory()
-              .forceAllocateForTabletWithRetry(tabletMemoryUsageInBytes));
+              .forceAllocateForTabletWithRetry(
+                  PipeInsertNodeTabletInsertionEvent.class.getSimpleName(),
+                  tabletMemoryUsageInBytes));
     }
     return tablets;
   }

diff --git a/...c/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java b/...c/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
@@ -116,7 +116,8 @@ private PipeRawTabletInsertionEvent(
 
     // Allocate empty memory block, will be resized later.
     this.allocatedMemoryBlock =
-        PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
+        PipeDataNodeResourceManager.memory()
+            .forceAllocateForTabletWithRetry(PipeRawTabletInsertionEvent.class.getSimpleName(), 0);
 
     if (needToReport) {
       addOnCommittedHook(

diff --git a/.../java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParser.java b/.../java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParser.java
@@ -138,7 +138,9 @@ protected TsFileInsertionEventParser(
     this.sourceEvent = sourceEvent;
     this.memoryManager = memoryManager;
 
-    this.allocatedMemoryBlockForTablet = memoryManager.forceAllocateForTabletWithRetry(0);
+    this.allocatedMemoryBlockForTablet =
+        memoryManager.forceAllocateForTabletWithRetry(
+            TsFileInsertionEventParser.class.getSimpleName() + "#tablet", 0);
 
     LOGGER.debug(
         DataNodePipeMessages.TSFILE_HAS_INITIALIZED_PIPENAME_CREATION_TIME_PATTERN,

diff --git a/...che/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java b/...che/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java
@@ -25,9 +25,10 @@
 /** Allocates parser working memory from the pool owned by the caller. */
 public interface TsFileInsertionEventParserMemoryManager {
 
-  TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry(long sizeInBytes);
+  TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry(
+      String name, long sizeInBytes);
 
-  TsFileInsertionEventParserMemoryBlock forceAllocate(long sizeInBytes);
+  TsFileInsertionEventParserMemoryBlock forceAllocate(String name, long sizeInBytes);
 
   static TsFileInsertionEventParserMemoryManager pipe() {
     return PipeHolder.INSTANCE;
@@ -38,14 +39,17 @@ final class PipeHolder {
         new TsFileInsertionEventParserMemoryManager() {
           @Override
           public TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry(
-              final long sizeInBytes) {
+              final String name, final long sizeInBytes) {
             return new PipeBlock(
-                PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(sizeInBytes));
+                PipeDataNodeResourceManager.memory()
+                    .forceAllocateForTabletWithRetry(name, sizeInBytes));
           }
 
           @Override
-          public TsFileInsertionEventParserMemoryBlock forceAllocate(final long sizeInBytes) {
-            return new PipeBlock(PipeDataNodeResourceManager.memory().forceAllocate(sizeInBytes));
+          public TsFileInsertionEventParserMemoryBlock forceAllocate(
+              final String name, final long sizeInBytes) {
+            return new PipeBlock(
+                PipeDataNodeResourceManager.memory().forceAllocate(name, sizeInBytes));
           }
         };
   }

diff --git a/...pache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java b/...pache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java
@@ -221,7 +221,9 @@ public TsFileInsertionEventQueryParser(
               ? ModsOperationUtil.loadModificationsFromTsFile(tsFile)
               : PatternTreeMapFactory.getModsPatternTreeMap();
       allocatedMemoryBlockForModifications =
-          memoryManager.forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed());
+          memoryManager.forceAllocateForTabletWithRetry(
+              TsFileInsertionEventQueryParser.class.getSimpleName() + "#modifications",
+              currentModifications.ramBytesUsed());
 
       final PipeTsFileResourceManager tsFileResourceManager = PipeDataNodeResourceManager.tsfile();
       final Map<IDeviceID, List<String>> deviceMeasurementsMap;
@@ -282,7 +284,10 @@ public TsFileInsertionEventQueryParser(
         memoryRequiredInBytes +=
             PipeMemoryWeightUtil.memoryOfIDeviceID2StrList(deviceMeasurementsMap);
       }
-      allocatedMemoryBlock = memoryManager.forceAllocate(memoryRequiredInBytes);
+      allocatedMemoryBlock =
+          memoryManager.forceAllocate(
+              TsFileInsertionEventQueryParser.class.getSimpleName() + "#metadata",
+              memoryRequiredInBytes);
 
       final Iterator<Map.Entry<IDeviceID, List<String>>> iterator =
           deviceMeasurementsMap.entrySet().iterator();

diff --git a/.../apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java b/.../apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java
@@ -181,16 +181,22 @@ public TsFileInsertionEventScanParser(
     this.endTime = endTime;
     filter = Objects.nonNull(timeFilterExpression) ? timeFilterExpression.getFilter() : null;
 
-    this.allocatedMemoryBlockForBatchData = memoryManager.forceAllocateForTabletWithRetry(0);
-    this.allocatedMemoryBlockForChunk = memoryManager.forceAllocateForTabletWithRetry(0);
+    this.allocatedMemoryBlockForBatchData =
+        memoryManager.forceAllocateForTabletWithRetry(
+            TsFileInsertionEventScanParser.class.getSimpleName() + "#batchData", 0);
+    this.allocatedMemoryBlockForChunk =
+        memoryManager.forceAllocateForTabletWithRetry(
+            TsFileInsertionEventScanParser.class.getSimpleName() + "#chunk", 0);
 
     try {
       currentModifications =
           isWithMod
               ? ModsOperationUtil.loadModificationsFromTsFile(tsFile)
               : PatternTreeMapFactory.getModsPatternTreeMap();
       allocatedMemoryBlockForModifications =
-          memoryManager.forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed());
+          memoryManager.forceAllocateForTabletWithRetry(
+              TsFileInsertionEventScanParser.class.getSimpleName() + "#modifications",
+              currentModifications.ramBytesUsed());
 
       tsFileSequenceReader = createTsFileSequenceReader(tsFile, !currentModifications.isEmpty());
       tsFileSequenceReader.position((long) TSFileConfig.MAGIC_STRING.getBytes().length + 1);
@@ -257,7 +263,9 @@ private TsFileSequenceReader createTsFileSequenceReader(
     }
 
     allocatedMemoryBlockForTsFileInput =
-        memoryManager.forceAllocateForTabletWithRetry(TS_FILE_INPUT_BUFFER_SIZE_IN_BYTES);
+        memoryManager.forceAllocateForTabletWithRetry(
+            TsFileInsertionEventScanParser.class.getSimpleName() + "#tsFileInput",
+            TS_FILE_INPUT_BUFFER_SIZE_IN_BYTES);
     return new TsFileSequenceReader(
         new BufferedTsFileInput(tsFile.toPath(), TS_FILE_INPUT_BUFFER_SIZE_IN_BYTES),
         false,

diff --git a/...pache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java b/...pache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java
@@ -119,11 +119,21 @@ public TsFileInsertionEventTableParser(
               ? ModsOperationUtil.loadModificationsFromTsFile(tsFile)
               : PatternTreeMapFactory.getModsPatternTreeMap();
       allocatedMemoryBlockForModifications =
-          memoryManager.forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed());
-      this.allocatedMemoryBlockForChunk = memoryManager.forceAllocateForTabletWithRetry(0);
-      this.allocatedMemoryBlockForBatchData = memoryManager.forceAllocateForTabletWithRetry(0);
-      this.allocatedMemoryBlockForChunkMeta = memoryManager.forceAllocateForTabletWithRetry(0);
-      this.allocatedMemoryBlockForTableSchemas = memoryManager.forceAllocateForTabletWithRetry(0);
+          memoryManager.forceAllocateForTabletWithRetry(
+              TsFileInsertionEventTableParser.class.getSimpleName() + "#modifications",
+              currentModifications.ramBytesUsed());
+      this.allocatedMemoryBlockForChunk =
+          memoryManager.forceAllocateForTabletWithRetry(
+              TsFileInsertionEventTableParser.class.getSimpleName() + "#chunk", 0);
+      this.allocatedMemoryBlockForBatchData =
+          memoryManager.forceAllocateForTabletWithRetry(
+              TsFileInsertionEventTableParser.class.getSimpleName() + "#batchData", 0);
+      this.allocatedMemoryBlockForChunkMeta =
+          memoryManager.forceAllocateForTabletWithRetry(
+              TsFileInsertionEventTableParser.class.getSimpleName() + "#chunkMetadata", 0);
+      this.allocatedMemoryBlockForTableSchemas =
+          memoryManager.forceAllocateForTabletWithRetry(
+              TsFileInsertionEventTableParser.class.getSimpleName() + "#tableSchemas", 0);
 
       this.startTime = startTime;
       this.endTime = endTime;

diff --git a/...main/java/org/apache/iotdb/db/pipe/processor/downsampling/PartialPathLastObjectCache.java b/...main/java/org/apache/iotdb/db/pipe/processor/downsampling/PartialPathLastObjectCache.java
@@ -39,7 +39,9 @@ public abstract class PartialPathLastObjectCache<T> implements AutoCloseable {
   private final Cache<String, T> partialPath2ObjectCache;
 
   protected PartialPathLastObjectCache(final long memoryLimitInBytes) {
-    allocatedMemoryBlock = PipeDataNodeResourceManager.memory().tryAllocate(memoryLimitInBytes);
+    allocatedMemoryBlock =
+        PipeDataNodeResourceManager.memory()
+            .tryAllocate(PartialPathLastObjectCache.class.getSimpleName(), memoryLimitInBytes);
 
     // Currently disable the metric here because it's not a constant cache and the number may
     // fluctuate. In the future all the "processorCache"s may be recorded in single metric entry

diff --git a/...rc/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java b/...rc/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
@@ -1109,7 +1109,9 @@ private void closeMemoryBlock(final PipeMemoryBlock memoryBlock) {
   private PipeMemoryBlock tryAllocateReceiverMemory(final long requestedMemorySizeInBytes)
       throws PipeRuntimeOutOfMemoryCriticalException {
     return PipeDataNodeResourceManager.memory()
-        .forceAllocate(Math.max(requestedMemorySizeInBytes, 0));
+        .forceAllocate(
+            IoTDBDataNodeReceiver.class.getSimpleName() + "#request",
+            Math.max(requestedMemorySizeInBytes, 0));
   }
 
   @Override
@@ -1189,6 +1191,7 @@ private TSStatus executeStatementAndClassifyExceptions(
             allocatedMemoryBlock =
                 PipeDataNodeResourceManager.memory()
                     .forceAllocate(
+                        IoTDBDataNodeReceiver.class.getSimpleName() + "#statement",
                         (long) (estimatedMemory * pipeReceiverActualToEstimatedMemoryRatio));
             break;
           } catch (final PipeRuntimeOutOfMemoryCriticalException e) {

diff --git a/...datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeFixedMemoryBlock.java b/...datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeFixedMemoryBlock.java
@@ -26,8 +26,8 @@
 
 public abstract class PipeFixedMemoryBlock extends PipeMemoryBlock {
 
-  public PipeFixedMemoryBlock(long memoryUsageInBytes) {
-    super(memoryUsageInBytes);
+  public PipeFixedMemoryBlock(final String name, final long memoryUsageInBytes) {
+    super(name, memoryUsageInBytes);
   }
 
   @Override

diff --git a/...core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlock.java b/...core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlock.java
@@ -41,6 +41,8 @@ public class PipeMemoryBlock implements AutoCloseable {
 
   private final ReentrantLock lock = new ReentrantLock();
 
+  private final String name;
+
   private final AtomicLong memoryUsageInBytes = new AtomicLong(0);
 
   private final AtomicReference<LongUnaryOperator> shrinkMethod = new AtomicReference<>();
@@ -50,10 +52,15 @@ public class PipeMemoryBlock implements AutoCloseable {
 
   private volatile boolean isReleased = false;
 
-  public PipeMemoryBlock(final long memoryUsageInBytes) {
+  public PipeMemoryBlock(final String name, final long memoryUsageInBytes) {
+    this.name = Objects.requireNonNull(name);
     this.memoryUsageInBytes.set(memoryUsageInBytes);
   }
 
+  public String getName() {
+    return name;
+  }
+
   public long getMemoryUsageInBytes() {
     return memoryUsageInBytes.get();
   }
@@ -165,7 +172,10 @@ void markAsReleased() {
   @Override
   public String toString() {
     return "PipeMemoryBlock{"
-        + "usedMemoryInBytes="
+        + "name='"
+        + name
+        + '\''
+        + ", usedMemoryInBytes="
         + memoryUsageInBytes.get()
         + ", isReleased="
         + isReleased