This is an automated email from the ASF dual-hosted git repository. jianyun pushed a commit to branch rocksdb/dev in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit c2d9d588ca085e6f944afdfc7d0506cf8590788e Author: chengjianyun <[email protected]> AuthorDate: Mon Mar 7 11:47:28 2022 +0800 [rocksdb] refine key exist check logic --- .../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 3 + .../iotdb/db/metadata/rocksdb/CheckKeyResult.java | 9 +++ .../iotdb/db/metadata/rocksdb/MRocksDBManager.java | 11 ++- .../metadata/rocksdb/RocksDBReadWriteHandler.java | 81 +++++++++------------- .../rocksdb/RocksDBReadWriteHandlerTest.java | 57 +++++++++++++++ 5 files changed, 106 insertions(+), 55 deletions(-) diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java index 5751a45..85383e3 100644 --- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java +++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java @@ -849,6 +849,9 @@ public class IoTDBDescriptor { // CQ loadCQProps(properties); + // meta manager + loadMetadataConfig(properties); + } catch (FileNotFoundException e) { logger.warn("Fail to find config file {}", url, e); } catch (IOException e) { diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/rocksdb/CheckKeyResult.java b/server/src/main/java/org/apache/iotdb/db/metadata/rocksdb/CheckKeyResult.java index 5f6ce05..25f9ab1 100644 --- a/server/src/main/java/org/apache/iotdb/db/metadata/rocksdb/CheckKeyResult.java +++ b/server/src/main/java/org/apache/iotdb/db/metadata/rocksdb/CheckKeyResult.java @@ -25,6 +25,7 @@ public class CheckKeyResult { private boolean[] result = new boolean[MAX_NODE_TYPE_NUM]; private boolean existAnyKey = false; + private byte[] value; public void setSingleCheckValue(char index, boolean value) { if (value) { @@ -37,6 +38,14 @@ public class CheckKeyResult { return existAnyKey; } + public byte[] getValue() { + return value; + } + + public void setValue(byte[] value) { + this.value = value; + } + public boolean getResult(RocksDBMNodeType type) { return result[type.value]; } diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/rocksdb/MRocksDBManager.java b/server/src/main/java/org/apache/iotdb/db/metadata/rocksdb/MRocksDBManager.java index 9ca93ad..ff41b13 100644 --- a/server/src/main/java/org/apache/iotdb/db/metadata/rocksdb/MRocksDBManager.java +++ b/server/src/main/java/org/apache/iotdb/db/metadata/rocksdb/MRocksDBManager.java @@ -120,7 +120,6 @@ import java.util.stream.Collectors; import static org.apache.iotdb.db.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD; import static org.apache.iotdb.db.conf.IoTDBConstant.ONE_LEVEL_PATH_WILDCARD; import static org.apache.iotdb.db.metadata.rocksdb.RockDBConstants.*; -import static org.apache.iotdb.db.metadata.rocksdb.RocksDBUtils.*; import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR; /** @@ -375,12 +374,11 @@ public class MRocksDBManager implements IMetaManager { return; } String levelPath = RocksDBUtils.getLevelPath(nodes, start - 1); - Holder<byte[]> holder = new Holder<>(); Lock lock = locksPool.get(levelPath); if (lock.tryLock(MAX_LOCK_WAIT_TIME, TimeUnit.MILLISECONDS)) { lockedLocks.push(lock); try { - CheckKeyResult checkResult = readWriteHandler.keyExistByAllTypes(levelPath, holder); + CheckKeyResult checkResult = readWriteHandler.keyExistByAllTypes(levelPath); if (!checkResult.existAnyKey()) { createTimeSeriesRecursively( nodes, start - 1, end, schema, alias, tags, attributes, lockedLocks); @@ -406,7 +404,7 @@ public class MRocksDBManager implements IMetaManager { // convert the parent node to entity if it is internal node readWriteHandler.convertToEntityNode(levelPath, DEFAULT_NODE_VALUE); } else if (checkResult.getResult(RocksDBMNodeType.ENTITY)) { - if ((holder.getValue()[1] & FLAG_IS_ALIGNED) != 0) { + if ((checkResult.getValue()[1] & FLAG_IS_ALIGNED) != 0) { throw new AlignedTimeseriesException( "Timeseries under this entity is aligned, please use createAlignedTimeseries or change entity.", levelPath); @@ -582,11 +580,10 @@ public class MRocksDBManager implements IMetaManager { return; } String levelPath = RocksDBUtils.getLevelPath(nodes, start - 1); - Holder<byte[]> holder = new Holder<>(); Lock lock = locksPool.get(levelPath); if (lock.tryLock(MAX_LOCK_WAIT_TIME, TimeUnit.MILLISECONDS)) { try { - CheckKeyResult checkResult = readWriteHandler.keyExistByAllTypes(levelPath, holder); + CheckKeyResult checkResult = readWriteHandler.keyExistByAllTypes(levelPath); if (!checkResult.existAnyKey()) { createEntityRecursively(nodes, start - 1, end, aligned, lockedLocks); if (start == nodes.length) { @@ -602,7 +599,7 @@ public class MRocksDBManager implements IMetaManager { throw new PathAlreadyExistException("Node already exists but not entity"); } - if ((holder.getValue()[1] & FLAG_IS_ALIGNED) != 0) { + if ((checkResult.getValue()[1] & FLAG_IS_ALIGNED) != 0) { throw new PathAlreadyExistException("Entity node exists but not aligned"); } } else if (checkResult.getResult(RocksDBMNodeType.MEASUREMENT) diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/rocksdb/RocksDBReadWriteHandler.java b/server/src/main/java/org/apache/iotdb/db/metadata/rocksdb/RocksDBReadWriteHandler.java index 9ab5f42..0e701ea 100644 --- a/server/src/main/java/org/apache/iotdb/db/metadata/rocksdb/RocksDBReadWriteHandler.java +++ b/server/src/main/java/org/apache/iotdb/db/metadata/rocksdb/RocksDBReadWriteHandler.java @@ -28,6 +28,7 @@ import org.apache.iotdb.db.utils.TestOnly; import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils; import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; +import com.google.common.primitives.Bytes; import org.rocksdb.ColumnFamilyDescriptor; import org.rocksdb.ColumnFamilyHandle; import org.rocksdb.ColumnFamilyOptions; @@ -88,7 +89,7 @@ public class RocksDBReadWriteHandler { RocksDB.loadLibrary(); } - private RocksDBReadWriteHandler() throws RocksDBException { + public RocksDBReadWriteHandler() throws RocksDBException { Options options = new Options(); options.setCreateIfMissing(true); options.setAllowMmapReads(true); @@ -165,8 +166,9 @@ public class RocksDBReadWriteHandler { rocksDB.put(key, value); } - public void createNode(String key, RocksDBMNodeType type, byte[] value) throws RocksDBException { - byte[] nodeKey = RocksDBUtils.toRocksDBKey(key, type.value); + public void createNode(String levelKey, RocksDBMNodeType type, byte[] value) + throws RocksDBException { + byte[] nodeKey = RocksDBUtils.toRocksDBKey(levelKey, type.value); rocksDB.put(nodeKey, value); } @@ -220,11 +222,6 @@ public class RocksDBReadWriteHandler { } public CheckKeyResult keyExistByAllTypes(String levelKey) throws RocksDBException { - return keyExistByAllTypes(levelKey, new Holder<>()); - } - - public CheckKeyResult keyExistByAllTypes(String levelKey, Holder<byte[]> holder) - throws RocksDBException { RocksDBMNodeType[] types = new RocksDBMNodeType[] { RocksDBMNodeType.ALISA, @@ -233,50 +230,36 @@ public class RocksDBReadWriteHandler { RocksDBMNodeType.MEASUREMENT, RocksDBMNodeType.STORAGE_GROUP }; - return keyExistByTypes(levelKey, holder, types); + return keyExistByTypes(levelKey, types); } public CheckKeyResult keyExistByTypes(String levelKey, RocksDBMNodeType... types) throws RocksDBException { - return keyExistByTypes(levelKey, new Holder<>(), types); - } - - public CheckKeyResult keyExistByTypes( - String levelKey, Holder<byte[]> holder, RocksDBMNodeType... types) throws RocksDBException { - // TODO: compare the performance between two methods CheckKeyResult result = new CheckKeyResult(); - for (RocksDBMNodeType type : types) { - byte[] key = RocksDBUtils.toRocksDBKey(levelKey, type.value); - if (keyExist(key, holder)) { - result.setSingleCheckValue(type.value, keyExist(key, holder)); - break; + try { + Arrays.stream(types) + // .parallel() + .forEach( + x -> { + byte[] key = Bytes.concat(new byte[] {(byte) x.value}, levelKey.getBytes()); + try { + Holder<byte[]> holder = new Holder<>(); + boolean keyExisted = keyExist(key, holder); + if (keyExisted) { + result.setSingleCheckValue(x.value, true); + result.setValue(holder.getValue()); + } + } catch (RocksDBException e) { + throw new RuntimeException(e); + } + }); + } catch (Exception e) { + if (e.getCause() instanceof RocksDBException) { + throw (RocksDBException) e.getCause(); } + throw e; } return result; - - // try { - // Arrays.stream(types) - // .parallel() - // .forEach( - // x -> { - // byte[] key = Bytes.concat(new byte[] {x.value}, levelKey.getBytes()); - // try { - // boolean keyExisted = keyExist(key, holder); - // if (keyExisted) { - // holder.getValue(); - // result.setSingleCheckValue(x.value, true); - // } - // } catch (RocksDBException e) { - // throw new RuntimeException(e); - // } - // }); - // } catch (Exception e) { - // if (e.getCause() instanceof RocksDBException) { - // throw (RocksDBException) e.getCause(); - // } - // throw e; - // } - // return result; } public boolean keyExist(byte[] key, Holder<byte[]> holder) throws RocksDBException { @@ -284,10 +267,12 @@ public class RocksDBReadWriteHandler { if (!rocksDB.keyMayExist(key, holder)) { exist = false; } else { - byte[] value = rocksDB.get(key); - if (value != null) { - exist = true; - holder.setValue(value); + if (holder.getValue() != null) { + byte[] value = rocksDB.get(key); + if (value != null) { + exist = true; + holder.setValue(value); + } } } return exist; diff --git a/server/src/test/java/org/apache/iotdb/db/metadata/rocksdb/RocksDBReadWriteHandlerTest.java b/server/src/test/java/org/apache/iotdb/db/metadata/rocksdb/RocksDBReadWriteHandlerTest.java new file mode 100644 index 0000000..5f605c7 --- /dev/null +++ b/server/src/test/java/org/apache/iotdb/db/metadata/rocksdb/RocksDBReadWriteHandlerTest.java @@ -0,0 +1,57 @@ +package org.apache.iotdb.db.metadata.rocksdb; + +import org.apache.iotdb.db.exception.metadata.IllegalPathException; +import org.apache.iotdb.db.exception.metadata.MetadataException; +import org.apache.iotdb.db.metadata.path.PartialPath; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.rocksdb.RocksDBException; + +import java.io.File; +import java.util.ArrayList; +import java.util.List; + +import static org.apache.iotdb.db.metadata.rocksdb.RocksDBReadWriteHandler.ROCKSDB_PATH; + +public class RocksDBReadWriteHandlerTest { + + private RocksDBReadWriteHandler readWriteHandler; + + @Before + public void setUp() throws MetadataException, RocksDBException { + File file = new File(ROCKSDB_PATH); + if (!file.exists()) { + file.mkdirs(); + } + readWriteHandler = new RocksDBReadWriteHandler(); + } + + @Test + public void testKeyExistByTypes() throws IllegalPathException, RocksDBException { + List<PartialPath> timeseries = new ArrayList<>(); + timeseries.add(new PartialPath("root.sg.d1.m1")); + timeseries.add(new PartialPath("root.sg.d1.m2")); + timeseries.add(new PartialPath("root.sg.d2.m1")); + timeseries.add(new PartialPath("root.sg.d2.m2")); + timeseries.add(new PartialPath("root.sg1.d1.m1")); + timeseries.add(new PartialPath("root.sg1.d1.m2")); + timeseries.add(new PartialPath("root.sg1.d2.m1")); + timeseries.add(new PartialPath("root.sg1.d2.m2")); + + for (PartialPath path : timeseries) { + String levelPath = RocksDBUtils.getLevelPath(path.getNodes(), path.getNodeLength() - 1); + readWriteHandler.createNode( + levelPath, RocksDBMNodeType.MEASUREMENT, path.getFullPath().getBytes()); + } + + for (PartialPath path : timeseries) { + String levelPath = RocksDBUtils.getLevelPath(path.getNodes(), path.getNodeLength() - 1); + CheckKeyResult result = readWriteHandler.keyExistByAllTypes(levelPath); + Assert.assertTrue(result.existAnyKey()); + Assert.assertNotNull(result.getValue()); + Assert.assertEquals(path.getFullPath(), new String(result.getValue())); + } + } +}
