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 c9e2a53506e9b..6276ba27806d3 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 857031856b756..31d0682f5316a 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 6fc521a7f724c..053282119470f 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 RegionExecutionResult execute(final ConsensusGroupId groupId, final PlanN 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 @@ private RegionExecutionResult executeInternalCreateMultiTimeSeries( } /** - * 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 @@ private RegionExecutionResult checkQuotaAndTypeBeforeCreatingTimeSeries( final int size, final List measurements, final List 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 @@ private RegionExecutionResult executeAlterTimeSeries( 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 @@ private RegionExecutionResult executeAlterTimeSeries( 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 @@ private RegionExecutionResult executeCreateLogicalView( 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 @@ private RegionExecutionResult executeCreateOrUpdateTableDevice( 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 @@ private RegionExecutionResult executeCreateOrUpdateTableDevice( : 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 58a8c96a9dd9d..ce8ba551a4019 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 static SchemaEngine getInstance() { } public void init() { + clearing = false; logger.info( DataNodeSchemaMessages.USED_SCHEMA_ENGINE_MODE, CommonDescriptor.getInstance().getConfig().getSchemaEngineMode()); @@ -244,6 +248,7 @@ public void forceMlog() { } 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 synchronized void clear() { 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 c29f4ec7d3941..f920c6d0efecc 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 @@ protected Void collectEntity(final IDeviceMNode node) { 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 @@ protected Void collectMeasurement(final IMeasurementMNode node) { // region table device management public int getTableDeviceNotExistNum(final String tableName, final List 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 188620ef6a51e..e6ee1e36ffac2 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 ICachedMNode getNodeByPath(PartialPath path) throws MetadataException { 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 ICachedMNode getNodeByPath(PartialPath path) throws MetadataException { 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 ac3215d7027b5..22521458c17b6 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 @@ 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.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 1dd3fa456b791..2b811f1e94572 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.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 @@ 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 81936640b4946..5c0ebf82d405b 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 @@ private MTreeBelowSGCachedImpl newMTree() throws Exception { 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 @@ protected Void collectMeasurement(final IMeasurementMNode node) { } 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