[Pipe] Add SHOW PIPE MEMORY for named memory blocks

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,21 @@ 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(
+                  "block_id,INT64,TAG,",
+                  "name,STRING,TAG,",
+                  "category,STRING,TAG,",
+                  "memory_usage_in_bytes,INT64,ATTRIBUTE,",
+                  "max_memory_size_in_bytes,INT64,ATTRIBUTE,",
+                  "allocation_time,TIMESTAMP,ATTRIBUTE,",
+                  "assigner,STRING,ATTRIBUTE,",
+                  "parent_block_id,INT64,ATTRIBUTE,",
+                  "hierarchy_level,INT32,ATTRIBUTE,",
+                  "accounted_memory_usage_in_bytes,INT64,ATTRIBUTE,")));
       TestUtils.assertResultSetEqual(
           statement.executeQuery("desc pipe_plugins"),
           "ColumnName,DataType,Category,",
@@ -731,6 +747,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 +784,40 @@ 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(10, metaData.getColumnCount());
+        assertEquals("block_id", metaData.getColumnName(1));
+        assertEquals("name", metaData.getColumnName(2));
+        assertEquals("category", metaData.getColumnName(3));
+        assertEquals("memory_usage_in_bytes", metaData.getColumnName(4));
+        assertEquals("max_memory_size_in_bytes", metaData.getColumnName(5));
+        assertEquals("allocation_time", metaData.getColumnName(6));
+        assertEquals("assigner", metaData.getColumnName(7));
+        assertEquals("parent_block_id", metaData.getColumnName(8));
+        assertEquals("hierarchy_level", metaData.getColumnName(9));
+        assertEquals("accounted_memory_usage_in_bytes", metaData.getColumnName(10));
+        boolean hasFloatingMemory = false;
+        while (resultSet.next()) {
+          if ("FloatingMemory".equals(resultSet.getString(2))) {
+            assertTrue(resultSet.getLong(4) >= 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(2))) {
+            assertTrue(resultSet.getLong(4) >= 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 +866,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 +888,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'"),

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java

@@ -45,6 +45,7 @@
 import org.apache.iotdb.db.pipe.resource.PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner;
 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.source.schemaregion.SchemaRegionListeningQueue;
 import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeDevicePathCache;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
@@ -102,7 +103,11 @@ private void initLoggerPeriodicalLogReducer() {
     if (pipeLogReducerMemoryBlock == null) {
       pipeLogReducerMemoryBlock =
           PipeDataNodeResourceManager.memory()
-              .tryAllocate(PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes());
+              .tryAllocate(
+                  PipeDataNodeRuntimeAgent.class.getSimpleName() + "#logger",
+                  PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes(),
+                  PipeMemoryBlockCategory.CACHE,
+                  PipeDataNodeRuntimeAgent.class.getSimpleName());
     }
 
     LoggerPeriodicalLogReducer.setMemoryResizeFunction(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/PipeInsertionEvent.java

@@ -26,6 +26,7 @@
 import org.apache.iotdb.commons.utils.PathUtils;
 import org.apache.iotdb.db.i18n.DataNodePipeMessages;
 import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
 
 import jakarta.validation.constraints.NotNull;
 
@@ -50,6 +51,14 @@
  */
 public abstract class PipeInsertionEvent extends EnrichedEvent {
 
+  /**
+   * Returns the event-level memory block used as the parent of parser/converted-data blocks. Events
+   * that do not retain a dedicated block return {@code null}.
+   */
+  public PipeMemoryBlock getEventMemoryBlock() {
+    return null;
+  }
+
   // Record the database name of the DataRegion corresponding to the SourceEvent
   private final String sourceDatabaseNameFromDataRegion;
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java

@@ -31,6 +31,7 @@
 import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
 import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics;
 import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory;
 import org.apache.iotdb.db.pipe.resource.memory.PipeTabletMemoryBlock;
 import org.apache.iotdb.db.queryengine.plan.statement.Statement;
 import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement;
@@ -86,7 +87,14 @@ 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,
+                PipeMemoryBlockCategory.EVENT,
+                this,
+                null);
+    this.allocatedMemoryBlock.setAssigner(this);
   }
 
   @Override
@@ -170,6 +178,11 @@ public Statement getStatement() {
     return statement;
   }
 
+  @Override
+  public PipeTabletMemoryBlock getEventMemoryBlock() {
+    return allocatedMemoryBlock;
+  }
+
   /////////////////////////// Object ///////////////////////////
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java

@@ -44,6 +44,8 @@
 import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics;
 import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
 import org.apache.iotdb.db.pipe.resource.memory.InsertNodeMemoryEstimator;
+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.PipeMemoryWeightUtil;
 import org.apache.iotdb.db.pipe.resource.memory.PipeTabletMemoryBlock;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
@@ -93,6 +95,7 @@ public class PipeInsertNodeTabletInsertionEvent extends PipeInsertionEvent
           + RamUsageEstimator.shallowSizeOf(Boolean.class);
 
   private final AtomicReference<PipeTabletMemoryBlock> allocatedMemoryBlock;
+  private final PipeMemoryBlock eventMemoryBlock;
   private volatile List<Tablet> tablets;
   // Calculated together with tablets so downstream batching does not rescan Tablet internals.
   private volatile long tabletsMemoryUsageInBytes;
@@ -159,6 +162,14 @@ public PipeInsertNodeTabletInsertionEvent(
     this.insertNode = insertNode;
     this.progressIndex = insertNode.getProgressIndex();
 
+    this.eventMemoryBlock =
+        PipeDataNodeResourceManager.memory()
+            .forceAllocate(
+                PipeInsertNodeTabletInsertionEvent.class.getSimpleName(),
+                0,
+                PipeMemoryBlockCategory.EVENT,
+                this,
+                null);
     this.allocatedMemoryBlock = new AtomicReference<>();
   }
 
@@ -498,7 +509,12 @@ public synchronized List<Tablet> convertToTablets() {
       allocatedMemoryBlock.compareAndSet(
           null,
           PipeDataNodeResourceManager.memory()
-              .forceAllocateForTabletWithRetry(tabletMemoryUsageInBytes));
+              .forceAllocateForTabletWithRetry(
+                  PipeInsertNodeTabletInsertionEvent.class.getSimpleName(),
+                  tabletMemoryUsageInBytes,
+                  PipeMemoryBlockCategory.TABLET,
+                  this,
+                  eventMemoryBlock));
     }
     return tablets;
   }
@@ -508,6 +524,11 @@ public long getTabletsMemoryUsageInBytes() {
     return tabletsMemoryUsageInBytes;
   }
 
+  @Override
+  public PipeMemoryBlock getEventMemoryBlock() {
+    return eventMemoryBlock;
+  }
+
   /////////////////////////// event parser ///////////////////////////
 
   private List<TabletInsertionEventParser> initEventParsers() {
@@ -653,7 +674,7 @@ protected void trackResource() {
   @Override
   public PipeEventResource eventResourceBuilder() {
     return new PipeInsertNodeTabletInsertionEventResource(
-        this.isReleased, this.referenceCount, this.allocatedMemoryBlock);
+        this.isReleased, this.referenceCount, this.allocatedMemoryBlock, this.eventMemoryBlock);
   }
 
   // Notes:
@@ -677,13 +698,16 @@ public long ramBytesUsed() {
   private static class PipeInsertNodeTabletInsertionEventResource extends PipeEventResource {
 
     private final AtomicReference<PipeTabletMemoryBlock> allocatedMemoryBlock;
+    private final PipeMemoryBlock eventMemoryBlock;
 
     private PipeInsertNodeTabletInsertionEventResource(
         final AtomicBoolean isReleased,
         final AtomicInteger referenceCount,
-        final AtomicReference<PipeTabletMemoryBlock> allocatedMemoryBlock) {
+        final AtomicReference<PipeTabletMemoryBlock> allocatedMemoryBlock,
+        final PipeMemoryBlock eventMemoryBlock) {
       super(isReleased, referenceCount);
       this.allocatedMemoryBlock = allocatedMemoryBlock;
+      this.eventMemoryBlock = eventMemoryBlock;
     }
 
     @Override
@@ -696,6 +720,7 @@ protected void finalizeResource() {
               }
               return null;
             });
+        eventMemoryBlock.close();
       } catch (final Exception e) {
         LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_ERROR, e);
       }
@@ -713,6 +738,7 @@ public synchronized void close() {
           }
           return null;
         });
