This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 d74b2744ed1 Fix schema region snapshot memory accounting (#18750)
d74b2744ed1 is described below
commit d74b2744ed13d08c619a99f4038a30b465adb25e
Author: Caideyipi <[email protected]>
AuthorDate: Tue Sep 29 14:50:12 2026 +0800
Fix schema region snapshot memory accounting (#18750)
---
.../schemaregion/impl/SchemaRegionMemoryImpl.java | 10 ++++--
.../schemaregion/impl/SchemaRegionPBTreeImpl.java | 32 ++++++++++++------
.../schemaRegion/SchemaRegionManagementTest.java | 38 +++++++++++++++++-----
3 files changed, 59 insertions(+), 21 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
index 7c6fd4d7e53..14f147d69f2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
@@ -205,6 +205,7 @@ public class SchemaRegionMemoryImpl implements
ISchemaRegion {
private boolean isRecovering = true;
private volatile boolean initialized = false;
+ private boolean ratisLogAppenderBufferReserved = false;
private final String databaseDirPath;
private final String schemaRegionDirPath;
@@ -260,7 +261,10 @@ public class SchemaRegionMemoryImpl implements
ISchemaRegion {
return;
}
- if
(config.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS))
{
+ if (!ratisLogAppenderBufferReserved
+ && config
+ .getSchemaRegionConsensusProtocolClass()
+ .equals(ConsensusFactory.RATIS_CONSENSUS)) {
final long memCost =
config.getSchemaRatisConsensusLogAppenderBufferSizeMax();
if (!SystemInfo.getInstance().addDirectBufferMemoryCost(memCost)) {
throw new MetadataException(
@@ -269,6 +273,7 @@ public class SchemaRegionMemoryImpl implements
ISchemaRegion {
+ DataNodeSchemaMessages.DIRECT_BUFFER_MEMORY_LIMIT
+
SystemInfo.getInstance().getTotalDirectBufferMemorySizeLimit());
}
+ ratisLogAppenderBufferReserved = true;
}
initDir();
@@ -497,9 +502,10 @@ public class SchemaRegionMemoryImpl implements
ISchemaRegion {
// delete all the schema region files
SchemaRegionUtils.deleteSchemaRegionFolder(
schemaRegionDirPath, logger,
RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
- if
(config.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS))
{
+ if (ratisLogAppenderBufferReserved) {
SystemInfo.getInstance()
.decreaseDirectBufferMemoryCost(config.getSchemaRatisConsensusLogAppenderBufferSizeMax());
+ ratisLogAppenderBufferReserved = false;
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
index 502afed74ea..fa4b815e224 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
@@ -166,6 +166,7 @@ public class SchemaRegionPBTreeImpl implements
ISchemaRegion {
private boolean isRecovering = true;
private volatile boolean initialized = false;
private boolean isClearing = false;
+ private boolean ratisLogAppenderBufferReserved = false;
private final String storageGroupDirPath;
private final String schemaRegionDirPath;
@@ -208,7 +209,10 @@ public class SchemaRegionPBTreeImpl implements
ISchemaRegion {
return;
}
- if
(config.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS))
{
+ if (!ratisLogAppenderBufferReserved
+ && config
+ .getSchemaRegionConsensusProtocolClass()
+ .equals(ConsensusFactory.RATIS_CONSENSUS)) {
long memCost = config.getSchemaRatisConsensusLogAppenderBufferSizeMax();
if (!SystemInfo.getInstance().addDirectBufferMemoryCost(memCost)) {
throw new MetadataException(
@@ -217,6 +221,7 @@ public class SchemaRegionPBTreeImpl implements
ISchemaRegion {
+ DataNodeSchemaMessages.DIRECT_BUFFER_MEMORY_LIMIT
+
SystemInfo.getInstance().getTotalDirectBufferMemorySizeLimit());
}
+ ratisLogAppenderBufferReserved = true;
}
initDir();
@@ -507,9 +512,10 @@ public class SchemaRegionPBTreeImpl implements
ISchemaRegion {
SchemaRegionUtils.deleteSchemaRegionFolder(
schemaRegionDirPath, logger,
RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
- if
(config.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS))
{
+ if (ratisLogAppenderBufferReserved) {
SystemInfo.getInstance()
.decreaseDirectBufferMemoryCost(config.getSchemaRatisConsensusLogAppenderBufferSizeMax());
+ ratisLogAppenderBufferReserved = false;
}
}
@@ -520,28 +526,34 @@ public class SchemaRegionPBTreeImpl implements
ISchemaRegion {
return false;
}
logger.info(DataNodeSchemaMessages.START_CREATE_SNAPSHOT, schemaRegionId);
- boolean isSuccess = true;
+ boolean isSuccess;
+ boolean currentResult;
long startTime = System.currentTimeMillis();
long mtreeSnapshotStartTime = System.currentTimeMillis();
- isSuccess = isSuccess && mtree.createSnapshot(snapshotDir);
+ isSuccess = mtree.createSnapshot(snapshotDir);
logger.info(
- DataNodeSchemaMessages.MTREE_SNAPSHOT_CREATION_COST,
+ DataNodeSchemaMessages.MTREE_SNAPSHOT_CREATION_COST_WITH_STATUS,
schemaRegionId,
- System.currentTimeMillis() - mtreeSnapshotStartTime);
+ System.currentTimeMillis() - mtreeSnapshotStartTime,
+ isSuccess);
long tagSnapshotStartTime = System.currentTimeMillis();
- isSuccess = isSuccess && tagManager.createSnapshot(snapshotDir);
+ currentResult = tagManager.createSnapshot(snapshotDir);
+ isSuccess = isSuccess && currentResult;
logger.info(
- DataNodeSchemaMessages.TAG_SNAPSHOT_CREATION_COST,
+ DataNodeSchemaMessages.TAG_SNAPSHOT_CREATION_COST_WITH_STATUS,
schemaRegionId,
- System.currentTimeMillis() - tagSnapshotStartTime);
+ System.currentTimeMillis() - tagSnapshotStartTime,
+ currentResult);
logger.info(
DataNodeSchemaMessages.SNAPSHOT_CREATION_COST,
schemaRegionId,
System.currentTimeMillis() - startTime);
- logger.info(DataNodeSchemaMessages.SUCCESSFULLY_CREATE_SNAPSHOT,
schemaRegionId);
+ if (isSuccess) {
+ logger.info(DataNodeSchemaMessages.SUCCESSFULLY_CREATE_SNAPSHOT,
schemaRegionId);
+ }
return isSuccess;
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionManagementTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionManagementTest.java
index 1bbb17bca4a..7b0b416ea0d 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionManagementTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionManagementTest.java
@@ -28,11 +28,13 @@ import org.apache.iotdb.commons.schema.template.Template;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.schemaengine.SchemaEngine;
import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion;
import org.apache.iotdb.db.schemaengine.schemaregion.SchemaRegionPlanType;
import
org.apache.iotdb.db.schemaengine.schemaregion.read.resp.info.ISchemaInfo;
import
org.apache.iotdb.db.schemaengine.schemaregion.read.resp.info.ITimeSeriesSchemaInfo;
import
org.apache.iotdb.db.schemaengine.schemaregion.write.req.SchemaRegionWritePlanFactory;
+import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
@@ -236,16 +238,34 @@ public class SchemaRegionManagementTest extends
AbstractSchemaRegionTest {
String schemaRegionConsensusProtocolClass =
config.getSchemaRegionConsensusProtocolClass();
config.setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS);
try {
+ long directBufferMemoryCostBeforeRegion =
+ SystemInfo.getInstance().getDirectBufferMemoryCost();
ISchemaRegion schemaRegion = getSchemaRegion("root.sg", 0);
-
- // Loading from a directory that does not contain a snapshot must report
failure rather than
- // silently falling back to an empty region and reporting success.
Callers (the AddPeer flow
- // and the Ratis snapshot-install path) rely on this success/failure
contract.
- File missingSnapshotDir =
- new File(config.getSchemaDir() + File.separator +
"non-existent-snapshot");
- Assert.assertFalse(missingSnapshotDir.exists());
-
- Assert.assertFalse(schemaRegion.loadSnapshot(missingSnapshotDir));
+ long expectedDirectBufferMemoryCost =
+ directBufferMemoryCostBeforeRegion
+ + config.getSchemaRatisConsensusLogAppenderBufferSizeMax();
+ Assert.assertEquals(
+ expectedDirectBufferMemoryCost,
SystemInfo.getInstance().getDirectBufferMemoryCost());
+
+ try {
+ // Loading from a directory that does not contain a snapshot must
report failure rather than
+ // silently falling back to an empty region and reporting success.
Callers (the AddPeer flow
+ // and the Ratis snapshot-install path) rely on this success/failure
contract.
+ File missingSnapshotDir =
+ new File(config.getSchemaDir() + File.separator +
"non-existent-snapshot");
+ Assert.assertFalse(missingSnapshotDir.exists());
+
+ for (int i = 0; i < 2; i++) {
+ Assert.assertFalse(schemaRegion.loadSnapshot(missingSnapshotDir));
+ Assert.assertEquals(
+ expectedDirectBufferMemoryCost,
SystemInfo.getInstance().getDirectBufferMemoryCost());
+ }
+ } finally {
+
SchemaEngine.getInstance().deleteSchemaRegion(schemaRegion.getSchemaRegionId());
+ Assert.assertEquals(
+ directBufferMemoryCostBeforeRegion,
+ SystemInfo.getInstance().getDirectBufferMemoryCost());
+ }
} finally {
config.setSchemaRegionConsensusProtocolClass(schemaRegionConsensusProtocolClass);
}
