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 index 3d24915c1fed4..ac3500f236462 100644 --- 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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java index 83c115a704e19..ff2e3aba009bc 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java +++ b/iotdb-core/datanode/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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java index 9f812d90b3c8b..c4b2e32c8e0fe 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java +++ b/iotdb-core/datanode/src/main/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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java index 3bebbc9115305..500ce6c1884f4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java @@ -498,7 +498,9 @@ public synchronized List convertToTablets() { allocatedMemoryBlock.compareAndSet( null, PipeDataNodeResourceManager.memory() - .forceAllocateForTabletWithRetry(tabletMemoryUsageInBytes)); + .forceAllocateForTabletWithRetry( + PipeInsertNodeTabletInsertionEvent.class.getSimpleName(), + tabletMemoryUsageInBytes)); } return tablets; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java index dc2ab1d381fd8..da92db058931d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java +++ b/iotdb-core/datanode/src/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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParser.java index eefcafec9c53b..e50baaada581a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParser.java +++ b/iotdb-core/datanode/src/main/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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java index 68669ce4b55d8..df32157baf273 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java index c26910b426f18..e2f68e79535b1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java +++ b/iotdb-core/datanode/src/main/java/org/apache/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> 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>> iterator = deviceMeasurementsMap.entrySet().iterator(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java index e18494ea492c5..cf6906cb2003b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java @@ -181,8 +181,12 @@ 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 = @@ -190,7 +194,9 @@ public TsFileInsertionEventScanParser( ? 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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java index 0fd7470e0e21f..5d46c3b15fceb 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java +++ b/iotdb-core/datanode/src/main/java/org/apache/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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/PartialPathLastObjectCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/PartialPathLastObjectCache.java index 225e0c3e86eb8..499ec411cca6c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/PartialPathLastObjectCache.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/PartialPathLastObjectCache.java @@ -39,7 +39,9 @@ public abstract class PartialPathLastObjectCache implements AutoCloseable { private final Cache 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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java index e7d4053ea64ae..2b9285c56990f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java +++ b/iotdb-core/datanode/src/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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeFixedMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeFixedMemoryBlock.java index 47073fbbedd13..0aa326c392bfd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeFixedMemoryBlock.java +++ b/iotdb-core/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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlock.java index 72ba8e35ea0f4..12954b4637bed 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlock.java +++ b/iotdb-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 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 diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java index 90a45c1542ec3..69bb726eae7ca 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java @@ -34,6 +34,7 @@ import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Collections; +import java.util.Comparator; import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashSet; @@ -41,6 +42,7 @@ import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.WeakHashMap; import java.util.function.LongSupplier; import java.util.function.LongUnaryOperator; @@ -48,6 +50,8 @@ public class PipeMemoryManager { private static final Logger LOGGER = LoggerFactory.getLogger(PipeMemoryManager.class); + public static final String FLOATING_MEMORY_BLOCK_NAME = "FloatingMemory"; + private static final PipeConfig PIPE_CONFIG = PipeConfig.getInstance(); private static final boolean PIPE_MEMORY_MANAGEMENT_ENABLED = @@ -76,7 +80,11 @@ public class PipeMemoryManager { private final ArrayDeque waitingTsFileParserPipeOrder = new ArrayDeque<>(); private PipeIdentity lastAdmittedWaitingTsFileParserPipe; - // Only non-zero memory blocks will be added to this set. + // All reachable unreleased memory blocks, including zero-sized blocks, are kept for inspection. + // A weak set avoids retaining an otherwise unreachable zero-sized block solely for diagnostics. + private final Set memoryBlocks = Collections.newSetFromMap(new WeakHashMap<>()); + + // Only non-zero memory blocks will be added to this set for memory accounting. private final Set allocatedBlocks = new HashSet<>(); private final Set shrinkableBlocks = new HashSet<>(); private final Set expandableBlocks = new HashSet<>(); @@ -503,29 +511,30 @@ < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() return true; } - public synchronized PipeMemoryBlock forceAllocate(long sizeInBytes) + public synchronized PipeMemoryBlock forceAllocate(final String name, final long sizeInBytes) throws PipeRuntimeOutOfMemoryCriticalException { if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { // No need to calculate the tablet size, skip it to save time - return new PipeMemoryBlock(0); + return registerMemoryBlock(name, 0); } if (sizeInBytes == 0) { - return registerMemoryBlock(0); + return registerMemoryBlock(name, 0); } - return forceAllocateWithRetry(sizeInBytes, PipeMemoryBlockType.NORMAL); + return forceAllocateWithRetry(name, sizeInBytes, PipeMemoryBlockType.NORMAL); } - public PipeTabletMemoryBlock forceAllocateForTabletWithRetry(long tabletSizeInBytes) + public PipeTabletMemoryBlock forceAllocateForTabletWithRetry( + final String name, final long tabletSizeInBytes) throws PipeRuntimeOutOfMemoryCriticalException { if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { // No need to calculate the tablet size, skip it to save time - return new PipeTabletMemoryBlock(0); + return (PipeTabletMemoryBlock) registerMemoryBlock(name, 0, PipeMemoryBlockType.TABLET); } if (tabletSizeInBytes == 0) { - return (PipeTabletMemoryBlock) registerMemoryBlock(0, PipeMemoryBlockType.TABLET); + return (PipeTabletMemoryBlock) registerMemoryBlock(name, 0, PipeMemoryBlockType.TABLET); } for (int i = 1, size = PIPE_CONFIG.getPipeMemoryAllocateMaxRetries(); i <= size; i++) { @@ -557,20 +566,21 @@ public PipeTabletMemoryBlock forceAllocateForTabletWithRetry(long tabletSizeInBy synchronized (this) { final PipeTabletMemoryBlock block = (PipeTabletMemoryBlock) - forceAllocateWithRetry(tabletSizeInBytes, PipeMemoryBlockType.TABLET); + forceAllocateWithRetry(name, tabletSizeInBytes, PipeMemoryBlockType.TABLET); usedMemorySizeInBytesOfTablets += block.getMemoryUsageInBytes(); return block; } } - public PipeTsFileMemoryBlock forceAllocateForTsFileWithRetry(long tsFileSizeInBytes) + public PipeTsFileMemoryBlock forceAllocateForTsFileWithRetry( + final String name, final long tsFileSizeInBytes) throws PipeRuntimeOutOfMemoryCriticalException { if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { - return new PipeTsFileMemoryBlock(0); + return (PipeTsFileMemoryBlock) registerMemoryBlock(name, 0, PipeMemoryBlockType.TS_FILE); } if (tsFileSizeInBytes == 0) { - return (PipeTsFileMemoryBlock) registerMemoryBlock(0, PipeMemoryBlockType.TS_FILE); + return (PipeTsFileMemoryBlock) registerMemoryBlock(name, 0, PipeMemoryBlockType.TS_FILE); } for (int i = 1, size = PIPE_CONFIG.getPipeMemoryAllocateMaxRetries(); i <= size; i++) { @@ -602,21 +612,21 @@ public PipeTsFileMemoryBlock forceAllocateForTsFileWithRetry(long tsFileSizeInBy synchronized (this) { final PipeTsFileMemoryBlock block = (PipeTsFileMemoryBlock) - forceAllocateWithRetry(tsFileSizeInBytes, PipeMemoryBlockType.TS_FILE); + forceAllocateWithRetry(name, tsFileSizeInBytes, PipeMemoryBlockType.TS_FILE); usedMemorySizeInBytesOfTsFiles += block.getMemoryUsageInBytes(); return block; } } public PipeModelFixedMemoryBlock forceAllocateForModelFixedMemoryBlock( - long fixedSizeInBytes, PipeMemoryBlockType type) + final String name, final long fixedSizeInBytes, final PipeMemoryBlockType type) throws PipeRuntimeOutOfMemoryCriticalException { if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { - return new PipeModelFixedMemoryBlock(Long.MAX_VALUE, new ThresholdAllocationStrategy()); + return (PipeModelFixedMemoryBlock) registerMemoryBlock(name, Long.MAX_VALUE, type); } if (fixedSizeInBytes == 0) { - return (PipeModelFixedMemoryBlock) registerMemoryBlock(0, type); + return (PipeModelFixedMemoryBlock) registerMemoryBlock(name, 0, type); } for (int i = 1, size = PIPE_CONFIG.getPipeMemoryAllocateMaxRetries(); i <= size; i++) { @@ -637,34 +647,26 @@ public PipeModelFixedMemoryBlock forceAllocateForModelFixedMemoryBlock( synchronized (this) { if (getFreeMemorySizeInBytes() < fixedSizeInBytes) { - return (PipeModelFixedMemoryBlock) forceAllocateWithRetry(getFreeMemorySizeInBytes(), type); + return (PipeModelFixedMemoryBlock) + forceAllocateWithRetry(name, getFreeMemorySizeInBytes(), type); } - return (PipeModelFixedMemoryBlock) forceAllocateWithRetry(fixedSizeInBytes, type); + return (PipeModelFixedMemoryBlock) forceAllocateWithRetry(name, fixedSizeInBytes, type); } } - private PipeMemoryBlock forceAllocateWithRetry(long sizeInBytes, PipeMemoryBlockType type) + private PipeMemoryBlock forceAllocateWithRetry( + final String name, final long sizeInBytes, final PipeMemoryBlockType type) throws PipeRuntimeOutOfMemoryCriticalException { if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { - switch (type) { - case TABLET: - return new PipeTabletMemoryBlock(sizeInBytes); - case TS_FILE: - return new PipeTsFileMemoryBlock(sizeInBytes); - case BATCH: - case WAL: - return new PipeModelFixedMemoryBlock(sizeInBytes, new ThresholdAllocationStrategy()); - default: - return new PipeMemoryBlock(sizeInBytes); - } + return registerMemoryBlock(name, sizeInBytes, type); } final int memoryAllocateMaxRetries = PIPE_CONFIG.getPipeMemoryAllocateMaxRetries(); for (int i = 1; i <= memoryAllocateMaxRetries; i++) { if (getTotalNonFloatingMemorySizeInBytes() - memoryBlock.getUsedMemoryInBytes() >= sizeInBytes) { - return registerMemoryBlock(sizeInBytes, type); + return registerMemoryBlock(name, sizeInBytes, type); } try { @@ -783,41 +785,41 @@ && getTotalNonFloatingMemorySizeInBytes() - memoryBlock.getUsedMemoryInBytes() * usedThreshold}. Will return a memory block otherwise. */ public synchronized PipeMemoryBlock forceAllocateIfSufficient( - long sizeInBytes, float usedThreshold) { + final String name, final long sizeInBytes, final float usedThreshold) { if (usedThreshold < 0.0f || usedThreshold > 1.0f) { return null; } if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { - return new PipeMemoryBlock(sizeInBytes); + return registerMemoryBlock(name, sizeInBytes); } if (sizeInBytes == 0) { - return registerMemoryBlock(0); + return registerMemoryBlock(name, 0); } if ((float) (memoryBlock.getUsedMemoryInBytes() + sizeInBytes) <= getTotalNonFloatingMemorySizeInBytes() * usedThreshold) { - return forceAllocate(sizeInBytes); + return forceAllocate(name, sizeInBytes); } return null; } - public synchronized PipeMemoryBlock tryAllocate(long sizeInBytes) { - return tryAllocate(sizeInBytes, currentSize -> currentSize * 2 / 3); + public synchronized PipeMemoryBlock tryAllocate(final String name, final long sizeInBytes) { + return tryAllocate(name, sizeInBytes, currentSize -> currentSize * 2 / 3); } public synchronized PipeMemoryBlock tryAllocate( - long sizeInBytes, LongUnaryOperator customAllocateStrategy) { + final String name, final long sizeInBytes, final LongUnaryOperator customAllocateStrategy) { if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { - return new PipeMemoryBlock(sizeInBytes); + return registerMemoryBlock(name, sizeInBytes); } if (sizeInBytes == 0 || getTotalNonFloatingMemorySizeInBytes() - memoryBlock.getUsedMemoryInBytes() >= sizeInBytes) { - return registerMemoryBlock(sizeInBytes); + return registerMemoryBlock(name, sizeInBytes); } long sizeToAllocateInBytes = sizeInBytes; @@ -832,7 +834,7 @@ public synchronized PipeMemoryBlock tryAllocate( memoryBlock.getUsedMemoryInBytes(), sizeInBytes, sizeToAllocateInBytes); - return registerMemoryBlock(sizeToAllocateInBytes); + return registerMemoryBlock(name, sizeToAllocateInBytes); } sizeToAllocateInBytes = @@ -848,14 +850,14 @@ public synchronized PipeMemoryBlock tryAllocate( memoryBlock.getUsedMemoryInBytes(), sizeInBytes, sizeToAllocateInBytes); - return registerMemoryBlock(sizeToAllocateInBytes); + return registerMemoryBlock(name, sizeToAllocateInBytes); } else { LOGGER.warn( DataNodePipeMessages.TRYALLOCATE_FAILED_TO_ALLOCATE_MEMORY_TOTAL_MEMORY, getTotalNonFloatingMemorySizeInBytes(), memoryBlock.getUsedMemoryInBytes(), sizeInBytes); - return registerMemoryBlock(0); + return registerMemoryBlock(name, 0); } } @@ -884,33 +886,34 @@ public synchronized boolean tryAllocate( return false; } - private PipeMemoryBlock registerMemoryBlock(long sizeInBytes) { - return registerMemoryBlock(sizeInBytes, PipeMemoryBlockType.NORMAL); + private PipeMemoryBlock registerMemoryBlock(final String name, final long sizeInBytes) { + return registerMemoryBlock(name, sizeInBytes, PipeMemoryBlockType.NORMAL); } - private PipeMemoryBlock registerMemoryBlock(long sizeInBytes, PipeMemoryBlockType type) { + private synchronized PipeMemoryBlock registerMemoryBlock( + final String name, final long sizeInBytes, final PipeMemoryBlockType type) { final PipeMemoryBlock returnedMemoryBlock; switch (type) { case TABLET: - returnedMemoryBlock = new PipeTabletMemoryBlock(sizeInBytes); + returnedMemoryBlock = new PipeTabletMemoryBlock(name, sizeInBytes); break; case TS_FILE: - returnedMemoryBlock = new PipeTsFileMemoryBlock(sizeInBytes); + returnedMemoryBlock = new PipeTsFileMemoryBlock(name, sizeInBytes); break; case BATCH: case WAL: returnedMemoryBlock = - new PipeModelFixedMemoryBlock(sizeInBytes, new ThresholdAllocationStrategy()); + new PipeModelFixedMemoryBlock(name, sizeInBytes, new ThresholdAllocationStrategy()); break; default: - returnedMemoryBlock = new PipeMemoryBlock(sizeInBytes); + returnedMemoryBlock = new PipeMemoryBlock(name, sizeInBytes); break; } - // For memory block whose size is 0, we do not need to add it to the allocated blocks now. - // It's good for performance and will not trigger concurrent issues. - // If forceResize is called on it, we will add it to the allocated blocks. - if (sizeInBytes > 0) { + memoryBlocks.add(returnedMemoryBlock); + + // Zero-sized blocks do not participate in memory accounting until they are resized. + if (PIPE_MEMORY_MANAGEMENT_ENABLED && sizeInBytes > 0) { memoryBlock.forceAllocateWithoutLimitation(sizeInBytes); allocatedBlocks.add(returnedMemoryBlock); } @@ -997,17 +1000,20 @@ void removeExpandableBlock(final PipeMemoryBlock block) { } public synchronized void release(PipeMemoryBlock block) { - if (!PIPE_MEMORY_MANAGEMENT_ENABLED || block == null || block.isReleased()) { + if (block == null || block.isReleased()) { return; } + memoryBlocks.remove(block); allocatedBlocks.remove(block); - memoryBlock.release(block.getMemoryUsageInBytes()); - if (block instanceof PipeTabletMemoryBlock) { - usedMemorySizeInBytesOfTablets -= block.getMemoryUsageInBytes(); - } - if (block instanceof PipeTsFileMemoryBlock) { - usedMemorySizeInBytesOfTsFiles -= block.getMemoryUsageInBytes(); + if (PIPE_MEMORY_MANAGEMENT_ENABLED) { + memoryBlock.release(block.getMemoryUsageInBytes()); + if (block instanceof PipeTabletMemoryBlock) { + usedMemorySizeInBytesOfTablets -= block.getMemoryUsageInBytes(); + } + if (block instanceof PipeTsFileMemoryBlock) { + usedMemorySizeInBytesOfTsFiles -= block.getMemoryUsageInBytes(); + } } block.markAsReleased(); @@ -1028,6 +1034,9 @@ public synchronized boolean release(PipeMemoryBlock block, long sizeInBytes) { usedMemorySizeInBytesOfTsFiles -= sizeInBytes; } block.setMemoryUsageInBytes(block.getMemoryUsageInBytes() - sizeInBytes); + if (block.getMemoryUsageInBytes() == 0) { + allocatedBlocks.remove(block); + } notifyNextTsFileParserMemoryReservationInternal(); this.notifyAll(); @@ -1079,6 +1088,39 @@ public long getTotalMemorySizeInBytes() { return memoryBlock.getTotalMemorySizeInBytes(); } + public synchronized List getPipeMemoryBlockInfoList() { + final List memoryBlockInfoList = new ArrayList<>(); + memoryBlocks.forEach( + block -> + memoryBlockInfoList.add( + new PipeMemoryBlockInfo(block.getName(), block.getMemoryUsageInBytes()))); + memoryBlockInfoList.add( + new PipeMemoryBlockInfo(FLOATING_MEMORY_BLOCK_NAME, getUsedFloatingMemorySizeInBytes())); + memoryBlockInfoList.sort( + Comparator.comparing(PipeMemoryBlockInfo::getName) + .thenComparingLong(PipeMemoryBlockInfo::getMemoryUsageInBytes)); + return memoryBlockInfoList; + } + + public static final class PipeMemoryBlockInfo { + + private final String name; + private final long memoryUsageInBytes; + + private PipeMemoryBlockInfo(final String name, final long memoryUsageInBytes) { + this.name = name; + this.memoryUsageInBytes = memoryUsageInBytes; + } + + public String getName() { + return name; + } + + public long getMemoryUsageInBytes() { + return memoryUsageInBytes; + } + } + private static class PipeIdentity { private final String pipeName; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeModelFixedMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeModelFixedMemoryBlock.java index 90b3d0329f153..56872f4839441 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeModelFixedMemoryBlock.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeModelFixedMemoryBlock.java @@ -37,8 +37,10 @@ public class PipeModelFixedMemoryBlock extends PipeFixedMemoryBlock { private volatile long memoryAllocatedInBytes; public PipeModelFixedMemoryBlock( - final long memoryUsageInBytes, final DynamicMemoryAllocationStrategy allocationStrategy) { - super(memoryUsageInBytes); + final String name, + final long memoryUsageInBytes, + final DynamicMemoryAllocationStrategy allocationStrategy) { + super(name, memoryUsageInBytes); this.memoryAllocatedInBytes = 0; this.allocationStrategy = allocationStrategy; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTabletMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTabletMemoryBlock.java index 529a2e1ac5c56..094658fa23455 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTabletMemoryBlock.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTabletMemoryBlock.java @@ -21,7 +21,7 @@ public class PipeTabletMemoryBlock extends PipeFixedMemoryBlock { - public PipeTabletMemoryBlock(long memoryUsageInBytes) { - super(memoryUsageInBytes); + public PipeTabletMemoryBlock(final String name, final long memoryUsageInBytes) { + super(name, memoryUsageInBytes); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTsFileMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTsFileMemoryBlock.java index 268388d080009..33a66e180fe96 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTsFileMemoryBlock.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTsFileMemoryBlock.java @@ -21,7 +21,7 @@ public class PipeTsFileMemoryBlock extends PipeFixedMemoryBlock { - public PipeTsFileMemoryBlock(long memoryUsageInBytes) { - super(memoryUsageInBytes); + public PipeTsFileMemoryBlock(final String name, final long memoryUsageInBytes) { + super(name, memoryUsageInBytes); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFilePublicResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFilePublicResource.java index fe54ee48b547a..181d8d65c304e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFilePublicResource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFilePublicResource.java @@ -117,6 +117,7 @@ synchronized boolean cacheDeviceIsAlignedMapIfAbsent(final File tsFile) throws I allocatedMemoryBlock = PipeDataNodeResourceManager.memory() .forceAllocateIfSufficient( + PipeTsFilePublicResource.class.getSimpleName() + "#sequenceReader", PipeConfig.getInstance().getPipeMemoryAllocateForTsFileSequenceReaderInBytes(), MEMORY_SUFFICIENT_THRESHOLD); if (allocatedMemoryBlock == null) { @@ -145,7 +146,10 @@ synchronized boolean cacheDeviceIsAlignedMapIfAbsent(final File tsFile) throws I // Allocate again for the cached objects. allocatedMemoryBlock = PipeDataNodeResourceManager.memory() - .forceAllocateIfSufficient(memoryRequiredInBytes, MEMORY_SUFFICIENT_THRESHOLD); + .forceAllocateIfSufficient( + PipeTsFilePublicResource.class.getSimpleName() + "#metadata", + memoryRequiredInBytes, + MEMORY_SUFFICIENT_THRESHOLD); if (allocatedMemoryBlock == null) { LOGGER.info( DataNodePipeMessages.PIPETSFILERESOURCE_FAILED_TO_CACHE_OBJECTS_FOR_TSFILE, @@ -177,6 +181,7 @@ synchronized boolean cacheObjectsIfAbsent(final File tsFile) throws IOException allocatedMemoryBlock = PipeDataNodeResourceManager.memory() .forceAllocateIfSufficient( + PipeTsFilePublicResource.class.getSimpleName() + "#sequenceReader", PipeConfig.getInstance().getPipeMemoryAllocateForTsFileSequenceReaderInBytes(), MEMORY_SUFFICIENT_THRESHOLD); if (allocatedMemoryBlock == null) { @@ -214,7 +219,10 @@ synchronized boolean cacheObjectsIfAbsent(final File tsFile) throws IOException // Allocate again for the cached objects. allocatedMemoryBlock = PipeDataNodeResourceManager.memory() - .forceAllocateIfSufficient(memoryRequiredInBytes, MEMORY_SUFFICIENT_THRESHOLD); + .forceAllocateIfSufficient( + PipeTsFilePublicResource.class.getSimpleName() + "#metadata", + memoryRequiredInBytes, + MEMORY_SUFFICIENT_THRESHOLD); if (allocatedMemoryBlock == null) { LOGGER.info( DataNodePipeMessages.PIPETSFILERESOURCE_FAILED_TO_CACHE_OBJECTS_FOR_TSFILE, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeCacheLeaderClientManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeCacheLeaderClientManager.java index f32c8cb72bbd5..cb7a9759600d6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeCacheLeaderClientManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeCacheLeaderClientManager.java @@ -53,7 +53,10 @@ public LeaderCacheManager() { // properties required by pipe memory control framework final PipeMemoryBlock allocatedMemoryBlock = - PipeDataNodeResourceManager.memory().tryAllocate(initMemorySizeInBytes); + PipeDataNodeResourceManager.memory() + .tryAllocate( + IoTDBDataNodeCacheLeaderClientManager.class.getSimpleName(), + initMemorySizeInBytes); device2endpoint = Caffeine.newBuilder() diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeSchemaRegionWritePlanEventBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeSchemaRegionWritePlanEventBatch.java index 39a861c01b824..5263eac0ff391 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeSchemaRegionWritePlanEventBatch.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeSchemaRegionWritePlanEventBatch.java @@ -112,7 +112,10 @@ public PipeSchemaRegionWritePlanEventBatch(final PipeParameters parameters) { parameters.getLongOrDefault( Arrays.asList(CONNECTOR_IOTDB_BATCH_SIZE_KEY, SINK_IOTDB_BATCH_SIZE_KEY), CONNECTOR_IOTDB_PLAIN_BATCH_SIZE_DEFAULT_VALUE); - allocatedMemoryBlock = PipeDataNodeResourceManager.memory().forceAllocate(maxBatchSizeInBytes); + allocatedMemoryBlock = + PipeDataNodeResourceManager.memory() + .forceAllocate( + PipeSchemaRegionWritePlanEventBatch.class.getSimpleName(), maxBatchSizeInBytes); } public synchronized boolean onEvent(final PipeSchemaRegionWritePlanEvent event) { 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 39d3bea1dbe15..30b6d30321033 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 @@ -61,7 +61,9 @@ protected PipeTabletEventBatch( // limit in buffer size this.maxBatchSizeInBytes = requestMaxBatchSizeInBytes; - this.allocatedMemoryBlock = PipeDataNodeResourceManager.memory().forceAllocate(0); + this.allocatedMemoryBlock = + PipeDataNodeResourceManager.memory() + .forceAllocate(PipeTabletEventBatch.class.getSimpleName(), 0); if (recordMetric != null) { this.recordMetric = recordMetric; } else { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java index ff624204e2478..1a60c9cab4011 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java @@ -542,7 +542,8 @@ private void transferFilePieces( final int readFileBufferSize = getReadFileBufferSize(file); try (final PipeTsFileMemoryBlock ignored = PipeDataNodeResourceManager.memory() - .forceAllocateForTsFileWithRetry(readFileBufferSize); + .forceAllocateForTsFileWithRetry( + IoTDBDataRegionAirGapSink.class.getSimpleName(), readFileBufferSize); final RandomAccessFile reader = new RandomAccessFile(file, "r")) { final byte[] readBuffer = new byte[readFileBufferSize]; long position = 0; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java index 2f2741898094d..8d5369ba5770e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java @@ -452,7 +452,8 @@ protected void transferFilePieces( final int readFileBufferSize = getReadFileBufferSize(file); try (final PipeTsFileMemoryBlock ignored = PipeDataNodeResourceManager.memory() - .forceAllocateForTsFileWithRetry(readFileBufferSize); + .forceAllocateForTsFileWithRetry( + IoTConsensusV2SyncSink.class.getSimpleName(), readFileBufferSize); final RandomAccessFile reader = new RandomAccessFile(file, "r")) { final byte[] readBuffer = new byte[readFileBufferSize]; long position = 0; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java index 8ebff37453196..4bf544787c28e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java @@ -140,7 +140,10 @@ public void transfer(final AsyncIoTConsensusV2ServiceClient client) if (readBuffer == null) { memoryBlock = - PipeDataNodeResourceManager.memory().forceAllocateForTsFileWithRetry(readFileBufferSize); + PipeDataNodeResourceManager.memory() + .forceAllocateForTsFileWithRetry( + IoTConsensusV2TsFileInsertionEventHandler.class.getSimpleName(), + readFileBufferSize); readBuffer = new byte[readFileBufferSize]; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java index 8c9e0299f31ff..9fad5d8d14b8b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java @@ -97,7 +97,9 @@ protected IoTConsensusV2TransferBatchReqBuilder( Arrays.asList(CONNECTOR_IOTDB_BATCH_SIZE_KEY, SINK_IOTDB_BATCH_SIZE_KEY), CONNECTOR_IOTDB_PLAIN_BATCH_SIZE_DEFAULT_VALUE); - allocatedMemoryBlock = PipeDataNodeResourceManager.memory().forceAllocate(0); + allocatedMemoryBlock = + PipeDataNodeResourceManager.memory() + .forceAllocate(IoTConsensusV2TransferBatchReqBuilder.class.getSimpleName(), 0); } /** diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java index a74cfcbd2d8b1..f37ac39550806 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java @@ -527,7 +527,8 @@ private void transportSingleFilePieceByPiece(final File file) throws IOException final int readFileBufferSize = getReadFileBufferSize(file); try (final PipeTsFileMemoryBlock ignored = PipeDataNodeResourceManager.memory() - .forceAllocateForTsFileWithRetry(readFileBufferSize); + .forceAllocateForTsFileWithRetry( + IoTDBLegacyPipeSink.class.getSimpleName(), readFileBufferSize); final RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r")) { final byte[] buffer = new byte[readFileBufferSize]; while (true) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java index b79e7d6cb2bcf..255dd743fc700 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java @@ -199,7 +199,9 @@ public void transfer( // Delay creation of resources to avoid OOM or too many open files if (readBuffer == null) { memoryBlock = - PipeDataNodeResourceManager.memory().forceAllocateForTsFileWithRetry(readFileBufferSize); + PipeDataNodeResourceManager.memory() + .forceAllocateForTsFileWithRetry( + PipeTransferTsFileHandler.class.getSimpleName(), readFileBufferSize); readBuffer = new byte[readFileBufferSize]; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java index 16c044b491d4a..ba3c73ab7c801 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java @@ -651,7 +651,8 @@ protected void transferFilePieces( final int readFileBufferSize = getReadFileBufferSize(file); try (final PipeTsFileMemoryBlock ignored = PipeDataNodeResourceManager.memory() - .forceAllocateForTsFileWithRetry(readFileBufferSize); + .forceAllocateForTsFileWithRetry( + IoTDBDataRegionSyncSink.class.getSimpleName(), readFileBufferSize); final RandomAccessFile reader = new RandomAccessFile(file, "r")) { final byte[] readBuffer = new byte[readFileBufferSize]; long position = 0; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java index 3ef29dbc9001b..e98852f5b8d8b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java @@ -65,7 +65,9 @@ public DisruptorQueue( allocatedMemoryBlock = PipeDataNodeResourceManager.memory() .tryAllocate( - ringBufferSize * ringBufferEntrySizeInBytes, currentSize -> currentSize / 2); + DisruptorQueue.class.getSimpleName(), + ringBufferSize * ringBufferEntrySizeInBytes, + currentSize -> currentSize / 2); disruptor = new Disruptor<>( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java index c0e7ca6d91a57..6227ab0df1133 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java @@ -75,6 +75,8 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; 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.PipeMemoryManager.PipeMemoryBlockInfo; import org.apache.iotdb.db.protocol.client.ConfigNodeClient; import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager; import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; @@ -186,6 +188,8 @@ public static IInformationSchemaContentSupplier getSupplier( return new RegionSupplier(dataTypes, userEntity); case InformationSchema.PIPES: return new PipeSupplier(dataTypes, userEntity.getUsername()); + case InformationSchema.PIPE_MEMORY: + return new PipeMemorySupplier(dataTypes, userEntity); case InformationSchema.PIPE_PLUGINS: return new PipePluginSupplier(dataTypes, userEntity); case InformationSchema.TOPICS: @@ -728,6 +732,30 @@ public boolean hasNext() { } } + private static class PipeMemorySupplier extends TsBlockSupplier { + + private final Iterator iterator; + + private PipeMemorySupplier(final List dataTypes, final UserEntity userEntity) { + super(dataTypes); + accessControl.checkUserGlobalSysPrivilege(userEntity); + iterator = PipeDataNodeResourceManager.memory().getPipeMemoryBlockInfoList().iterator(); + } + + @Override + protected void constructLine() { + final PipeMemoryBlockInfo memoryBlockInfo = iterator.next(); + columnBuilders[0].writeBinary(BytesUtils.valueOf(memoryBlockInfo.getName())); + columnBuilders[1].writeLong(memoryBlockInfo.getMemoryUsageInBytes()); + resultBuilder.declarePosition(); + } + + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + } + private static class PipePluginSupplier extends TsBlockSupplier { private final Iterator iterator; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java index 7df20573d9911..4c5259c5f76f5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java @@ -150,6 +150,7 @@ public List getDataNodeLocations(final String tableName) { case InformationSchema.CONFIG_NODES: case InformationSchema.DATA_NODES: case InformationSchema.SERVICES: + case InformationSchema.PIPE_MEMORY: return Collections.singletonList(DataNodeEndPoints.getLocalDataNodeLocation()); default: throw new UnsupportedOperationException(DataNodeQueryMessages.UNKNOWN_TABLE + tableName); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java index f7a99ab3485c2..b602d38a52e17 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java @@ -1356,6 +1356,18 @@ public Node visitShowPipesStatement(RelationalSqlParser.ShowPipesStatementContex return new ShowPipes(pipeName, hasWhereClause); } + @Override + public Node visitShowPipeMemoryStatement( + final RelationalSqlParser.ShowPipeMemoryStatementContext ctx) { + return new ShowStatement( + getLocation(ctx), + InformationSchema.PIPE_MEMORY, + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty()); + } + @Override public Node visitShowCreatePipeStatement(RelationalSqlParser.ShowCreatePipeStatementContext ctx) { return new ShowCreatePipe(((Identifier) visit(ctx.pipeName)).getValue()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileParserMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileParserMemoryManager.java index a7430a3f819fc..ae2d12b94ad76 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileParserMemoryManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileParserMemoryManager.java @@ -39,12 +39,13 @@ public static LoadTsFileParserMemoryManager getInstance() { @Override public TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry( - final long sizeInBytes) { + final String name, final long sizeInBytes) { return new LoadParserMemoryBlock(sizeInBytes); } @Override - public TsFileInsertionEventParserMemoryBlock forceAllocate(final long sizeInBytes) { + public TsFileInsertionEventParserMemoryBlock forceAllocate( + final String name, final long sizeInBytes) { return new LoadParserMemoryBlock(sizeInBytes); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/cache/SubscriptionPollResponseCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/cache/SubscriptionPollResponseCache.java index c4b37b88263d3..67f9a7633aeea 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/cache/SubscriptionPollResponseCache.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/cache/SubscriptionPollResponseCache.java @@ -114,7 +114,9 @@ private SubscriptionPollResponseCache() { // properties required by pipe memory control framework final PipeMemoryBlock allocatedMemoryBlock = - PipeDataNodeResourceManager.memory().tryAllocate(initMemorySizeInBytes); + PipeDataNodeResourceManager.memory() + .tryAllocate( + SubscriptionPollResponseCache.class.getSimpleName(), initMemorySizeInBytes); this.cache = Caffeine.newBuilder() diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java index 1392c0cbb2865..e1226be7433e5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java @@ -278,7 +278,9 @@ private synchronized CachedSubscriptionPollResponse generateNextTabletResponse() final List tablets = ((TabletsPayload) response.getPayload()).getTablets(); if (Objects.nonNull(tablets) && !tablets.isEmpty()) { final PipeTabletMemoryBlock memoryBlock = - PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(currentBufferSize); + PipeDataNodeResourceManager.memory() + .forceAllocateForTabletWithRetry( + SubscriptionEventTabletResponse.class.getSimpleName(), currentBufferSize); response.setMemoryBlock(memoryBlock); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTsFileResponse.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTsFileResponse.java index 9ddeca25c4cca..2e68b2bff6320 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTsFileResponse.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTsFileResponse.java @@ -208,7 +208,9 @@ private CachedSubscriptionPollResponse generateResponseWithPieceOrSealPayload( reader.seek(writingOffset); final PipeTsFileMemoryBlock memoryBlock = - PipeDataNodeResourceManager.memory().forceAllocateForTsFileWithRetry(bufferSize); + PipeDataNodeResourceManager.memory() + .forceAllocateForTsFileWithRetry( + SubscriptionEventTsFileResponse.class.getSimpleName(), bufferSize); final byte[] readBuffer = new byte[(int) bufferSize]; reader.readFully(readBuffer); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java index d63fe1c6e8309..401a7a02d70fe 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java @@ -30,6 +30,7 @@ import org.junit.Before; import org.junit.Test; +import java.util.List; import java.util.concurrent.atomic.AtomicLong; public class PipeMemoryManagerResizeTest { @@ -87,7 +88,7 @@ public void testTabletResizeCannotCrossTabletHardLimit() { null, TOTAL_MEMORY_SIZE_IN_BYTES, MemoryBlockType.DYNAMIC)); - final PipeTabletMemoryBlock tablet = manager.forceAllocateForTabletWithRetry(0); + final PipeTabletMemoryBlock tablet = manager.forceAllocateForTabletWithRetry("tablet", 0); try { Assert.assertThrows( @@ -111,9 +112,10 @@ public void testTabletResizeLeavesMemoryForSinkForwardProgress() { TOTAL_MEMORY_SIZE_IN_BYTES, MemoryBlockType.DYNAMIC)); final PipeTabletMemoryBlock retainedTablet = - manager.forceAllocateForTabletWithRetry(TABLET_MEMORY_SIZE_IN_BYTES); - final PipeTabletMemoryBlock pendingTablet = manager.forceAllocateForTabletWithRetry(0); - final PipeMemoryBlock sinkBatch = manager.forceAllocate(0); + manager.forceAllocateForTabletWithRetry("retainedTablet", TABLET_MEMORY_SIZE_IN_BYTES); + final PipeTabletMemoryBlock pendingTablet = + manager.forceAllocateForTabletWithRetry("pendingTablet", 0); + final PipeMemoryBlock sinkBatch = manager.forceAllocate("sinkBatch", 0); try { Assert.assertThrows( @@ -155,7 +157,8 @@ public void testFloatingAndNonFloatingMemoryShareTheSamePool() { Assert.assertEquals( TOTAL_MEMORY_SIZE_IN_BYTES / 2, manager.getTotalFloatingMemorySizeInBytes()); - final PipeTsFileMemoryBlock nonFloatingMemory = manager.forceAllocateForTsFileWithRetry(1200); + final PipeTsFileMemoryBlock nonFloatingMemory = + manager.forceAllocateForTsFileWithRetry("tsFile", 1200); try { // Non-floating memory can borrow the unused half that was previously reserved for InsertNode // queues. Its usage also reduces the current floating-memory limit symmetrically. @@ -167,9 +170,57 @@ public void testFloatingAndNonFloatingMemoryShareTheSamePool() { Assert.assertEquals(300, manager.getFreeMemorySizeInBytes()); Assert.assertThrows( - PipeRuntimeOutOfMemoryCriticalException.class, () -> manager.forceAllocate(301)); + PipeRuntimeOutOfMemoryCriticalException.class, + () -> manager.forceAllocate("normal", 301)); } finally { manager.release(nonFloatingMemory); } } + + @Test + public void testMemoryBlockInfoIncludesNamesAndSeparatesFloatingMemory() { + final AtomicLong floatingMemoryUsageInBytes = new AtomicLong(0); + final PipeMemoryManager manager = + new PipeMemoryManager( + new AtomicLongMemoryBlock( + "PipeMemoryManagerResizeTest", + null, + TOTAL_MEMORY_SIZE_IN_BYTES, + MemoryBlockType.DYNAMIC), + floatingMemoryUsageInBytes::get); + final PipeMemoryBlock normalMemory = manager.forceAllocate("normal", 100); + final PipeMemoryBlock zeroSizedMemory = manager.forceAllocate("zero", 0); + + try { + floatingMemoryUsageInBytes.set(250); + final List memoryBlockInfoList = + manager.getPipeMemoryBlockInfoList(); + + Assert.assertEquals(3, memoryBlockInfoList.size()); + Assert.assertEquals("FloatingMemory", memoryBlockInfoList.get(0).getName()); + Assert.assertEquals(250, memoryBlockInfoList.get(0).getMemoryUsageInBytes()); + Assert.assertEquals("normal", memoryBlockInfoList.get(1).getName()); + Assert.assertEquals(100, memoryBlockInfoList.get(1).getMemoryUsageInBytes()); + Assert.assertEquals("zero", memoryBlockInfoList.get(2).getName()); + Assert.assertEquals(0, memoryBlockInfoList.get(2).getMemoryUsageInBytes()); + Assert.assertEquals(100, manager.getUsedMemorySizeInBytes()); + + manager.forceResize(normalMemory, 0); + final List memoryBlockInfoListAfterResize = + manager.getPipeMemoryBlockInfoList(); + Assert.assertEquals(3, memoryBlockInfoListAfterResize.size()); + Assert.assertEquals("normal", memoryBlockInfoListAfterResize.get(1).getName()); + Assert.assertEquals(0, memoryBlockInfoListAfterResize.get(1).getMemoryUsageInBytes()); + Assert.assertEquals(0, manager.getUsedMemorySizeInBytes()); + } finally { + manager.release(normalMemory); + manager.release(zeroSizedMemory); + } + + final List memoryBlockInfoListAfterRelease = + manager.getPipeMemoryBlockInfoList(); + Assert.assertEquals(1, memoryBlockInfoListAfterRelease.size()); + Assert.assertEquals("FloatingMemory", memoryBlockInfoListAfterRelease.get(0).getName()); + Assert.assertEquals(250, memoryBlockInfoListAfterRelease.get(0).getMemoryUsageInBytes()); + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/ShowPipeMemoryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/ShowPipeMemoryTest.java new file mode 100644 index 0000000000000..d0722c940a2dc --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/ShowPipeMemoryTest.java @@ -0,0 +1,39 @@ +/* + * 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.queryengine.plan.relational.sql.parser; + +import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowStatement; + +import org.junit.Assert; +import org.junit.Test; + +import java.time.ZoneId; + +public class ShowPipeMemoryTest { + + @Test + public void testShowPipeMemoryStatement() { + final Statement statement = + new SqlParser().createStatement("SHOW PIPE MEMORY", ZoneId.systemDefault(), null); + + Assert.assertTrue(statement instanceof ShowStatement); + Assert.assertEquals("pipe_memory", ((ShowStatement) statement).getTableName()); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManagerTest.java index f0ffe1d5f9985..967552b75448d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManagerTest.java @@ -89,7 +89,7 @@ public void testParserMemoryBlockGrowsAndReleasesFromQueryPool() throws Exceptio final LoadTsFileMemoryManager manager = LoadTsFileMemoryManager.getInstance(); final long usedMemoryBefore = manager.getUsedMemorySizeInBytes(); final TsFileInsertionEventParserMemoryBlock block = - LoadTsFileParserMemoryManager.getInstance().forceAllocate(0); + LoadTsFileParserMemoryManager.getInstance().forceAllocate("test", 0); Assert.assertEquals(0L, block.getMemoryUsageInBytes()); block.forceResize(1024); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java index d66b5a8a301eb..9013bc5fad539 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java @@ -295,6 +295,8 @@ private ColumnHeaderConstant() { "estimated_remaining_seconds"; public static final String IS_DEGRADED_TABLE_MODEL = "is_degraded"; public static final String RECENT_FAILURES_TABLE_MODEL = "recent_failures"; + public static final String NAME_TABLE_MODEL = "name"; + public static final String MEMORY_USAGE_IN_BYTES_TABLE_MODEL = "memory_usage_in_bytes"; public static final String PLUGIN_NAME_TABLE_MODEL = "plugin_name"; public static final String PLUGIN_TYPE_TABLE_MODEL = "plugin_type"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java index 048ce8f8763bf..021cfccd7911c 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java @@ -47,6 +47,7 @@ public class InformationSchema { public static final String COLUMNS = "columns"; public static final String REGIONS = "regions"; public static final String PIPES = "pipes"; + public static final String PIPE_MEMORY = "pipe_memory"; public static final String PIPE_PLUGINS = "pipe_plugins"; public static final String TOPICS = "topics"; public static final String SUBSCRIPTIONS = "subscriptions"; @@ -241,6 +242,14 @@ public class InformationSchema { ColumnHeaderConstant.RECENT_FAILURES_TABLE_MODEL, TSDataType.STRING)); schemaTables.put(PIPES, pipeTable); + final TsTable pipeMemoryTable = new TsTable(PIPE_MEMORY); + pipeMemoryTable.addColumnSchema( + new TagColumnSchema(ColumnHeaderConstant.NAME_TABLE_MODEL, TSDataType.STRING)); + pipeMemoryTable.addColumnSchema( + new AttributeColumnSchema( + ColumnHeaderConstant.MEMORY_USAGE_IN_BYTES_TABLE_MODEL, TSDataType.INT64)); + schemaTables.put(PIPE_MEMORY, pipeMemoryTable); + final TsTable pipePluginTable = new TsTable(PIPE_PLUGINS); pipePluginTable.addColumnSchema( new TagColumnSchema(ColumnHeaderConstant.PLUGIN_NAME_TABLE_MODEL, TSDataType.STRING)); diff --git a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 index 037778597d498..3b2452eb100df 100644 --- a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 +++ b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 @@ -102,6 +102,7 @@ statement | dropPipeStatement | startPipeStatement | stopPipeStatement + | showPipeMemoryStatement | showPipesStatement | showCreatePipeStatement | createPipePluginStatement @@ -527,6 +528,10 @@ showPipesStatement : SHOW ((PIPE pipeName=identifier) | PIPES (WHERE (CONNECTOR | SINK) USED BY pipeName=identifier)?) ; +showPipeMemoryStatement + : SHOW PIPE MEMORY + ; + showCreatePipeStatement : SHOW CREATE PIPE pipeName=identifier ; @@ -1521,7 +1526,7 @@ nonReserved | JSON | KEEP | KEY | KEYS | KILL | LANGUAGE | LAST | LATERAL | LEADING | LEAVE | LEVEL | LIMIT | LINEAR | LOAD | LOCAL | LOGICAL | LOOP - | MANAGE_ROLE | MANAGE_USER | MAP | MATCH | MATCHED | MATCHES | MATCH_RECOGNIZE | MATERIALIZED | MEASURES | MEMORY_THRESHOLD | METHOD | MERGE | MICROSECOND | MIGRATE | MILLISECOND | MINUTE | MODEL | MODELS | MODIFY | MONTH + | MANAGE_ROLE | MANAGE_USER | MAP | MATCH | MATCHED | MATCHES | MATCH_RECOGNIZE | MATERIALIZED | MEASURES | MEMORY | MEMORY_THRESHOLD | METHOD | MERGE | MICROSECOND | MIGRATE | MILLISECOND | MINUTE | MODEL | MODELS | MODIFY | MONTH | NANOSECOND | NESTED | NEXT | NFC | NFD | NFKC | NFKD | NO | NODEID | NONE | NULLIF | NULLS | OBJECT | OF | OFFSET | OMIT | ONE | ONLY | OPTION | ORDINALITY | OUTPUT | OVER | OVERFLOW | PARTITION | PARTITIONS | PASSING | PAST | PATH | PATTERN | PER | PERIOD | PERMUTE | PIPE | PIPEPLUGIN | PIPEPLUGINS | PIPES | PLAN | POSITION | PRECEDING | PRECISION | PRIVILEGES | PREVIOUS | PROCESSLIST | PROCESSOR | PROPERTIES | PRUNE @@ -1742,6 +1747,7 @@ MATCHES: 'MATCHES'; MATCH_RECOGNIZE: 'MATCH_RECOGNIZE'; MATERIALIZED: 'MATERIALIZED'; MEASURES: 'MEASURES'; +MEMORY: 'MEMORY'; MEMORY_THRESHOLD: 'MEMORY_THRESHOLD'; METHOD: 'METHOD'; MERGE: 'MERGE';