(iotdb) branch master updated: Pipe: Fixed the bug that batch memory is not visible to pipe memory management (#16962)

[jiangtian]

This is an automated email from the ASF dual-hosted git repository.

jiangtian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new b75bcc62f9b Pipe: Fixed the bug that batch memory is not visible to 
pipe memory management (#16962)
b75bcc62f9b is described below

commit b75bcc62f9bfa4b945d7e468356b5f4a3cd49ee2
Author: Caideyipi <87789683+caidey...@users.noreply.github.com>
AuthorDate: Tue Dec 30 11:52:18 2025 +0800

    Pipe: Fixed the bug that batch memory is not visible to pipe memory 
management (#16962)
---
 .../db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java  | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
index 0e13feb8ac4..96bddd0d672 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
@@ -20,6 +20,8 @@
 package org.apache.iotdb.db.pipe.sink.payload.evolvable.batch;
 
 import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
 import 
org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink;
 import 
org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException;
 import org.apache.iotdb.pipe.api.event.Event;
@@ -45,6 +47,7 @@ public abstract class PipeTabletEventBatch implements 
AutoCloseable {
   private long firstEventProcessingTime = Long.MIN_VALUE;
 
   protected long totalBufferSize = 0;
+  private final PipeMemoryBlock allocatedMemoryBlock;
 
   protected volatile boolean isClosed = false;
 
@@ -56,6 +59,8 @@ public abstract class PipeTabletEventBatch implements 
AutoCloseable {
 
     // limit in buffer size
     this.maxBatchSizeInBytes = requestMaxBatchSizeInBytes;
+    this.allocatedMemoryBlock =
+        
PipeDataNodeResourceManager.memory().forceAllocate(requestMaxBatchSizeInBytes);
     if (recordMetric != null) {
       this.recordMetric = recordMetric;
     } else {
@@ -142,6 +147,7 @@ public abstract class PipeTabletEventBatch implements 
AutoCloseable {
 
     clearEventsReferenceCount(PipeTabletEventBatch.class.getName());
     events.clear();
+    allocatedMemoryBlock.close();
   }
 
   /**