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 312c205c8b5 Fix schema region NPE during DataNode shutdown (#18821)
312c205c8b5 is described below
commit 312c205c8b5dc3701dd51ccd676f1c6f26796dc6
Author: Jiang Tian <[email protected]>
AuthorDate: Mon Oct 12 13:56:16 2026 +0800
Fix schema region NPE during DataNode shutdown (#18821)
* Fix schema region NPE during DataNode shutdown
* Handle missing schema regions in schema writes
* Handle cleared MTree roots during lookup
* Return retryable status while SchemaEngine clears regions
* Return retryable status for alter during schema teardown
---
.../iotdb/db/i18n/DataNodeQueryMessages.java | 2 +
.../iotdb/db/i18n/DataNodeQueryMessages.java | 2 +
.../execution/executor/RegionWriteExecutor.java | 36 ++-
.../apache/iotdb/db/schemaengine/SchemaEngine.java | 9 +
.../mtree/impl/mem/MTreeBelowSGMemoryImpl.java | 6 +
.../mtree/impl/pbtree/MTreeBelowSGCachedImpl.java | 7 +-
.../executor/RegionWriteExecutorTest.java | 268 +++++++++++++++++++++
.../mtree/impl/mem/MTreeBelowSGMemoryImplTest.java | 14 ++
.../impl/pbtree/MTreeBelowSGCachedImplTest.java | 18 +-
9 files changed, 358 insertions(+), 4 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index 1672c9024f5..73b27b06223 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.db.i18n;
public final class DataNodeQueryMessages {
+ public static final String
MESSAGE_SCHEMA_REGION_IS_UNAVAILABLE_PLEASE_RETRY_LATER_D642510A =
+ "Schema region is unavailable. Please retry later.";
public static final String
EXCEPTION_FRAGMENT_INSTANCE_ARG_IS_ALREADY_ARG_B44984B4 =
"Fragment instance %s is already %s";
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index 114fc1240ff..ed1fefe479a 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.db.i18n;
public final class DataNodeQueryMessages {
+ public static final String
MESSAGE_SCHEMA_REGION_IS_UNAVAILABLE_PLEASE_RETRY_LATER_D642510A =
+ "SchemaRegion 不可用，请稍后重试。";
public static final String
EXCEPTION_FRAGMENT_INSTANCE_ARG_IS_ALREADY_ARG_B44984B4 =
"Fragment instance %s 已处于 %s 状态";
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutor.java
index 6fc521a7f72..05328211947 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutor.java
@@ -176,6 +176,12 @@ public class RegionWriteExecutor {
return RegionExecutionResult.create(
false, errorMsg,
RpcUtils.getStatus(TSStatusCode.NO_AVAILABLE_REGION_GROUP));
}
+ // SchemaEngine clears regions before DataNodeRegionManager removes
their validation locks.
+ // Failures from an in-flight schema write during that window must
remain retryable.
+ if (groupId instanceof SchemaRegionId && schemaEngine.isClearing()) {
+ return RegionExecutionResult.create(
+ false, e.getMessage(),
RpcUtils.getStatus(TSStatusCode.NO_AVAILABLE_REGION_GROUP));
+ }
LOGGER.warn(e.getMessage(), e);
return RegionExecutionResult.create(
false,
@@ -842,9 +848,9 @@ public class RegionWriteExecutor {
}
/**
- * Check the quota before creating time series.
+ * Check region availability, data types and quota before creating time
series.
*
- * @return null if the quota is not exceeded, otherwise return the
execution result.
+ * @return null if all checks pass, otherwise return the execution result.
*/
private RegionExecutionResult checkQuotaAndTypeBeforeCreatingTimeSeries(
final ISchemaRegion schemaRegion,
@@ -852,6 +858,11 @@ public class RegionWriteExecutor {
final int size,
final List<String> measurements,
final List<TSDataType> dataTypes) {
+ // Shutdown may clear SchemaEngine while the region's validation lock
still exists.
+ // Reject the request as retryable before dereferencing the missing
region.
+ if (schemaRegion == null) {
+ return createSchemaRegionUnavailableResult();
+ }
for (int i = 0; i < measurements.size(); ++i) {
if (dataTypes.get(i) == TSDataType.OBJECT) {
final String errorStr =
@@ -933,6 +944,9 @@ public class RegionWriteExecutor {
AlterTimeSeriesNode node, WritePlanNodeExecutionContext context,
boolean receivedFromPipe) {
ISchemaRegion schemaRegion =
schemaEngine.getSchemaRegion((SchemaRegionId) context.getRegionId());
+ if (schemaRegion == null) {
+ return createSchemaRegionUnavailableResult();
+ }
try {
MeasurementPath measurementPath =
schemaRegion.fetchMeasurementPath(node.getPath());
if (node.isAlterView() &&
!measurementPath.getMeasurementSchema().isLogicalView()) {
@@ -957,6 +971,9 @@ public class RegionWriteExecutor {
new PipeEnrichedWritePlanNode(node), context)
: PlanVisitor.super.visitAlterTimeSeries(node, context);
} catch (MetadataException e) {
+ if (schemaEngine.isClearing()) {
+ return createSchemaRegionUnavailableResult();
+ }
return RegionExecutionResult.create(
true, e.getMessage(), RpcUtils.getStatus(e.getErrorCode(),
e.getMessage()));
}
@@ -1123,6 +1140,9 @@ public class RegionWriteExecutor {
final ISchemaRegion schemaRegion =
schemaEngine.getSchemaRegion((SchemaRegionId) context.getRegionId());
if
(CONFIG.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS))
{
+ if (schemaRegion == null) {
+ return createSchemaRegionUnavailableResult();
+ }
context.getRegionWriteValidationRWLock().writeLock().lock();
try {
// step 1. make sure all target paths do NOT exist.
@@ -1180,6 +1200,9 @@ public class RegionWriteExecutor {
final boolean receivedFromPipe) {
final ISchemaRegion schemaRegion =
schemaEngine.getSchemaRegion((SchemaRegionId) context.getRegionId());
+ if (schemaRegion == null) {
+ return createSchemaRegionUnavailableResult();
+ }
try {
schemaRegion.checkSchemaQuota(node.getTableName(),
node.getDeviceIdList());
} catch (final SchemaQuotaExceededException e) {
@@ -1192,6 +1215,15 @@ public class RegionWriteExecutor {
: PlanVisitor.super.visitCreateOrUpdateTableDevice(node, context);
}
+ private RegionExecutionResult createSchemaRegionUnavailableResult() {
+ // A validation lock can outlive its region during shutdown, so all
local schema checks
+ // must report a missing region as retryable even when the lock is still
registered.
+ final String message =
+
DataNodeQueryMessages.MESSAGE_SCHEMA_REGION_IS_UNAVAILABLE_PLEASE_RETRY_LATER_D642510A;
+ return RegionExecutionResult.create(
+ false, message,
RpcUtils.getStatus(TSStatusCode.NO_AVAILABLE_REGION_GROUP, message));
+ }
+
@Override
public RegionExecutionResult visitPipeEnrichedWritePlanNode(
final PipeEnrichedWritePlanNode node, final
WritePlanNodeExecutionContext context) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java
index 58a8c96a9dd..ce8ba551a40 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java
@@ -91,6 +91,9 @@ public class SchemaEngine {
private SchemaMetricManager schemaMetricManager;
+ // Published so write handlers can identify teardown failures before region
locks are removed.
+ private volatile boolean clearing;
+
private final DataNodeSchemaQuotaManager schemaQuotaManager =
DataNodeSchemaQuotaManager.getInstance();
@@ -110,6 +113,7 @@ public class SchemaEngine {
}
public void init() {
+ clearing = false;
logger.info(
DataNodeSchemaMessages.USED_SCHEMA_ENGINE_MODE,
CommonDescriptor.getInstance().getConfig().getSchemaEngineMode());
@@ -244,6 +248,7 @@ public class SchemaEngine {
}
public synchronized void clear() {
+ clearing = true;
schemaRegionLoader.clear();
// clearSchemaResource will shut down release and flush task in PBTree
mode, which must be
@@ -268,6 +273,10 @@ public class SchemaEngine {
ClusterTemplateManager.getInstance().clear();
}
+ public boolean isClearing() {
+ return clearing;
+ }
+
public ISchemaRegion getSchemaRegion(final SchemaRegionId regionId) {
return schemaRegionMap.get(regionId);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java
index c29f4ec7d39..f920c6d0efe 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java
@@ -1041,6 +1041,9 @@ public class MTreeBelowSGMemoryImpl {
public IMemMNode getNodeByPath(final PartialPath path) throws
PathNotExistException {
final String[] nodes = path.getNodes();
IMemMNode cur = databaseMNode;
+ if (cur == null) {
+ throw new PathNotExistException(path.getFullPath(), true);
+ }
IMemMNode next;
for (int i = levelOfDB + 1; i < nodes.length; i++) {
next = cur.getChild(nodes[i]);
@@ -1924,6 +1927,9 @@ public class MTreeBelowSGMemoryImpl {
// region table device management
public int getTableDeviceNotExistNum(final String tableName, final
List<Object[]> deviceIdList) {
+ if (databaseMNode == null) {
+ return deviceIdList.size();
+ }
final IMemMNode tableNode = databaseMNode.getChild(tableName);
int notExistNum = deviceIdList.size();
if (tableNode == null) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImpl.java
index 188620ef6a5..e6ee1e36ffa 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImpl.java
@@ -1074,6 +1074,9 @@ public class MTreeBelowSGCachedImpl {
ICachedMNode cur = databaseMNode;
ICachedMNode next;
try {
+ if (cur == null) {
+ throw new PathNotExistException(path.getFullPath(), true);
+ }
for (int i = levelOfDB + 1; i < nodes.length; i++) {
next = store.getChild(cur, nodes[i]);
if (next == null) {
@@ -1090,7 +1093,9 @@ public class MTreeBelowSGCachedImpl {
pinMNode(cur);
return cur;
} finally {
- unPinPath(cur);
+ if (cur != null) {
+ unPinPath(cur);
+ }
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java
index ac3215d7027..22521458c17 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java
@@ -22,14 +22,30 @@ package org.apache.iotdb.db.queryengine.execution.executor;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.commons.consensus.SchemaRegionId;
import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import
org.apache.iotdb.commons.schema.view.viewExpression.leaf.TimeSeriesViewOperand;
+import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.IConsensus;
import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.exception.metadata.PathNotExistException;
import org.apache.iotdb.db.protocol.thrift.impl.DataNodeRegionManager;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.AlterTimeSeriesNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.InternalCreateMultiTimeSeriesNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.InternalCreateTimeSeriesNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.MeasurementGroup;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.view.CreateLogicalViewNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedWritePlanNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
+import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
+import
org.apache.iotdb.db.queryengine.plan.statement.metadata.AlterTimeSeriesStatement.AlterType;
import org.apache.iotdb.db.schemaengine.SchemaEngine;
+import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion;
import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager;
import org.apache.iotdb.db.trigger.executor.TriggerFireResult;
import org.apache.iotdb.db.trigger.executor.TriggerFireVisitor;
@@ -37,17 +53,269 @@ import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.trigger.api.enums.TriggerEvent;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.utils.Pair;
import org.junit.Test;
import org.mockito.Mockito;
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class RegionWriteExecutorTest {
+ @Test
+ public void testAlterTimeSeriesRegionAvailability() throws Exception {
+ // Alter validation fetches the measurement before consensus; a missing
region must be
+ // retryable.
+ checkSchemaWriteWithRegionAvailability(
+ new AlterTimeSeriesNode(
+ new PlanNodeId("alter"),
+ new MeasurementPath("root.sg.d.s"),
+ AlterType.ADD_TAGS,
+ Collections.singletonMap("tag", "value"),
+ null,
+ null,
+ null,
+ false));
+ }
+
+ @Test
+ public void testCreateLogicalViewRegionAvailability() throws Exception {
+ // Ratis checks target existence locally. SimpleConsensus must retain its
direct write path.
+ checkSchemaWriteWithRegionAvailability(
+ new CreateLogicalViewNode(
+ new PlanNodeId("view"),
+ Collections.singletonMap(
+ new MeasurementPath("root.sg.d.v"), new
TimeSeriesViewOperand("root.sg.d.s"))));
+ }
+
+ @Test
+ public void testCreateOrUpdateTableDeviceRegionAvailability() throws
Exception {
+ // Table-device quota validation has the same missing-region window as
tree auto-creation.
+ checkSchemaWriteWithRegionAvailability(
+ new CreateOrUpdateTableDeviceNode(
+ new PlanNodeId("table"),
+ "db",
+ "table1",
+ Collections.singletonList(new Object[] {"device1"}),
+ Collections.emptyList(),
+ Collections.singletonList(new Object[0])));
+ }
+
+ private void checkSchemaWriteWithRegionAvailability(final WritePlanNode
node) throws Exception {
+ final String originalProtocol =
+
IoTDBDescriptor.getInstance().getConfig().getSchemaRegionConsensusProtocolClass();
+ try {
+ for (String protocol :
+ new String[] {ConsensusFactory.RATIS_CONSENSUS,
ConsensusFactory.SIMPLE_CONSENSUS}) {
+
IoTDBDescriptor.getInstance().getConfig().setSchemaRegionConsensusProtocolClass(protocol);
+ for (boolean available : new boolean[] {false, true}) {
+ for (boolean fromPipe : new boolean[] {false, true}) {
+ final IConsensus consensus = Mockito.mock(IConsensus.class);
+ final DataNodeRegionManager regionManager =
Mockito.mock(DataNodeRegionManager.class);
+ final SchemaEngine schemaEngine = Mockito.mock(SchemaEngine.class);
+ final SchemaRegionId regionId = new SchemaRegionId(1);
+ final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+ // Retain the lock even when the region has already been cleared
during shutdown.
+
Mockito.when(regionManager.getRegionLock(regionId)).thenReturn(lock);
+ final ISchemaRegion region = Mockito.mock(ISchemaRegion.class);
+ Mockito.when(schemaEngine.getSchemaRegion(regionId))
+ .thenReturn(available ? region : null);
+ if (node instanceof AlterTimeSeriesNode alterNode) {
+ Mockito.when(region.fetchMeasurementPath(alterNode.getPath()))
+ .thenReturn(new MeasurementPath("root.sg.d.s",
TSDataType.INT64));
+ }
+ final RegionWriteExecutor executor =
+ new RegionWriteExecutor(
+ Mockito.mock(IConsensus.class),
+ consensus,
+ regionManager,
+ schemaEngine,
+ Mockito.mock(ClusterTemplateManager.class),
+ Mockito.mock(TriggerFireVisitor.class));
+ final WritePlanNode request = fromPipe ? new
PipeEnrichedWritePlanNode(node) : node;
+ Mockito.when(consensus.write(regionId, request))
+ .thenReturn(new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+
+ final boolean validatesRegion =
+ !(node instanceof CreateLogicalViewNode)
+ || ConsensusFactory.RATIS_CONSENSUS.equals(protocol);
+ final boolean accepted = available || !validatesRegion;
+ final RegionExecutionResult result = executor.execute(regionId,
request);
+ assertEquals(accepted, result.isAccepted());
+ assertEquals(
+ (accepted ? TSStatusCode.SUCCESS_STATUS :
TSStatusCode.NO_AVAILABLE_REGION_GROUP)
+ .getStatusCode(),
+ result.getStatus().getCode());
+ assertFalse(lock.isWriteLocked());
+ assertEquals(0, lock.getReadLockCount());
+ if (accepted) {
+ Mockito.verify(consensus).write(regionId, request);
+ if (node instanceof AlterTimeSeriesNode alterNode) {
+
Mockito.verify(region).fetchMeasurementPath(alterNode.getPath());
+ } else if (node instanceof CreateOrUpdateTableDeviceNode
tableNode) {
+ Mockito.verify(region)
+ .checkSchemaQuota(tableNode.getTableName(),
tableNode.getDeviceIdList());
+ } else if (validatesRegion) {
+ Mockito.verify(region)
+ .checkMeasurementExistence(
+ new PartialPath("root.sg.d"),
Collections.singletonList("v"), null);
+ }
+ } else {
+ Mockito.verifyZeroInteractions(consensus);
+ assertEquals(result.getMessage(),
result.getStatus().getMessage());
+ }
+ }
+ }
+ }
+ } finally {
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setSchemaRegionConsensusProtocolClass(originalProtocol);
+ }
+ }
+
+ @Test
+ public void testInternalCreateTimeSeriesAfterSchemaRegionCleared() throws
Exception {
+ // A shutdown can clear the region while its validation lock remains
registered.
+ checkInternalCreateWithRegionAvailability(false, false);
+ }
+
+ @Test
+ public void testInternalCreateMultiTimeSeriesAfterSchemaRegionCleared()
throws Exception {
+ // Batch and Pipe auto-creation must return a retryable status for the
same shutdown window.
+ checkInternalCreateWithRegionAvailability(true, false);
+ }
+
+ @Test
+ public void testInternalCreateWithAvailableSchemaRegion() throws Exception {
+ // The guard must preserve quota checks and consensus writes for both
live-region requests.
+ checkInternalCreateWithRegionAvailability(false, true);
+ checkInternalCreateWithRegionAvailability(true, true);
+ }
+
+ @Test
+ public void testSchemaWriteWhenRegionIsClearedAfterLookup() throws Exception
{
+ final IConsensus consensus = Mockito.mock(IConsensus.class);
+ final DataNodeRegionManager regionManager =
Mockito.mock(DataNodeRegionManager.class);
+ final SchemaEngine schemaEngine = Mockito.mock(SchemaEngine.class);
+ final SchemaRegionId regionId = new SchemaRegionId(1);
+ final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+ final ISchemaRegion region = Mockito.mock(ISchemaRegion.class);
+ final AtomicBoolean clearing = new AtomicBoolean(false);
+ Mockito.when(regionManager.getRegionLock(regionId)).thenReturn(lock);
+ Mockito.when(schemaEngine.getSchemaRegion(regionId)).thenReturn(region);
+ Mockito.when(schemaEngine.isClearing()).thenAnswer(invocation ->
clearing.get());
+ Mockito.doAnswer(
+ invocation -> {
+ // The region was obtained before shutdown cleared its MTree.
+ clearing.set(true);
+ throw new PathNotExistException("root.sg.d.s", true);
+ })
+ .when(region)
+ .fetchMeasurementPath(Mockito.any());
+
+ final RegionWriteExecutor executor =
+ new RegionWriteExecutor(
+ Mockito.mock(IConsensus.class),
+ consensus,
+ regionManager,
+ schemaEngine,
+ Mockito.mock(ClusterTemplateManager.class),
+ Mockito.mock(TriggerFireVisitor.class));
+ final AlterTimeSeriesNode request =
+ new AlterTimeSeriesNode(
+ new PlanNodeId("alter"),
+ new MeasurementPath("root.sg.d.s"),
+ AlterType.ADD_TAGS,
+ Collections.singletonMap("tag", "value"),
+ null,
+ null,
+ null,
+ false);
+
+ final RegionExecutionResult result = executor.execute(regionId, request);
+
+ assertFalse(result.isAccepted());
+ assertEquals(
+ TSStatusCode.NO_AVAILABLE_REGION_GROUP.getStatusCode(),
result.getStatus().getCode());
+ assertFalse(lock.isWriteLocked());
+ assertEquals(0, lock.getReadLockCount());
+ Mockito.verifyZeroInteractions(consensus);
+ }
+
+ private void checkInternalCreateWithRegionAvailability(boolean batch,
boolean available)
+ throws Exception {
+ final String originalProtocol =
+
IoTDBDescriptor.getInstance().getConfig().getSchemaRegionConsensusProtocolClass();
+ try {
+ for (String protocol :
+ new String[] {ConsensusFactory.SIMPLE_CONSENSUS,
ConsensusFactory.RATIS_CONSENSUS}) {
+
IoTDBDescriptor.getInstance().getConfig().setSchemaRegionConsensusProtocolClass(protocol);
+ for (boolean fromPipe : new boolean[] {false, true}) {
+ final IConsensus consensus = Mockito.mock(IConsensus.class);
+ final DataNodeRegionManager regionManager =
Mockito.mock(DataNodeRegionManager.class);
+ final SchemaEngine schemaEngine = Mockito.mock(SchemaEngine.class);
+ final SchemaRegionId regionId = new SchemaRegionId(1);
+ final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+ Mockito.when(regionManager.getRegionLock(regionId)).thenReturn(lock);
+ final ISchemaRegion region = Mockito.mock(ISchemaRegion.class);
+ Mockito.when(schemaEngine.getSchemaRegion(regionId))
+ .thenReturn(available ? region : null);
+ final RegionWriteExecutor executor =
+ new RegionWriteExecutor(
+ Mockito.mock(IConsensus.class),
+ consensus,
+ regionManager,
+ schemaEngine,
+ Mockito.mock(ClusterTemplateManager.class),
+ Mockito.mock(TriggerFireVisitor.class));
+ final PartialPath device = new PartialPath("root.sg.d");
+ final MeasurementGroup measurements = new MeasurementGroup();
+ measurements.addMeasurement(
+ "s", TSDataType.INT64, TSEncoding.PLAIN, CompressionType.SNAPPY);
+ final WritePlanNode node =
+ batch
+ ? new InternalCreateMultiTimeSeriesNode(
+ new PlanNodeId("test"),
+ Collections.singletonMap(device, new Pair<>(false,
measurements)))
+ : new InternalCreateTimeSeriesNode(
+ new PlanNodeId("test"), device, measurements, false);
+ final WritePlanNode request = fromPipe ? new
PipeEnrichedWritePlanNode(node) : node;
+ Mockito.when(consensus.write(regionId, request))
+ .thenReturn(new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+
+ final RegionExecutionResult result = executor.execute(regionId,
request);
+ assertEquals(available, result.isAccepted());
+ assertEquals(
+ (available ? TSStatusCode.SUCCESS_STATUS :
TSStatusCode.NO_AVAILABLE_REGION_GROUP)
+ .getStatusCode(),
+ result.getStatus().getCode());
+ assertFalse(lock.isWriteLocked());
+ assertEquals(0, lock.getReadLockCount());
+ if (available) {
+ Mockito.verify(region).checkSchemaQuota(device, 1);
+ Mockito.verify(consensus).write(regionId, request);
+ } else {
+ Mockito.verifyZeroInteractions(consensus);
+ assertEquals(1, measurements.size());
+ }
+ }
+ }
+ } finally {
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setSchemaRegionConsensusProtocolClass(originalProtocol);
+ }
+ }
+
@Test
public void testInsertRowNode() throws ConsensusException {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImplTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImplTest.java
index 1dd3fa456b7..2b811f1e945 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImplTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImplTest.java
@@ -43,6 +43,7 @@ import org.junit.Assert;
import org.junit.Test;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -50,6 +51,19 @@ import java.util.concurrent.atomic.AtomicInteger;
public class MTreeBelowSGMemoryImplTest {
+ @Test
+ public void testPathAndTableDeviceChecksAfterClear() throws Exception {
+ final MTreeBelowSGMemoryImpl mtree = newMTree();
+ mtree.clear();
+
+ Assert.assertFalse(mtree.checkDeviceNodeExists(new
PartialPath("root.sg.d")));
+ assertPathNotExist(mtree, new PartialPath("root.sg.d"));
+ Assert.assertEquals(
+ 2,
+ mtree.getTableDeviceNotExistNum(
+ "table", Arrays.asList(new Object[] {"a"}, new Object[] {"b"})));
+ }
+
@Test
public void
testDeviceDescendantFlagIsMaintainedAcrossCreateDeleteAndRebuild() throws
Exception {
final MTreeBelowSGMemoryImpl mtree = newMTree();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImplTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImplTest.java
index 81936640b49..5c0ebf82d40 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImplTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/MTreeBelowSGCachedImplTest.java
@@ -178,6 +178,17 @@ public class MTreeBelowSGCachedImplTest {
new SchemaRegionCachedMetric(regionStatistics, "root.sg"));
}
+ @Test
+ public void testPathLookupAfterClear() throws Exception {
+ mtree = newMTree();
+ mtree.clear();
+ final MTreeBelowSGCachedImpl clearedMTree = mtree;
+ mtree = null;
+
+ Assert.assertFalse(clearedMTree.checkDeviceNodeExists(new
PartialPath("root.sg.d")));
+ assertPathNotExist(clearedMTree, new PartialPath("root.sg.d"));
+ }
+
private void createBooleanTimeSeries(final String path) throws
MetadataException {
mtree.createTimeSeries(
new MeasurementPath(path),
@@ -207,8 +218,13 @@ public class MTreeBelowSGCachedImplTest {
}
private void assertPathNotExist(final PartialPath path) throws
MetadataException {
+ assertPathNotExist(mtree, path);
+ }
+
+ private void assertPathNotExist(final MTreeBelowSGCachedImpl targetMTree,
final PartialPath path)
+ throws MetadataException {
try {
- mtree.getNodeByPath(path);
+ targetMTree.getNodeByPath(path);
Assert.fail("Expected path not exist: " + path.getFullPath());
} catch (final PathNotExistException ignored) {
// expected
