This is an automated email from the ASF dual-hosted git repository. qiaojialin pushed a commit to branch ImproveLock in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 94d1146ed2c9f384ba3bf5794f4edde3d01028a5 Author: qiaojialin <[email protected]> AuthorDate: Thu Sep 24 11:33:36 2020 +0800 Improve the readUnLock way while inserting --- .../org/apache/iotdb/db/metadata/MManager.java | 124 ++++++++++----------- .../apache/iotdb/db/qp/executor/PlanExecutor.java | 25 +++-- .../iotdb/db/qp/physical/crud/InsertPlan.java | 2 + .../iotdb/db/qp/physical/crud/InsertRowPlan.java | 49 ++++---- .../iotdb/db/writelog/recover/LogReplayer.java | 3 +- .../iotdb/db/engine/storagegroup/TTLTest.java | 11 +- 6 files changed, 101 insertions(+), 113 deletions(-) diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java index 6deb9c8..353cfff 100644 --- a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java +++ b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java @@ -183,8 +183,6 @@ public class MManager { /** * we should not use this function in other place, but only in IoTDB class - * - * @return */ public static MManager getInstance() { return MManagerHolder.INSTANCE; @@ -347,7 +345,7 @@ public class MManager { break; case MetadataOperationType.DELETE_STORAGE_GROUP: List<PartialPath> storageGroups = new ArrayList<>(); - for(int i = 1; i <= args.length; i++) { + for (int i = 1; i <= args.length; i++) { storageGroups.add(new PartialPath(args[i])); } deleteStorageGroups(storageGroups); @@ -437,9 +435,9 @@ public class MManager { /** * Add one timeseries to metadata tree, if the timeseries already exists, throw exception * - * @param path the timeseries path - * @param dataType the dateType {@code DataType} of the timeseries - * @param encoding the encoding function {@code Encoding} of the timeseries + * @param path the timeseries path + * @param dataType the dateType {@code DataType} of the timeseries + * @param encoding the encoding function {@code Encoding} of the timeseries * @param compressor the compressor function {@code Compressor} of the time series * @return whether the measurement occurs for the first time in this storage group (if true, the * measurement should be registered to the StorageEngine too) @@ -549,7 +547,8 @@ public class MManager { throws MetadataException, IOException { lock.writeLock().lock(); try { - Pair<PartialPath, MeasurementMNode> pair = mtree.deleteTimeseriesAndReturnEmptyStorageGroup(path); + Pair<PartialPath, MeasurementMNode> pair = mtree + .deleteTimeseriesAndReturnEmptyStorageGroup(path); removeFromTagInvertedIndex(pair.right); PartialPath storageGroupPath = pair.left; @@ -708,7 +707,7 @@ public class MManager { * Get all devices under given prefixPath. * * @param prefixPath a prefix of a full path. if the wildcard is not at the tail, then each - * wildcard can only match one level, otherwise it can match to the tail. + * wildcard can only match one level, otherwise it can match to the tail. * @return A HashSet instance which stores devices paths with given prefixPath. */ public Set<PartialPath> getDevices(PartialPath prefixPath) throws MetadataException { @@ -724,16 +723,18 @@ public class MManager { * Get all nodes from the given level * * @param prefixPath can be a prefix of a full path. Can not be a full path. can not have - * wildcard. But, the level of the prefixPath can be smaller than the given - * level, e.g., prefixPath = root.a while the given level is 5 - * @param nodeLevel the level can not be smaller than the level of the prefixPath + * wildcard. But, the level of the prefixPath can be smaller than the given level, e.g., + * prefixPath = root.a while the given level is 5 + * @param nodeLevel the level can not be smaller than the level of the prefixPath * @return A List instance which stores all node at given level */ - public List<PartialPath> getNodesList(PartialPath prefixPath, int nodeLevel) throws MetadataException { + public List<PartialPath> getNodesList(PartialPath prefixPath, int nodeLevel) + throws MetadataException { return getNodesList(prefixPath, nodeLevel, null); } - public List<PartialPath> getNodesList(PartialPath prefixPath, int nodeLevel, StorageGroupFilter filter) + public List<PartialPath> getNodesList(PartialPath prefixPath, int nodeLevel, + StorageGroupFilter filter) throws MetadataException { lock.readLock().lock(); try { @@ -775,7 +776,7 @@ public class MManager { * Get all storage group under given prefixPath. * * @param prefixPath a prefix of a full path. if the wildcard is not at the tail, then each - * wildcard can only match one level, otherwise it can match to the tail. + * wildcard can only match one level, otherwise it can match to the tail. * @return A ArrayList instance which stores storage group paths with given prefixPath. */ public List<PartialPath> getStorageGroupPaths(PartialPath prefixPath) throws MetadataException { @@ -804,7 +805,7 @@ public class MManager { * expression in this method is formed by the amalgamation of seriesPath and the character '*'. * * @param prefixPath can be a prefix or a full path. if the wildcard is not at the tail, then each - * wildcard can only match one level, otherwise it can match to the tail. + * wildcard can only match one level, otherwise it can match to the tail. */ public List<PartialPath> getAllTimeseriesPath(PartialPath prefixPath) throws MetadataException { lock.readLock().lock(); @@ -816,10 +817,10 @@ public class MManager { } /** - * Similar to method getAllTimeseriesPath(), but return Path with alias - * alias. + * Similar to method getAllTimeseriesPath(), but return Path with alias alias. */ - public List<PartialPath> getAllTimeseriesPathWithAlias(PartialPath prefixPath) throws MetadataException { + public List<PartialPath> getAllTimeseriesPathWithAlias(PartialPath prefixPath) + throws MetadataException { lock.readLock().lock(); try { return mtree.getAllTimeseriesPathWithAlias(prefixPath); @@ -868,7 +869,7 @@ public class MManager { * To calculate the count of nodes in the given level for given prefix path. * * @param prefixPath a prefix path or a full path, can not contain '*' - * @param level the level can not be smaller than the level of the prefixPath + * @param level the level can not be smaller than the level of the prefixPath */ public int getNodesCountInGivenLevel(PartialPath prefixPath, int level) throws MetadataException { lock.readLock().lock(); @@ -940,7 +941,8 @@ public class MManager { pair.left.putAll(pair.right); MeasurementSchema measurementSchema = leaf.getSchema(); res.add(new ShowTimeSeriesResult(leaf.getFullPath(), leaf.getAlias(), - getStorageGroupPath(leaf.getPartialPath()).getFullPath(), measurementSchema.getType().toString(), + getStorageGroupPath(leaf.getPartialPath()).getFullPath(), + measurementSchema.getType().toString(), measurementSchema.getEncodingType().toString(), measurementSchema.getCompressor().toString(), pair.left)); if (limit != 0) { @@ -1005,19 +1007,22 @@ public class MManager { try { if (tagFileOffset < 0) { // no tags/attributes - res.add(new ShowTimeSeriesResult(ansString.left.getFullPath(), ansString.right[0], ansString.right[1], ansString.right[2], + res.add(new ShowTimeSeriesResult(ansString.left.getFullPath(), ansString.right[0], + ansString.right[1], ansString.right[2], ansString.right[3], ansString.right[4], Collections.emptyMap())); } else { // has tags/attributes Pair<Map<String, String>, Map<String, String>> pair = tagLogFile.read(config.getTagAttributeTotalSize(), tagFileOffset); pair.left.putAll(pair.right); - res.add(new ShowTimeSeriesResult(ansString.left.getFullPath(), ansString.right[0], ansString.right[1], ansString.right[2], + res.add(new ShowTimeSeriesResult(ansString.left.getFullPath(), ansString.right[0], + ansString.right[1], ansString.right[2], ansString.right[3], ansString.right[4], pair.left)); } } catch (IOException e) { throw new MetadataException( - "Something went wrong while deserialize tag info of " + ansString.left.getFullPath(), e); + "Something went wrong while deserialize tag info of " + ansString.left.getFullPath(), + e); } } return res; @@ -1088,13 +1093,12 @@ public class MManager { } /** - * E.g., root.sg is storage group - * given [root, sg], return the MNode of root.sg - * given [root, sg, device], return the MNode of root.sg - * Get storage group node by path. If storage group is not set, StorageGroupNotSetException will - * be thrown + * E.g., root.sg is storage group given [root, sg], return the MNode of root.sg given [root, sg, + * device], return the MNode of root.sg Get storage group node by path. If storage group is not + * set, StorageGroupNotSetException will be thrown */ - public StorageGroupMNode getStorageGroupNodeByStorageGroupPath(PartialPath path) throws MetadataException { + public StorageGroupMNode getStorageGroupNodeByStorageGroupPath(PartialPath path) + throws MetadataException { lock.readLock().lock(); try { return mtree.getStorageGroupNodeByStorageGroupPath(path); @@ -1274,7 +1278,7 @@ public class MManager { * Check whether the given path contains a storage group change or set the new offset of a * timeseries * - * @param path timeseries + * @param path timeseries * @param offset offset in the tag file */ public void changeOffset(PartialPath path, long offset) throws MetadataException { @@ -1304,14 +1308,15 @@ public class MManager { * upsert tags and attributes key-value for the timeseries if the key has existed, just use the * new value to update it. * - * @param alias newly added alias - * @param tagsMap newly added tags map + * @param alias newly added alias + * @param tagsMap newly added tags map * @param attributesMap newly added attributes map - * @param fullPath timeseries + * @param fullPath timeseries */ @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning public void upsertTagsAndAttributes(String alias, Map<String, String> tagsMap, - Map<String, String> attributesMap, PartialPath fullPath) throws MetadataException, IOException { + Map<String, String> attributesMap, PartialPath fullPath) + throws MetadataException, IOException { lock.writeLock().lock(); try { MNode mNode = mtree.getNodeByPath(fullPath); @@ -1411,7 +1416,7 @@ public class MManager { * add new attributes key-value for the timeseries * * @param attributesMap newly added attributes map - * @param fullPath timeseries + * @param fullPath timeseries */ public void addAttributes(Map<String, String> attributesMap, PartialPath fullPath) throws MetadataException, IOException { @@ -1453,7 +1458,7 @@ public class MManager { /** * add new tags key-value for the timeseries * - * @param tagsMap newly added tags map + * @param tagsMap newly added tags map * @param fullPath timeseries */ public void addTags(Map<String, String> tagsMap, PartialPath fullPath) @@ -1506,7 +1511,7 @@ public class MManager { /** * drop tags or attributes of the timeseries * - * @param keySet tags key or attributes key + * @param keySet tags key or attributes key * @param fullPath timeseries path */ @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning @@ -1657,8 +1662,8 @@ public class MManager { /** * rename the tag or attribute's key of the timeseries * - * @param oldKey old key of tag or attribute - * @param newKey new key of tag or attribute + * @param oldKey old key of tag or attribute + * @param newKey new key of tag or attribute * @param fullPath timeseries */ @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning @@ -1792,9 +1797,6 @@ public class MManager { /** * Collect the timeseries schemas under "startingPath". - * - * @param startingPath - * @param measurementSchemas */ public void collectSeries(PartialPath startingPath, List<MeasurementSchema> measurementSchemas) { MNode mNode; @@ -1936,17 +1938,20 @@ public class MManager { /** * get schema for device. Attention!!! Only support insertPlan - * - * @throws MetadataException */ @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning - public MeasurementMNode[] getSeriesSchemasAndReadLockDevice(PartialPath deviceId, - String[] measurementList, InsertPlan plan) throws MetadataException { - MeasurementMNode[] mNodes = new MeasurementMNode[measurementList.length]; + public MNode getSeriesSchemasAndReadLockDevice(InsertPlan plan) + throws MetadataException { + + PartialPath deviceId = plan.getDeviceId(); + String[] measurementList = plan.getMeasurements(); + MeasurementMNode[] measurementMNodes = plan.getMeasurementMNodes(); + // 1. get device node MNode deviceMNode = getDeviceNodeWithAutoCreateAndReadLock(deviceId); + // 2. get schema of each measurement for (int i = 0; i < measurementList.length; i++) { try { @@ -1994,10 +1999,10 @@ public class MManager { } } - mNodes[i] = measurementMNode; + measurementMNodes[i] = measurementMNode; // set measurementName instead of alias - measurementList[i] = mNodes[i].getName(); + measurementList[i] = measurementMNode.getName(); } catch (MetadataException e) { logger.warn("meet error when check {}.{}, message: {}", deviceId, measurementList[i], @@ -2011,14 +2016,15 @@ public class MManager { } } - return mNodes; + return deviceMNode; } /** * create timeseries with ignore PathAlreadyExistException */ - private void internalCreateTimeseries(PartialPath path, TSDataType dataType) throws MetadataException { + private void internalCreateTimeseries(PartialPath path, TSDataType dataType) + throws MetadataException { try { createTimeseries( path, @@ -2060,8 +2066,6 @@ public class MManager { /** * get dataType of plan, in loc measurements only support InsertRowPlan and InsertTabletPlan - * - * @throws MetadataException */ private TSDataType getTypeInLoc(InsertPlan plan, int loc) throws MetadataException { TSDataType dataType; @@ -2078,18 +2082,4 @@ public class MManager { return dataType; } - /** - * when insert, we lock device node for not create deleted time series after insert, we should - * call this function to unlock the device node - * - * @param deviceId - */ - public void unlockDeviceReadLock(PartialPath deviceId) { - try { - MNode mNode = getDeviceNode(deviceId); - mNode.readUnlock(); - } catch (MetadataException e) { - // ignore the exception - } - } } diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java index d3eff10..1085074 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java @@ -912,18 +912,18 @@ public class PlanExecutor implements IPlanExecutor { } } - protected MeasurementMNode[] getSeriesSchemas(InsertPlan insertPlan) + private MNode getSeriesSchemas(InsertPlan insertPlan) throws MetadataException { - return mManager - .getSeriesSchemasAndReadLockDevice(insertPlan.getDeviceId(), insertPlan.getMeasurements(), - insertPlan); + return mManager.getSeriesSchemasAndReadLockDevice(insertPlan); } @Override public void insert(InsertRowPlan insertRowPlan) throws QueryProcessException { + MNode deviceNode = null; try { - MeasurementMNode[] mNodes = getSeriesSchemas(insertRowPlan); - insertRowPlan.setMNodesAndTransferType(mNodes); + insertRowPlan.setMeasurementMNodes(new MeasurementMNode[insertRowPlan.getMeasurements().length]); + deviceNode = getSeriesSchemas(insertRowPlan); + insertRowPlan.transferType(); StorageEngine.getInstance().insert(insertRowPlan); if (insertRowPlan.getFailedMeasurements() != null) { throw new StorageEngineException( @@ -932,15 +932,18 @@ public class PlanExecutor implements IPlanExecutor { } catch (StorageEngineException | MetadataException e) { throw new QueryProcessException(e); } finally { - mManager.unlockDeviceReadLock(insertRowPlan.getDeviceId()); + if (deviceNode != null) { + deviceNode.readUnlock(); + } } } @Override public void insertTablet(InsertTabletPlan insertTabletPlan) throws QueryProcessException { + MNode deviceMNode = null; try { - MeasurementMNode[] mNodes = getSeriesSchemas(insertTabletPlan); - insertTabletPlan.setMeasurementMNodes(mNodes); + insertTabletPlan.setMeasurementMNodes(new MeasurementMNode[insertTabletPlan.getMeasurements().length]); + deviceMNode = getSeriesSchemas(insertTabletPlan); StorageEngine.getInstance().insertTablet(insertTabletPlan); if (insertTabletPlan.getFailedMeasurements() != null) { throw new StorageEngineException( @@ -949,7 +952,9 @@ public class PlanExecutor implements IPlanExecutor { } catch (StorageEngineException | MetadataException e) { throw new QueryProcessException(e); } finally { - mManager.unlockDeviceReadLock(insertTabletPlan.getDeviceId()); + if (deviceMNode != null) { + deviceMNode.readUnlock(); + } } } diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java index 38e9318..2d33a89 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java @@ -31,7 +31,9 @@ abstract public class InsertPlan extends PhysicalPlan { protected PartialPath deviceId; protected String[] measurements; + // get from client protected TSDataType[] dataTypes; + // get from MManager protected MeasurementMNode[] measurementMNodes; // record the failed measurements diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowPlan.java index 3b3f06c..1271ecd 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowPlan.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowPlan.java @@ -63,6 +63,19 @@ public class InsertRowPlan extends InsertPlan { super(OperatorType.INSERT); } + public InsertRowPlan(PartialPath deviceId, long insertTime, String[] measurementList, + String[] insertValues) { + super(Operator.OperatorType.INSERT); + this.time = insertTime; + this.deviceId = deviceId; + this.measurements = measurementList; + this.dataTypes = new TSDataType[measurements.length]; + // We need to create an Object[] for the data type casting, because we can not set Float, Long to String[i] + this.values = new Object[measurements.length]; + System.arraycopy(insertValues, 0, values, 0, measurements.length); + isNeedInferType = true; + } + @TestOnly public InsertRowPlan(PartialPath deviceId, long insertTime, String[] measurements, TSDataType[] dataTypes, String[] insertValues) { @@ -97,6 +110,7 @@ public class InsertRowPlan extends InsertPlan { } } + @TestOnly public InsertRowPlan(TSRecord tsRecord) throws IllegalPathException { super(OperatorType.INSERT); this.deviceId = new PartialPath(tsRecord.deviceId); @@ -115,30 +129,6 @@ public class InsertRowPlan extends InsertPlan { } } - public InsertRowPlan(PartialPath deviceId, long insertTime, String[] measurementList, - TSDataType[] dataTypes, Object[] insertValues) { - super(Operator.OperatorType.INSERT); - this.time = insertTime; - this.deviceId = deviceId; - this.measurements = measurementList; - this.dataTypes = dataTypes; - this.values = insertValues; - } - - public InsertRowPlan(PartialPath deviceId, long insertTime, String[] measurementList, - String[] insertValues) { - super(Operator.OperatorType.INSERT); - this.time = insertTime; - this.deviceId = deviceId; - this.measurements = measurementList; - this.dataTypes = new TSDataType[measurements.length]; - // We need to create an Object[] for the data type casting, because we can not set Float, Long to String[i] - this.values = new Object[measurements.length]; - System.arraycopy(insertValues, 0, values, 0, measurements.length); - isNeedInferType = true; - } - - public long getTime() { return time; } @@ -160,11 +150,10 @@ public class InsertRowPlan extends InsertPlan { * Double, Binary) */ @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning - public void setMNodesAndTransferType(MeasurementMNode[] mNodes) throws QueryProcessException { - this.measurementMNodes = mNodes; + public void transferType() throws QueryProcessException { if (isNeedInferType) { - for (int i = 0; i < mNodes.length; i++) { - if (mNodes[i] == null) { + for (int i = 0; i < measurementMNodes.length; i++) { + if (measurementMNodes[i] == null) { if (IoTDBDescriptor.getInstance().getConfig().isEnablePartialInsert()) { markFailedMeasurementInsertion(i); } else { @@ -173,7 +162,7 @@ public class InsertRowPlan extends InsertPlan { } continue; } - dataTypes[i] = mNodes[i].getSchema().getType(); + dataTypes[i] = measurementMNodes[i].getSchema().getType(); try { values[i] = CommonUtils.parseValue(dataTypes[i], values[i].toString()); } catch (Exception e) { @@ -181,7 +170,7 @@ public class InsertRowPlan extends InsertPlan { measurements[i], values[i], dataTypes[i]); if (IoTDBDescriptor.getInstance().getConfig().isEnablePartialInsert()) { markFailedMeasurementInsertion(i); - mNodes[i] = null; + measurementMNodes[i] = null; } else { throw e; } diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java b/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java index 08c403e..7326921 100644 --- a/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java +++ b/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java @@ -166,8 +166,9 @@ public class LogReplayer { InsertRowPlan tPlan = (InsertRowPlan) plan; //only infer type when users pass a String value //WAL already serializes the real data type, so no need to infer type + tPlan.setMeasurementMNodes(mNodes); ((InsertRowPlan) plan).setNeedInferType(false); - tPlan.setMNodesAndTransferType(mNodes); + tPlan.transferType(); //mark failed plan manually checkDataTypeAndMarkFailed(mNodes, tPlan); recoverMemTable.insert(tPlan); diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java index 9cff5e5..01c4ea8 100644 --- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java +++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java @@ -46,7 +46,6 @@ import org.apache.iotdb.db.exception.metadata.MetadataException; import org.apache.iotdb.db.exception.query.OutOfTTLException; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.metadata.PartialPath; -import org.apache.iotdb.db.metadata.mnode.MNode; import org.apache.iotdb.db.metadata.mnode.MeasurementMNode; import org.apache.iotdb.db.metadata.mnode.StorageGroupMNode; import org.apache.iotdb.db.qp.Planner; @@ -145,8 +144,9 @@ public class TTLTest { plan.setMeasurements(new String[]{"s1"}); plan.setDataTypes(new TSDataType[]{TSDataType.INT64}); plan.setValues(new Object[]{1L}); - plan.setMNodesAndTransferType(new MeasurementMNode[]{new MeasurementMNode(null, null, - new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN), null)}); + plan.setMeasurementMNodes(new MeasurementMNode[]{new MeasurementMNode(null, null, + new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN), null)}); + plan.transferType(); // ok without ttl insertToStorageGroupProcessor(plan); @@ -173,8 +173,9 @@ public class TTLTest { plan.setMeasurements(new String[]{"s1"}); plan.setDataTypes(new TSDataType[]{TSDataType.INT64}); plan.setValues(new Object[]{1L}); - plan.setMNodesAndTransferType(new MeasurementMNode[]{new MeasurementMNode(null, null, - new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN), null)}); + plan.setMeasurementMNodes(new MeasurementMNode[]{new MeasurementMNode(null, null, + new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN), null)}); + plan.transferType(); long initTime = System.currentTimeMillis(); // sequence data