+    eventMemoryBlock.close();
     tablets = null;
   }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java

@@ -38,6 +38,8 @@
 import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
 import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics;
 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.PipeMemoryWeightUtil;
 import org.apache.iotdb.db.pipe.resource.memory.PipeTabletMemoryBlock;
 import org.apache.iotdb.pipe.api.access.Row;
@@ -115,8 +117,21 @@ private PipeRawTabletInsertionEvent(
     inheritSourceEventReportSkippingIfNecessary();
 
     // Allocate empty memory block, will be resized later.
+    final PipeMemoryBlock parentMemoryBlock =
+        sourceEvent instanceof PipeInsertionEvent
+            ? ((PipeInsertionEvent) sourceEvent).getEventMemoryBlock()
+            : null;
     this.allocatedMemoryBlock =
-        PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
+        PipeDataNodeResourceManager.memory()
+            .forceAllocateForTabletWithRetry(
+                PipeRawTabletInsertionEvent.class.getSimpleName(),
+                0,
+                parentMemoryBlock == null
+                    ? PipeMemoryBlockCategory.EVENT
+                    : PipeMemoryBlockCategory.TABLET,
+                this,
+                parentMemoryBlock);
+    this.allocatedMemoryBlock.setAssigner(this);
 
     if (needToReport) {
       addOnCommittedHook(
@@ -269,6 +284,11 @@ public boolean internallyIncreaseResourceReferenceCount(final String holderMessa
     return true;
   }
 
+  @Override
+  public PipeTabletMemoryBlock getEventMemoryBlock() {
+    return allocatedMemoryBlock;
+  }
+
   @Override
   public boolean internallyDecreaseResourceReferenceCount(final String holderMessage) {
     if (Objects.nonNull(pipeName)) {