This is an automated email from the ASF dual-hosted git repository.
jackietien 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 9c9a70e14cd Implement DataNodeSchemaLockManager for distributed
concurrency control
9c9a70e14cd is described below
commit 9c9a70e14cd2d5fad9af1fd6af3b4ac907d5be73
Author: Marcos_Zyk <[email protected]>
AuthorDate: Sun Apr 14 14:09:48 2024 +0800
Implement DataNodeSchemaLockManager for distributed concurrency control
---
.../impl/DataNodeInternalRPCServiceImpl.java | 15 ++++--
.../db/queryengine/common/MPPQueryContext.java | 17 +++++--
.../iotdb/db/queryengine/plan/Coordinator.java | 13 +++--
.../queryengine/plan/analyze/AnalyzeVisitor.java | 35 +++++++++----
.../analyze/cache/schema/DataNodeSchemaCache.java | 22 --------
.../analyze/lock/DataNodeSchemaLockManager.java | 59 ++++++++++++++++++++++
.../plan/analyze/lock/SchemaLockType.java | 51 +++++++++++++++++++
.../plan/analyze/schema/ClusterSchemaFetcher.java | 14 ++---
8 files changed, 176 insertions(+), 50 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index ae9ded78d1a..fb3e2412e4a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -84,6 +84,8 @@ import
org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeDevicePathCache;
import
org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache;
+import
org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager;
+import org.apache.iotdb.db.queryengine.plan.analyze.lock.SchemaLockType;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
@@ -548,13 +550,13 @@ public class DataNodeInternalRPCServiceImpl implements
IDataNodeRPCService.Iface
@Override
public TSStatus
invalidateMatchedSchemaCache(TInvalidateMatchedSchemaCacheReq req) {
DataNodeSchemaCache cache = DataNodeSchemaCache.getInstance();
- cache.takeDeleteLock();
+
DataNodeSchemaLockManager.getInstance().takeWriteLock(SchemaLockType.VALIDATE_VS_DELETION);
cache.takeWriteLock();
try {
cache.invalidate(PathPatternTree.deserialize(req.pathPatternTree).getAllPathPatterns());
} finally {
cache.releaseWriteLock();
- cache.releaseDeleteLock();
+
DataNodeSchemaLockManager.getInstance().releaseWriteLock(SchemaLockType.VALIDATE_VS_DELETION);
}
return RpcUtils.SUCCESS_STATUS;
}
@@ -1697,7 +1699,14 @@ public class DataNodeInternalRPCServiceImpl implements
IDataNodeRPCService.Iface
public TSStatus updateTemplate(TUpdateTemplateReq req) {
switch (TemplateInternalRPCUpdateType.getType(req.type)) {
case ADD_TEMPLATE_SET_INFO:
-
ClusterTemplateManager.getInstance().addTemplateSetInfo(req.getTemplateInfo());
+ DataNodeSchemaLockManager.getInstance()
+ .takeWriteLock(SchemaLockType.TIMESERIES_VS_TEMPLATE);
+ try {
+
ClusterTemplateManager.getInstance().addTemplateSetInfo(req.getTemplateInfo());
+ } finally {
+ DataNodeSchemaLockManager.getInstance()
+ .releaseWriteLock(SchemaLockType.TIMESERIES_VS_TEMPLATE);
+ }
break;
case INVALIDATE_TEMPLATE_SET_INFO:
ClusterTemplateManager.getInstance().invalidateTemplateSetInfo(req.getTemplateInfo());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
index 0e50966d864..56b738b4e4c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
@@ -25,12 +25,15 @@ import
org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
import org.apache.iotdb.db.queryengine.plan.analyze.PredicateUtils;
import org.apache.iotdb.db.queryengine.plan.analyze.QueryType;
import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
+import org.apache.iotdb.db.queryengine.plan.analyze.lock.SchemaLockType;
import org.apache.iotdb.db.queryengine.statistics.QueryPlanStatistics;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import java.time.ZoneId;
+import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
/**
* This class is used to record the context of a query including QueryId,
query statement, session
@@ -65,7 +68,7 @@ public class MPPQueryContext {
private Filter globalTimeFilter;
- private int acquiredLockNum;
+ private Map<SchemaLockType, Integer> acquiredLockNumMap = new HashMap<>();
private boolean isExplainAnalyze = false;
@@ -187,12 +190,16 @@ public class MPPQueryContext {
return sql;
}
- public int getAcquiredLockNum() {
- return acquiredLockNum;
+ public Map<SchemaLockType, Integer> getAcquiredLockNumMap() {
+ return acquiredLockNumMap;
}
- public void addAcquiredLockNum() {
- acquiredLockNum++;
+ public void addAcquiredLockNum(SchemaLockType lockType) {
+ if (acquiredLockNumMap.containsKey(lockType)) {
+ acquiredLockNumMap.put(lockType, acquiredLockNumMap.get(lockType) + 1);
+ } else {
+ acquiredLockNumMap.put(lockType, 1);
+ }
}
public void generateGlobalTimeFilter(Analysis analysis) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
index 78c5be03828..3b406bdb48c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
@@ -35,7 +35,8 @@ import org.apache.iotdb.db.queryengine.common.QueryId;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.execution.QueryIdGenerator;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
-import
org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache;
+import
org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager;
+import org.apache.iotdb.db.queryengine.plan.analyze.lock.SchemaLockType;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
import org.apache.iotdb.db.queryengine.plan.execution.IQueryExecution;
@@ -52,6 +53,7 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
@@ -139,9 +141,12 @@ public class Coordinator {
}
return result;
} finally {
- if (queryContext != null && queryContext.getAcquiredLockNum() > 0) {
- for (int i = 0, lockNums = queryContext.getAcquiredLockNum(); i <
lockNums; i++) {
- DataNodeSchemaCache.getInstance().releaseInsertLock();
+ if (queryContext != null &&
!queryContext.getAcquiredLockNumMap().isEmpty()) {
+ Map<SchemaLockType, Integer> lockMap =
queryContext.getAcquiredLockNumMap();
+ for (Map.Entry<SchemaLockType, Integer> entry : lockMap.entrySet()) {
+ for (int i = 0; i < entry.getValue(); i++) {
+
DataNodeSchemaLockManager.getInstance().releaseReadLock(entry.getKey());
+ }
}
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
index f8f7c7ee12d..2638c29c257 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
@@ -56,6 +56,8 @@ import
org.apache.iotdb.db.queryengine.common.schematree.DeviceSchemaInfo;
import org.apache.iotdb.db.queryengine.common.schematree.ISchemaTree;
import org.apache.iotdb.db.queryengine.execution.operator.window.WindowType;
import org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet;
+import
org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager;
+import org.apache.iotdb.db.queryengine.plan.analyze.lock.SchemaLockType;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.SchemaValidator;
import org.apache.iotdb.db.queryengine.plan.expression.Expression;
@@ -2298,7 +2300,7 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
analysis.setStatement(createTimeSeriesStatement);
checkIsTemplateCompatible(
- createTimeSeriesStatement.getPath(),
createTimeSeriesStatement.getAlias());
+ createTimeSeriesStatement.getPath(),
createTimeSeriesStatement.getAlias(), context);
PathPatternTree patternTree = new PathPatternTree();
patternTree.appendFullPath(createTimeSeriesStatement.getPath());
@@ -2309,7 +2311,10 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
return analysis;
}
- private void checkIsTemplateCompatible(PartialPath timeseriesPath, String
alias) {
+ private void checkIsTemplateCompatible(
+ PartialPath timeseriesPath, String alias, MPPQueryContext context) {
+
DataNodeSchemaLockManager.getInstance().takeReadLock(SchemaLockType.TIMESERIES_VS_TEMPLATE);
+ context.addAcquiredLockNum(SchemaLockType.TIMESERIES_VS_TEMPLATE);
Pair<Template, PartialPath> templateInfo =
schemaFetcher.checkTemplateSetAndPreSetInfo(timeseriesPath, alias);
if (templateInfo != null) {
@@ -2320,7 +2325,12 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
}
private void checkIsTemplateCompatible(
- PartialPath devicePath, List<String> measurements, List<String>
aliasList) {
+ PartialPath devicePath,
+ List<String> measurements,
+ List<String> aliasList,
+ MPPQueryContext context) {
+
DataNodeSchemaLockManager.getInstance().takeReadLock(SchemaLockType.TIMESERIES_VS_TEMPLATE);
+ context.addAcquiredLockNum(SchemaLockType.TIMESERIES_VS_TEMPLATE);
for (int i = 0; i < measurements.size(); i++) {
Pair<Template, PartialPath> templateInfo =
schemaFetcher.checkTemplateSetAndPreSetInfo(
@@ -2388,7 +2398,8 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
checkIsTemplateCompatible(
createAlignedTimeSeriesStatement.getDevicePath(),
createAlignedTimeSeriesStatement.getMeasurements(),
- createAlignedTimeSeriesStatement.getAliasList());
+ createAlignedTimeSeriesStatement.getAliasList(),
+ context);
PathPatternTree pathPatternTree = new PathPatternTree();
for (String measurement :
createAlignedTimeSeriesStatement.getMeasurements()) {
@@ -2414,7 +2425,8 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
checkIsTemplateCompatible(
internalCreateTimeSeriesStatement.getDevicePath(),
internalCreateTimeSeriesStatement.getMeasurements(),
- null);
+ null,
+ context);
PathPatternTree pathPatternTree = new PathPatternTree();
for (String measurement :
internalCreateTimeSeriesStatement.getMeasurements()) {
@@ -2442,7 +2454,8 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
PathPatternTree pathPatternTree = new PathPatternTree();
for (Map.Entry<PartialPath, Pair<Boolean, MeasurementGroup>> entry :
internalCreateMultiTimeSeriesStatement.getDeviceMap().entrySet()) {
- checkIsTemplateCompatible(entry.getKey(),
entry.getValue().right.getMeasurements(), null);
+ checkIsTemplateCompatible(
+ entry.getKey(), entry.getValue().right.getMeasurements(), null,
context);
pathPatternTree.appendFullPath(entry.getKey().concatNode(ONE_LEVEL_PATH_WILDCARD));
}
@@ -2467,7 +2480,7 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
List<String> aliasList = createMultiTimeSeriesStatement.getAliasList();
for (int i = 0; i < timeseriesPathList.size(); i++) {
checkIsTemplateCompatible(
- timeseriesPathList.get(i), aliasList == null ? null :
aliasList.get(i));
+ timeseriesPathList.get(i), aliasList == null ? null :
aliasList.get(i), context);
}
PathPatternTree patternTree = new PathPatternTree();
@@ -3375,7 +3388,7 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
}
// Check target paths.
- checkTargetPathsInCreateLogicalView(analysis, createLogicalViewStatement);
+ checkTargetPathsInCreateLogicalView(analysis, createLogicalViewStatement,
context);
if (analysis.isFinishQueryAfterAnalyze()) {
return analysis;
}
@@ -3546,7 +3559,9 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
}
private void checkTargetPathsInCreateLogicalView(
- Analysis analysis, CreateLogicalViewStatement
createLogicalViewStatement) {
+ Analysis analysis,
+ CreateLogicalViewStatement createLogicalViewStatement,
+ MPPQueryContext context) {
Pair<Boolean, String> checkResult =
createLogicalViewStatement.checkTargetPaths();
if (Boolean.FALSE.equals(checkResult.left)) {
analysis.setFinishQueryAfterAnalyze(true);
@@ -3574,7 +3589,7 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
// Make sure all paths are not under any templates
try {
for (PartialPath path : createLogicalViewStatement.getTargetPathList()) {
- checkIsTemplateCompatible(path, null);
+ checkIsTemplateCompatible(path, null, context);
}
} catch (Exception e) {
analysis.setFinishQueryAfterAnalyze(true);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/DataNodeSchemaCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/DataNodeSchemaCache.java
index cc5dd91fea2..0284a62ee72 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/DataNodeSchemaCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/DataNodeSchemaCache.java
@@ -64,12 +64,6 @@ public class DataNodeSchemaCache {
// cache update or clean have higher priority than cache read
private final ReentrantReadWriteLock readWriteLock = new
ReentrantReadWriteLock(false);
- // To make insert and delete executive, I import a new rwlock to datanode
schema.
- // The reason is that some operations [loadtsfiles] will hold readlock then
acquire writelock
- // before
- // closing the tsfile and resting the datanodeCache.
- private final ReentrantReadWriteLock insertDeletionLock = new
ReentrantReadWriteLock(false);
-
private DataNodeSchemaCache() {
deviceUsingTemplateSchemaCache = new
DeviceUsingTemplateSchemaCache(templateManager);
timeSeriesSchemaCache = new TimeSeriesSchemaCache();
@@ -111,22 +105,6 @@ public class DataNodeSchemaCache {
readWriteLock.writeLock().unlock();
}
- public void takeInsertLock() {
- insertDeletionLock.readLock().lock();
- }
-
- public void releaseInsertLock() {
- insertDeletionLock.readLock().unlock();
- }
-
- public void takeDeleteLock() {
- insertDeletionLock.writeLock().lock();
- }
-
- public void releaseDeleteLock() {
- insertDeletionLock.writeLock().unlock();
- }
-
/**
* Get SchemaEntity info without auto create schema
*
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/lock/DataNodeSchemaLockManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/lock/DataNodeSchemaLockManager.java
new file mode 100644
index 00000000000..b57986ccdec
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/lock/DataNodeSchemaLockManager.java
@@ -0,0 +1,59 @@
+/*
+ * 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.analyze.lock;
+
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class DataNodeSchemaLockManager {
+
+ private final ReentrantReadWriteLock[] locks;
+
+ private static class DataNodeSchemaLockManagerHold {
+ private static final DataNodeSchemaLockManager INSTANCE = new
DataNodeSchemaLockManager();
+ }
+
+ public static DataNodeSchemaLockManager getInstance() {
+ return DataNodeSchemaLockManagerHold.INSTANCE;
+ }
+
+ private DataNodeSchemaLockManager() {
+ int lockNum = SchemaLockType.values().length;
+ this.locks = new ReentrantReadWriteLock[lockNum];
+ for (int i = 0; i < lockNum; i++) {
+ locks[i] = new ReentrantReadWriteLock(false);
+ }
+ }
+
+ public void takeReadLock(SchemaLockType lockType) {
+ locks[lockType.ordinal()].readLock().lock();
+ }
+
+ public void releaseReadLock(SchemaLockType lockType) {
+ locks[lockType.ordinal()].readLock().unlock();
+ }
+
+ public void takeWriteLock(SchemaLockType lockType) {
+ locks[lockType.ordinal()].writeLock().lock();
+ }
+
+ public void releaseWriteLock(SchemaLockType lockType) {
+ locks[lockType.ordinal()].writeLock().unlock();
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/lock/SchemaLockType.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/lock/SchemaLockType.java
new file mode 100644
index 00000000000..b72a67c1fac
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/lock/SchemaLockType.java
@@ -0,0 +1,51 @@
+/*
+ * 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.analyze.lock;
+
+/**
+ * This enum class is used for maintaining types of locks used on DataNode for
distributed
+ * concurrency safety, especially for scenarios with DataNode and ConfigNode
interactions.
+ */
+public enum SchemaLockType {
+
+ /**
+ * This lock is used for guarantee no timeseries under template set path.
+ *
+ * <ol>
+ * <li>Take read lock before creating timeseries.
+ * <li>Release read lock after finishing creating timeseries
+ * <li>Take write lock before pre-set template.
+ * <li>Release write lock after finishing pre-set template.
+ * </ol>
+ */
+ TIMESERIES_VS_TEMPLATE,
+
+ /**
+ * This lock is used for guarantee no data without schema after timeseries
deletion.
+ *
+ * <ol>
+ * <li>Take read lock before validating schema during inserting data or
loading TsFile.
+ * <li>Release read lock after finishing inserting data or loading TsFile.
+ * <li>Take write lock before invalidating schema cache during timeseries
deletion.
+ * <li>Release write lock after finishing invalidating schema cache.
+ * </ol>
+ */
+ VALIDATE_VS_DELETION
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ClusterSchemaFetcher.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ClusterSchemaFetcher.java
index 4e544bb241d..14957b843e9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ClusterSchemaFetcher.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ClusterSchemaFetcher.java
@@ -29,6 +29,8 @@ import
org.apache.iotdb.db.queryengine.common.schematree.DeviceSchemaInfo;
import org.apache.iotdb.db.queryengine.common.schematree.ISchemaTree;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
import
org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache;
+import
org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager;
+import org.apache.iotdb.db.queryengine.plan.analyze.lock.SchemaLockType;
import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager;
import org.apache.iotdb.db.schemaengine.template.ITemplateManager;
import org.apache.iotdb.db.schemaengine.template.Template;
@@ -165,8 +167,8 @@ public class ClusterSchemaFetcher implements ISchemaFetcher
{
MPPQueryContext context) {
// The schema cache R/W and fetch operation must be locked together thus
the cache clean
// operation executed by delete timeseries will be effective.
- schemaCache.takeInsertLock();
- context.addAcquiredLockNum();
+
DataNodeSchemaLockManager.getInstance().takeReadLock(SchemaLockType.VALIDATE_VS_DELETION);
+ context.addAcquiredLockNum(SchemaLockType.VALIDATE_VS_DELETION);
schemaCache.takeReadLock();
try {
Pair<Template, PartialPath> templateSetInfo =
@@ -203,8 +205,8 @@ public class ClusterSchemaFetcher implements ISchemaFetcher
{
MPPQueryContext context) {
// The schema cache R/W and fetch operation must be locked together thus
the cache clean
// operation executed by delete timeseries will be effective.
- schemaCache.takeInsertLock();
- context.addAcquiredLockNum();
+
DataNodeSchemaLockManager.getInstance().takeReadLock(SchemaLockType.VALIDATE_VS_DELETION);
+ context.addAcquiredLockNum(SchemaLockType.VALIDATE_VS_DELETION);
schemaCache.takeReadLock();
try {
@@ -247,8 +249,8 @@ public class ClusterSchemaFetcher implements ISchemaFetcher
{
MPPQueryContext context) {
// The schema cache R/W and fetch operation must be locked together thus
the cache clean
// operation executed by delete timeseries will be effective.
- schemaCache.takeInsertLock();
- context.addAcquiredLockNum();
+
DataNodeSchemaLockManager.getInstance().takeReadLock(SchemaLockType.VALIDATE_VS_DELETION);
+ context.addAcquiredLockNum(SchemaLockType.VALIDATE_VS_DELETION);
schemaCache.takeReadLock();
try {
ClusterSchemaTree schemaTree = new ClusterSchemaTree();
