anph
/
hugegraph-sync
forked from hugegraph/hugegraph-sync
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(store): integrate `store-rocksdb` submodule (#2513) 

TODO: delete "hugegraph-store/hg-store-rocksdb/src/main/java/place-holder.txt" later

---------

Co-authored-by: imbajin <jin@apache.org>
This commit is contained in:
Peng Junzhi 2024-04-18 22:11:44 -05:00 committed by GitHub
parent 7d335b2c72
commit b4bc1f066a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
23 changed files with 4294 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
hugegraph-store/hg-store-rocksdb/pom.xml Normal file
View File

@ -0,0 +1,79 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.hugegraph</groupId>
<artifactId>hugegraph-store</artifactId>
<version>${revision}</version>
<relativePath>../pom.xml</relativePath>
</parent>
<artifactId>hg-store-rocksdb</artifactId>
<dependencies>
<dependency>
<groupId>org.apache.hugegraph</groupId>
<artifactId>hugegraph-common</artifactId>
<version>1.3.0</version>
<exclusions>
<exclusion>
<groupId>org.glassfish.jersey.inject</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>org.glassfish.jersey.media</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>org.glassfish.jersey.connectors</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.hugegraph</groupId>
<artifactId>hg-store-common</artifactId>
</dependency>
<dependency>
<groupId>org.rocksdb</groupId>
<artifactId>rocksdbjni</artifactId>
<version>7.7.3</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.22</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
<version>2.15.0</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.83</version>
</dependency>
</dependencies>
</project>

56
hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/DBSnapshotMeta.java Normal file
View File

@ -0,0 +1,56 @@
/*
* 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.hugegraph.rocksdb.access;
import java.util.Date;
import java.util.HashMap;
import lombok.Data;
/**
* metadata of a Rocksdb snapshot
*/
@Data
public class DBSnapshotMeta {
/**
* graph name
*/
private String graphName;
/**
* partition id
*/
private int partitionId;
/**
* star key with base64 encoding
*/
private byte[] startKey;
/**
* end key with base64 encoding
*/
private byte[] endKey;
/**
* created time of the snapshot
*/
private Date createdDate;
/**
* sst files, key is column family name value is the sst file name of the column
* family
*/
private HashMap<String, String> sstFiles;
}

63
hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/DBStoreException.java Normal file
View File

@ -0,0 +1,63 @@
/*
* 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.hugegraph.rocksdb.access;
import java.io.InterruptedIOException;
public class DBStoreException extends RuntimeException {
private static final long serialVersionUID = 5956983547131986887L;
public DBStoreException(String message) {
super(message);
}
public DBStoreException(String message, Throwable cause) {
super(message, cause);
}
public DBStoreException(String message, Object... args) {
super(String.format(message, args));
}
public DBStoreException(String message, Throwable cause, Object... args) {
super(String.format(message, args), cause);
}
public DBStoreException(Throwable cause) {
this("Exception in DBStore " + cause.getMessage(), cause);
}
public static Throwable rootCause(Throwable e) {
Throwable cause = e;
while (cause.getCause() != null) {
cause = cause.getCause();
}
return cause;
}
public static boolean isInterrupted(Throwable e) {
Throwable rootCause = DBStoreException.rootCause(e);
return rootCause instanceof InterruptedException ||
rootCause instanceof InterruptedIOException;
}
public Throwable rootCause() {
return rootCause(this);
}
}

321
hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBFactory.java Normal file
View File

@ -0,0 +1,321 @@
/*
* 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.hugegraph.rocksdb.access;
import java.io.File;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;
import org.apache.commons.io.FileUtils;
import org.apache.hugegraph.config.HugeConfig;
import org.rocksdb.AbstractEventListener;
import org.rocksdb.Cache;
import org.rocksdb.CompactionJobInfo;
import org.rocksdb.MemoryUsageType;
import org.rocksdb.MemoryUtil;
import org.rocksdb.RocksDB;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public final class RocksDBFactory {
private static final List<RocksdbChangedListener> rocksdbChangedListeners = new ArrayList<>();
private static RocksDBFactory dbFactory;
static {
RocksDB.loadLibrary();
}
private final Map<String, RocksDBSession> dbSessionMap = new ConcurrentHashMap<>();
private final List<DBSessionWatcher> destroyGraphDBs = new CopyOnWriteArrayList<>();
private final ReentrantReadWriteLock operateLock;
ScheduledExecutorService scheduledExecutor;
private HugeConfig hugeConfig;
private RocksDBFactory() {
this.operateLock = new ReentrantReadWriteLock();
scheduledExecutor = Executors.newScheduledThreadPool(2);
scheduledExecutor.scheduleWithFixedDelay(() -> {
try {
Iterator<DBSessionWatcher> itr = destroyGraphDBs.listIterator();
while (itr.hasNext()) {
DBSessionWatcher watcher = itr.next();
if (0 == watcher.dbSession.getRefCount()) {
try {
watcher.dbSession.shutdown();
FileUtils.deleteDirectory(new File(watcher.dbSession.getDbPath()));
rocksdbChangedListeners.forEach(listener -> {
listener.onDBDeleted(watcher.dbSession.getGraphName(),
watcher.dbSession.getDbPath());
});
log.info("removed db {} and delete files",
watcher.dbSession.getDbPath());
} catch (Exception e) {
log.error("DestroyGraphDB exception {}", e);
}
destroyGraphDBs.remove(watcher);
} else if (watcher.timestamp < (System.currentTimeMillis() - 1800 * 1000)) {
log.warn("DB {} has not been deleted refCount is {}, time is {} seconds",
watcher.dbSession.getDbPath(),
watcher.dbSession.getRefCount(),
(System.currentTimeMillis() - watcher.timestamp) / 1000);
} else {
// 超时强制删除 (30min)
watcher.dbSession.forceResetRefCount();
}
}
} catch (Exception e) {
log.error("RocksDBFactory scheduledExecutor exception {}", e);
}
}, 60, 60, TimeUnit.SECONDS);
}
public static RocksDBFactory getInstance() {
if (dbFactory == null) {
synchronized (RocksDBFactory.class) {
if (dbFactory == null) {
dbFactory = new RocksDBFactory();
}
}
}
return dbFactory;
}
public int getSessionSize() {
return dbSessionMap.size();
}
public Set<String> getGraphNames() {
return dbSessionMap.keySet();
}
public HugeConfig getHugeConfig() {
return this.hugeConfig;
}
public void setHugeConfig(HugeConfig nodeConfig) {
this.hugeConfig = nodeConfig;
}
public boolean findPathInRemovedList(String path) {
for (DBSessionWatcher pair : destroyGraphDBs) {
if (pair.dbSession.getDbPath().equals(path)) {
return true;
}
}
return false;
}
public RocksDBSession queryGraphDB(String dbName) {
operateLock.readLock().lock();
try {
RocksDBSession session = dbSessionMap.get(dbName);
if (session != null) {
return session.clone();
}
} finally {
operateLock.readLock().unlock();
}
return null;
}
public RocksDBSession createGraphDB(String dbPath, String dbName) {
return createGraphDB(dbPath, dbName, 0);
}
public RocksDBSession createGraphDB(String dbPath, String dbName, long version) {
operateLock.writeLock().lock();
try {
RocksDBSession dbSession = dbSessionMap.get(dbName);
if (dbSession == null) {
log.info("create rocksdb for {}", dbName);
dbSession = new RocksDBSession(this.hugeConfig, dbPath, dbName, version);
dbSessionMap.put(dbName, dbSession);
}
return dbSession.clone();
} finally {
operateLock.writeLock().unlock();
}
}
/**
* @param :
* @return long
* @description the size(KB) of the total rocksdb's data.
*/
public long getTotalSize() {
long kbSize = dbSessionMap.entrySet()
.stream()
.map(e -> e.getValue().getApproximateDataSize())
.reduce(0L, Long::sum);
return kbSize;
}
public Map<String, Long> getTotalKey() {
Map<String, Long> totalKeys = dbSessionMap.entrySet().stream()
.collect(Collectors.toMap(e -> e.getKey(),
e -> e.getValue()
.getEstimateNumKeys()));
return totalKeys;
}
/**
* 释放rocksdb对象
*
* @param dbName
* @return
*/
public boolean releaseGraphDB(String dbName) {
log.info("close {} 's rocksdb.", dbName);
operateLock.writeLock().lock();
try {
RocksDBSession dbSession = dbSessionMap.get(dbName);
if (dbSession != null) {
dbSessionMap.remove(dbName);
rocksdbChangedListeners.forEach(listener -> {
listener.onDBSessionReleased(dbSession);
});
dbSession.close();
}
} finally {
operateLock.writeLock().unlock();
}
return false;
}
/**
* 销毁图，并删除数据文件
*
* @param dbName
*/
public void destroyGraphDB(String dbName) {
log.info("destroy {} 's rocksdb.", dbName);
RocksDBSession dbSession = dbSessionMap.get(dbName);
releaseGraphDB(dbName);
//增加删除标记
if (dbSession != null) {
destroyGraphDBs.add(new DBSessionWatcher(dbSession));
rocksdbChangedListeners.forEach(listener -> {
listener.onDBDeleteBegin(dbSession.getGraphName(), dbSession.getDbPath());
});
}
}
public void releaseAllGraphDB() {
log.info("close all rocksdb.");
operateLock.writeLock().lock();
try {
dbSessionMap.forEach((k, v) -> {
v.shutdown();
});
dbSessionMap.clear();
} finally {
operateLock.writeLock().unlock();
}
}
public Map<MemoryUsageType, Long> getApproximateMemoryUsageByType(List<RocksDB> dbs,
List<Cache> caches) {
if (dbs == null) {
dbs = new ArrayList<>();
} else {
dbs = new ArrayList<>(dbs);
}
List<RocksDBSession> sessions = new ArrayList<>();
for (String dbName : getGraphNames()) {
RocksDBSession session = this.queryGraphDB(dbName);
if (session != null) {
dbs.add(session.getDB());
sessions.add(session);
}
}
try {
HashSet<Cache> allCaches = new HashSet<>();
if (caches != null) {
allCaches.addAll(caches);
}
allCaches.add((Cache) hugeConfig.getProperty(RocksDBOptions.WRITE_CACHE));
allCaches.add((Cache) hugeConfig.getProperty(RocksDBOptions.BLOCK_CACHE));
return MemoryUtil.getApproximateMemoryUsageByType(dbs, allCaches);
} finally {
sessions.forEach(session -> {
session.close();
});
}
}
public void addRocksdbChangedListener(RocksdbChangedListener listener) {
rocksdbChangedListeners.add(listener);
}
public interface RocksdbChangedListener {
default void onCompacted(String dbName) {
}
default void onDBDeleteBegin(String dbName, String filePath) {
}
default void onDBDeleted(String dbName, String filePath) {
}
default void onDBSessionReleased(RocksDBSession dbSession) {
}
}
class RocksdbEventListener extends AbstractEventListener {
@Override
public void onCompactionCompleted(RocksDB db, CompactionJobInfo compactionJobInfo) {
super.onCompactionCompleted(db, compactionJobInfo);
rocksdbChangedListeners.forEach(listener -> {
listener.onCompacted(db.getName());
});
}
@Override
public void onCompactionBegin(final RocksDB db, final CompactionJobInfo compactionJobInfo) {
log.info("RocksdbEventListener onCompactionBegin");
}
}
class DBSessionWatcher {
public RocksDBSession dbSession;
public Long timestamp;
public DBSessionWatcher(RocksDBSession dbSession) {
this.dbSession = dbSession;
timestamp = System.currentTimeMillis();
}
}
}

435
hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBOptions.java Normal file
View File

@ -0,0 +1,435 @@
/*
* 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.hugegraph.rocksdb.access;
import static org.apache.hugegraph.config.OptionChecker.allowValues;
import static org.apache.hugegraph.config.OptionChecker.disallowEmpty;
import static org.apache.hugegraph.config.OptionChecker.inValues;
import static org.apache.hugegraph.config.OptionChecker.rangeDouble;
import static org.apache.hugegraph.config.OptionChecker.rangeInt;
import org.apache.hugegraph.config.ConfigConvOption;
import org.apache.hugegraph.config.ConfigListConvOption;
import org.apache.hugegraph.config.ConfigOption;
import org.apache.hugegraph.config.OptionHolder;
import org.apache.hugegraph.util.Bytes;
import org.rocksdb.CompactionStyle;
import org.rocksdb.CompressionType;
import org.rocksdb.InfoLogLevel;
public class RocksDBOptions extends OptionHolder {
public static final ConfigOption<Long> TOTAL_MEMORY_SIZE =
new ConfigOption<>(
"rocksdb.total_memory_size",
"Limit total memory of memtables for all dbs",
rangeInt(0L, Long.MAX_VALUE),
48L * Bytes.GB
);
public static final ConfigOption<Double> WRITE_BUFFER_RATIO =
new ConfigOption<>(
"rocksdb.write_buffer_ratio",
"write buffer ratio",
rangeDouble(0.0, 1.0),
0.66
);
public static final ConfigOption<Boolean> WRITE_BUFFER_ALLOW_STALL =
new ConfigOption<>(
"rocksdb.write_buffer_allow_stall",
" if set true, it will enable stalling of writes when memory_usage() exceeds " +
"buffer_size." +
" It will wait for flush to complete and memory usage to drop down",
disallowEmpty(),
false
);
// public static final ConfigListOption<String> DATA_DISKS =
// new ConfigListOption<>(
// "rocksdb.data_disks",
// false,
// "The optimized disks for storing data of RocksDB. " +
// "The format of each element: `STORE/TABLE: /path/disk`." +
// "Allowed keys are [g/vertex, g/edge_out, g/edge_in, " +
// "g/vertex_label_index, g/edge_label_index, " +
// "g/range_int_index, g/range_float_index, " +
// "g/range_long_index, g/range_double_index, " +
// "g/secondary_index, g/search_index, g/shard_index, " +
// "g/unique_index, g/olap]",
// null,
// String.class,
// ImmutableList.of()
// );
public static final ConfigOption<String> SST_PATH =
new ConfigOption<>(
"rocksdb.sst_path",
"The path for ingesting SST file into RocksDB.",
null,
""
);
public static final ConfigOption<InfoLogLevel> LOG_LEVEL =
new ConfigOption<>(
"rocksdb.log_level",
"The info log level of RocksDB.",
allowValues(InfoLogLevel.DEBUG_LEVEL,
InfoLogLevel.INFO_LEVEL,
InfoLogLevel.WARN_LEVEL,
InfoLogLevel.ERROR_LEVEL,
InfoLogLevel.FATAL_LEVEL,
InfoLogLevel.HEADER_LEVEL),
InfoLogLevel.INFO_LEVEL
);
public static final ConfigOption<Integer> NUM_LEVELS =
new ConfigOption<>(
"rocksdb.num_levels",
"Set the number of levels for this database.",
rangeInt(1, Integer.MAX_VALUE),
7
);
public static final ConfigOption<Long> BLOCK_CACHE_CAPACITY =
new ConfigOption<>(
"rocksdb.block_cache_capacity",
"The amount of block cache in bytes that will be used by all RocksDBs",
rangeInt(0L, Long.MAX_VALUE),
16L * Bytes.GB
);
public static final ConfigOption<String> SNAPSHOT_PATH =
new ConfigOption<>(
"rocksdb.snapshot_path",
"The path for storing snapshot of RocksDB.",
disallowEmpty(),
"rocksdb-snapshot"
);
public static final ConfigConvOption<String, CompactionStyle> COMPACTION_STYLE =
new ConfigConvOption<>(
"rocksdb.compaction_style",
"Set compaction style for RocksDB: LEVEL/UNIVERSAL/FIFO.",
allowValues("LEVEL", "UNIVERSAL", "FIFO"),
CompactionStyle::valueOf,
"LEVEL"
);
public static final ConfigOption<Boolean> OPTIMIZE_MODE =
new ConfigOption<>(
"rocksdb.optimize_mode",
"Optimize for heavy workloads and big datasets.",
disallowEmpty(),
true
);
public static final ConfigListConvOption<String, CompressionType> LEVELS_COMPRESSIONS =
new ConfigListConvOption<>(
"rocksdb.compression_per_level",
"The compression algorithms for different levels of RocksDB, " +
"allowed values are none/snappy/z/bzip2/lz4/lz4hc/xpress/zstd.",
inValues("none", "snappy", "z", "bzip2", "lz4", "lz4hc", "xpress", "zstd"),
CompressionType::getCompressionType,
"none", "none", "snappy", "snappy", "snappy", "snappy", "snappy"
);
public static final ConfigConvOption<String, CompressionType> BOTTOMMOST_COMPRESSION =
new ConfigConvOption<>(
"rocksdb.bottommost_compression",
"The compression algorithm for the bottommost level of RocksDB, " +
"allowed values are none/snappy/z/bzip2/lz4/lz4hc/xpress/zstd.",
allowValues("none", "snappy", "z", "bzip2", "lz4", "lz4hc", "xpress", "zstd"),
CompressionType::getCompressionType,
"none"
);
public static final ConfigConvOption<String, CompressionType> COMPRESSION =
new ConfigConvOption<>(
"rocksdb.compression",
"The compression algorithm for compressing blocks of RocksDB, " +
"allowed values are none/snappy/z/bzip2/lz4/lz4hc/xpress/zstd.",
allowValues("none", "snappy", "z", "bzip2", "lz4", "lz4hc", "xpress", "zstd"),
CompressionType::getCompressionType,
"snappy"
);
public static final ConfigOption<Integer> MAX_BG_JOBS =
new ConfigOption<>(
"rocksdb.max_background_jobs",
"Maximum number of concurrent background jobs, including flushes and " +
"compactions.",
rangeInt(1, Integer.MAX_VALUE),
8
);
public static final ConfigOption<Integer> MAX_SUB_COMPACTIONS =
new ConfigOption<>(
"rocksdb.max_subcompactions",
"The value represents the maximum number of threads per compaction job.",
rangeInt(1, Integer.MAX_VALUE),
4
);
public static final ConfigOption<Long> DELAYED_WRITE_RATE =
new ConfigOption<>(
"rocksdb.delayed_write_rate",
"The rate limit in bytes/s of user write requests " +
"when need to slow down if the compaction gets behind.",
rangeInt(1L, Long.MAX_VALUE),
64L * Bytes.MB
);
public static final ConfigOption<Integer> MAX_OPEN_FILES =
new ConfigOption<>(
"rocksdb.max_open_files",
"The maximum number of open files that can be cached by RocksDB, " +
"-1 means no limit.",
rangeInt(-1, Integer.MAX_VALUE),
-1
);
public static final ConfigOption<Long> MAX_MANIFEST_FILE_SIZE =
new ConfigOption<>(
"rocksdb.max_manifest_file_size",
"The max size of manifest file in bytes.",
rangeInt(1L, Long.MAX_VALUE),
100L * Bytes.MB
);
public static final ConfigOption<Boolean> SKIP_STATS_UPDATE_ON_DB_OPEN =
new ConfigOption<>(
"rocksdb.skip_stats_update_on_db_open",
"Whether to skip statistics update when opening the database, " +
"setting this flag true allows us to not update statistics.",
disallowEmpty(),
false
);
public static final ConfigOption<Integer> MAX_FILE_OPENING_THREADS =
new ConfigOption<>(
"rocksdb.max_file_opening_threads",
"The max number of threads used to open files.",
rangeInt(1, Integer.MAX_VALUE),
16
);
public static final ConfigOption<Long> MAX_TOTAL_WAL_SIZE =
new ConfigOption<>(
"rocksdb.max_total_wal_size",
"Total size of WAL files in bytes. Once WALs exceed this size, " +
"we will start forcing the flush of column families related, " +
"0 means no limit.",
rangeInt(0L, Long.MAX_VALUE),
0L
);
public static final ConfigOption<Long> DB_MEMTABLE_SIZE =
new ConfigOption<>(
"rocksdb.db_write_buffer_size",
"Total size of write buffers in bytes across all column families, " +
"0 means no limit.",
rangeInt(0L, Long.MAX_VALUE),
0L
);
public static final ConfigOption<Long> DELETE_OBSOLETE_FILE_PERIOD =
new ConfigOption<>(
"rocksdb.delete_obsolete_files_period",
"The periodicity in seconds when obsolete files get deleted, " +
"0 means always do full purge.",
rangeInt(0L, Long.MAX_VALUE),
6L * 60 * 60
);
public static final ConfigOption<Long> MEMTABLE_SIZE =
new ConfigOption<>(
"rocksdb.write_buffer_size",
"Amount of data in bytes to build up in memory.",
rangeInt(Bytes.MB, Long.MAX_VALUE),
32L * Bytes.MB
);
public static final ConfigOption<Integer> MAX_MEMTABLES =
new ConfigOption<>(
"rocksdb.max_write_buffer_number",
"The maximum number of write buffers that are built up in memory.",
rangeInt(1, Integer.MAX_VALUE),
32
);
public static final ConfigOption<Integer> MIN_MEMTABLES_TO_MERGE =
new ConfigOption<>(
"rocksdb.min_write_buffer_number_to_merge",
"The minimum number of write buffers that will be merged together.",
rangeInt(1, Integer.MAX_VALUE),
16
);
public static final ConfigOption<Integer> MAX_MEMTABLES_TO_MAINTAIN =
new ConfigOption<>(
"rocksdb.max_write_buffer_number_to_maintain",
"The total maximum number of write buffers to maintain in memory.",
rangeInt(0, Integer.MAX_VALUE),
0
);
public static final ConfigOption<Boolean> DYNAMIC_LEVEL_BYTES =
new ConfigOption<>(
"rocksdb.level_compaction_dynamic_level_bytes",
"Whether to enable level_compaction_dynamic_level_bytes, " +
"if it's enabled we give max_bytes_for_level_multiplier a " +
"priority against max_bytes_for_level_base, the bytes of " +
"base level is dynamic for a more predictable LSM tree, " +
"it is useful to limit worse case space amplification. " +
"Turning this feature on/off for an existing DB can cause " +
"unexpected LSM tree structure so it's not recommended.",
disallowEmpty(),
false
);
public static final ConfigOption<Long> MAX_LEVEL1_BYTES =
new ConfigOption<>(
"rocksdb.max_bytes_for_level_base",
"The upper-bound of the total size of level-1 files in bytes.",
rangeInt(Bytes.MB, Long.MAX_VALUE),
10L * Bytes.GB
);
public static final ConfigOption<Double> MAX_LEVEL_BYTES_MULTIPLIER =
new ConfigOption<>(
"rocksdb.max_bytes_for_level_multiplier",
"The ratio between the total size of level (L+1) files and " +
"the total size of level L files for all L.",
rangeDouble(1.0, Double.MAX_VALUE),
10.0
);
public static final ConfigOption<Long> TARGET_FILE_SIZE_BASE =
new ConfigOption<>(
"rocksdb.target_file_size_base",
"The target file size for compaction in bytes.",
rangeInt(Bytes.MB, Long.MAX_VALUE),
256L * Bytes.MB
);
public static final ConfigOption<Integer> TARGET_FILE_SIZE_MULTIPLIER =
new ConfigOption<>(
"rocksdb.target_file_size_multiplier",
"The size ratio between a level L file and a level (L+1) file.",
rangeInt(1, Integer.MAX_VALUE),
2
);
public static final ConfigOption<Integer> LEVEL0_COMPACTION_TRIGGER =
new ConfigOption<>(
"rocksdb.level0_file_num_compaction_trigger",
"Number of files to trigger level-0 compaction.",
rangeInt(0, Integer.MAX_VALUE),
10
);
public static final ConfigOption<Integer> LEVEL0_SLOWDOWN_WRITES_TRIGGER =
new ConfigOption<>(
"rocksdb.level0_slowdown_writes_trigger",
"Soft limit on number of level-0 files for slowing down writes.",
rangeInt(-1, Integer.MAX_VALUE),
256
);
public static final ConfigOption<Integer> LEVEL0_STOP_WRITES_TRIGGER =
new ConfigOption<>(
"rocksdb.level0_stop_writes_trigger",
"Hard limit on number of level-0 files for stopping writes.",
rangeInt(-1, Integer.MAX_VALUE),
1024
);
public static final ConfigOption<Long> SOFT_PENDING_COMPACTION_LIMIT =
new ConfigOption<>(
"rocksdb.soft_pending_compaction_bytes_limit",
"The soft limit to impose on pending compaction in bytes.",
rangeInt(Bytes.GB, Long.MAX_VALUE),
1024L * Bytes.GB
);
public static final ConfigOption<Long> HARD_PENDING_COMPACTION_LIMIT =
new ConfigOption<>(
"rocksdb.hard_pending_compaction_bytes_limit",
"The hard limit to impose on pending compaction in bytes.",
rangeInt(Bytes.GB, Long.MAX_VALUE),
2048L * Bytes.GB
);
public static final ConfigOption<Boolean> ALLOW_MMAP_WRITES =
new ConfigOption<>(
"rocksdb.allow_mmap_writes",
"Allow the OS to mmap file for writing.",
disallowEmpty(),
false
);
public static final ConfigOption<Boolean> ALLOW_MMAP_READS =
new ConfigOption<>(
"rocksdb.allow_mmap_reads",
"Allow the OS to mmap file for reading sst tables.",
disallowEmpty(),
false
);
public static final ConfigOption<Boolean> USE_DIRECT_READS =
new ConfigOption<>(
"rocksdb.use_direct_reads",
"Enable the OS to use direct I/O for reading sst tables.",
disallowEmpty(),
false
);
public static final ConfigOption<Boolean> USE_DIRECT_READS_WRITES_FC =
new ConfigOption<>(
"rocksdb.use_direct_io_for_flush_and_compaction",
"Enable the OS to use direct read/writes in flush and compaction.",
disallowEmpty(),
false
);
public static final ConfigOption<Boolean> PIN_L0_FILTER_AND_INDEX_IN_CACHE =
new ConfigOption<>(
"rocksdb.pin_l0_filter_and_index_blocks_in_cache",
"Indicating if we'd put index/filter blocks to the block cache.",
disallowEmpty(),
true
);
public static final ConfigOption<Boolean> PUT_FILTER_AND_INDEX_IN_CACHE =
new ConfigOption<>(
"rocksdb.cache_index_and_filter_blocks",
"Indicating if we'd put index/filter blocks to the block cache.",
disallowEmpty(),
true
);
public static final ConfigOption<Integer> BLOOM_FILTER_BITS_PER_KEY =
new ConfigOption<>(
"rocksdb.bloom_filter_bits_per_key",
"The bits per key in bloom filter, a good value is 10, " +
"which yields a filter with ~ 1% false positive rate, " +
"-1 means no bloom filter.",
rangeInt(-1, Integer.MAX_VALUE),
-1
);
public static final ConfigOption<Boolean> BLOOM_FILTER_MODE =
new ConfigOption<>(
"rocksdb.bloom_filter_block_based_mode",
"Use block based filter rather than full filter.",
disallowEmpty(),
false
);
public static final ConfigOption<Boolean> BLOOM_FILTER_WHOLE_KEY =
new ConfigOption<>(
"rocksdb.bloom_filter_whole_key_filtering",
"True if place whole keys in the bloom filter, " +
"else place the prefix of keys.",
disallowEmpty(),
true
);
public static final ConfigOption<Boolean> BLOOM_FILTERS_SKIP_LAST_LEVEL =
new ConfigOption<>(
"rocksdb.optimize_filters_for_hits",
"This flag allows us to not store filters for the last level.",
disallowEmpty(),
false
);
public static final String BLOCK_TABLE_CONFIG = "rocksdb.block_table_config";
public static final String WRITE_BUFFER_MANAGER = "rocksdb.write_buffer_manager";
public static final String BLOCK_CACHE = "rocksdb.block_cache";
public static final String WRITE_CACHE = "rocksdb.write_cache";
public static final String ENV = "rocksdb.env";
private static volatile RocksDBOptions instance;
private RocksDBOptions() {
super();
}
public static synchronized RocksDBOptions instance() {
if (instance == null) {
instance = new RocksDBOptions();
instance.registerOptions();
}
return instance;
}
}

237
hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBScanIterator.java Normal file
View File

@ -0,0 +1,237 @@
/*
* 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.hugegraph.rocksdb.access;
import java.util.Arrays;
import java.util.NoSuchElementException;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.hugegraph.rocksdb.access.RocksDBSession.BackendColumn;
import org.apache.hugegraph.util.Bytes;
import org.apache.hugegraph.util.E;
import org.rocksdb.RocksIterator;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class RocksDBScanIterator<T> implements ScanIterator {
private static final byte[] EMPTY_VALUE = new byte[0];
private final RocksIterator rawIt;
private final byte[] keyBegin;
private final byte[] keyEnd;
private final int scanType;
private final AtomicBoolean closed = new AtomicBoolean(false);
private final RocksDBSession.RefCounter iterReference;
private byte[] key;
private boolean matched;
public RocksDBScanIterator(RocksIterator rawIt, byte[] keyBegin, byte[] keyEnd,
int scanType, RocksDBSession.RefCounter iterReference) {
this.rawIt = rawIt;
this.keyBegin = keyBegin;
this.keyEnd = keyEnd;
this.scanType = scanType;
this.key = keyBegin;
this.matched = false;
this.iterReference = iterReference;
this.seek();
}
private void checkArguments() {
E.checkArgument(!(this.match(ScanIterator.Trait.SCAN_PREFIX_BEGIN) &&
this.match(ScanIterator.Trait.SCAN_PREFIX_END)),
"Can't set SCAN_PREFIX_WITH_BEGIN and " +
"SCAN_PREFIX_WITH_END at the same time");
E.checkArgument(!(this.match(ScanIterator.Trait.SCAN_PREFIX_BEGIN) &&
this.match(ScanIterator.Trait.SCAN_GT_BEGIN)),
"Can't set SCAN_PREFIX_WITH_BEGIN and " +
"SCAN_GT_BEGIN/SCAN_GTE_BEGIN at the same time");
E.checkArgument(!(this.match(ScanIterator.Trait.SCAN_PREFIX_END) &&
this.match(ScanIterator.Trait.SCAN_LT_END)),
"Can't set SCAN_PREFIX_WITH_END and " +
"SCAN_LT_END/SCAN_LTE_END at the same time");
if (this.match(ScanIterator.Trait.SCAN_PREFIX_BEGIN)) {
E.checkArgument(this.keyBegin != null,
"Parameter `keyBegin` can't be null " +
"if set SCAN_PREFIX_WITH_BEGIN");
E.checkArgument(this.keyEnd == null,
"Parameter `keyEnd` must be null " +
"if set SCAN_PREFIX_WITH_BEGIN");
}
if (this.match(ScanIterator.Trait.SCAN_PREFIX_END)) {
E.checkArgument(this.keyEnd != null,
"Parameter `keyEnd` can't be null " +
"if set SCAN_PREFIX_WITH_END");
}
if (this.match(ScanIterator.Trait.SCAN_GT_BEGIN)) {
E.checkArgument(this.keyBegin != null,
"Parameter `keyBegin` can't be null " +
"if set SCAN_GT_BEGIN or SCAN_GTE_BEGIN");
}
if (this.match(ScanIterator.Trait.SCAN_LT_END)) {
E.checkArgument(this.keyEnd != null,
"Parameter `keyEnd` can't be null " +
"if set SCAN_LT_END or SCAN_LTE_END");
}
}
@Override
public boolean hasNext() {
if (this.closed.get()) {
//log.warn("Iterator has been closed");
return false;
}
this.matched = this.rawIt.isOwningHandle();
if (!this.matched) {
// Maybe closed
return this.matched;
}
this.matched = this.rawIt.isValid();
if (this.matched) {
// Update position for paging
this.key = this.rawIt.key();
this.matched = this.filter(this.key);
}
if (!this.matched) {
// The end
this.key = null;
// Free the iterator if finished
this.close();
}
return this.matched;
}
private void seek() {
if (this.closed.get()) {
log.warn("Iterator has been closed");
return;
}
if (this.keyBegin == null) {
// Seek to the first if no `keyBegin`
this.rawIt.seekToFirst();
} else {
/*
* Seek to `keyBegin`:
* if set SCAN_GT_BEGIN/SCAN_GTE_BEGIN (key > / >= 'xx')
* or if set SCAN_PREFIX_WITH_BEGIN (key prefix with 'xx')
*/
this.rawIt.seek(this.keyBegin);
// Skip `keyBegin` if set SCAN_GT_BEGIN (key > 'xx')
if (this.match(ScanIterator.Trait.SCAN_GT_BEGIN) &&
!this.match(ScanIterator.Trait.SCAN_GTE_BEGIN)) {
while (this.rawIt.isValid() &&
Bytes.equals(this.rawIt.key(), this.keyBegin)) {
this.rawIt.next();
}
}
}
}
@Override
public boolean isValid() {
return this.rawIt.isValid();
}
@Override
public BackendColumn next() {
if (this.closed.get()) {
log.warn("Iterator has been closed");
throw new NoSuchElementException();
}
if (!this.matched) {
if (!this.hasNext()) {
throw new NoSuchElementException();
}
}
BackendColumn col = BackendColumn.of(this.key,
this.match(Trait.SCAN_KEYONLY) ? EMPTY_VALUE :
this.rawIt.value());
this.rawIt.next();
this.matched = false;
return col;
}
@Override
public long count() {
long count = 0L;
while (this.hasNext()) {
this.rawIt.next();
count++;
}
return count;
}
@Override
public byte[] position() {
return this.key;
}
private boolean filter(byte[] v) {
if (this.match(ScanIterator.Trait.SCAN_PREFIX_BEGIN)) {
return Bytes.prefixWith(v, this.keyBegin);
} else if (this.match(ScanIterator.Trait.SCAN_PREFIX_END)) {
assert this.keyEnd != null;
return Bytes.prefixWith(v, this.keyEnd);
} else if (this.match(ScanIterator.Trait.SCAN_LT_END)) {
assert this.keyEnd != null;
if (this.match(ScanIterator.Trait.SCAN_LTE_END)) {
v = Arrays.copyOfRange(v, 0, this.keyEnd.length);
return Bytes.compare(v, this.keyEnd) <= 0;
} else {
return Bytes.compare(v, this.keyEnd) < 0;
}
} else {
return true;
}
}
@Override
public void close() {
if (!this.closed.getAndSet(true)) {
if (this.rawIt.isOwningHandle()) {
this.rawIt.close();
}
this.iterReference.release();
}
}
private boolean match(int expected) {
return (expected & this.scanType) == expected;
}
}

1054
hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java Normal file
View File

File diff suppressed because it is too large Load Diff

56
hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/ScanIterator.java Normal file
View File

@ -0,0 +1,56 @@
/*
* 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.hugegraph.rocksdb.access;
import java.io.Closeable;
public interface ScanIterator extends Closeable {
boolean hasNext();
boolean isValid();
<T> T next();
default long count() {
return 0;
}
default byte[] position() {
return new byte[0];
}
default void seek(byte[] position) {
}
@Override
void close();
abstract class Trait {
public static final int SCAN_ANY = 0x80;
public static final int SCAN_PREFIX_BEGIN = 0x01;
public static final int SCAN_PREFIX_END = 0x02;
public static final int SCAN_GT_BEGIN = 0x04;
public static final int SCAN_GTE_BEGIN = 0x0c;
public static final int SCAN_LT_END = 0x10;
public static final int SCAN_LTE_END = 0x30;
public static final int SCAN_KEYONLY = 0x40;
public static final int SCAN_HASHCODE = 0x100;
}
}

78
hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/SessionOperator.java Normal file
View File

@ -0,0 +1,78 @@
/*
* 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.hugegraph.rocksdb.access;
import org.apache.hugegraph.store.term.HgPair;
public interface SessionOperator {
HgPair<byte[], byte[]> keyRange(String table);
void compactRange(String table) throws DBStoreException;
void compactRange() throws DBStoreException;
void put(String table, byte[] key, byte[] value) throws DBStoreException;
ScanIterator scan(String tableName);
ScanIterator scan(String tableName, byte[] prefix);
ScanIterator scan(String tableName, byte[] prefix, int scanType);
ScanIterator scan(String tableName, byte[] keyFrom, byte[] keyTo, int scanType);
/**
* 扫描所有cf指定范围的数据
*/
ScanIterator scanRaw(byte[] keyFrom, byte[] keyTo, long startSeqNum);
long keyCount(byte[] start, byte[] end, String tableName);
long estimatedKeyCount(String tableName);
/*
* only support 'long data' operator
* */
void merge(String table, byte[] key, byte[] value) throws DBStoreException;
void increase(String table, byte[] key, byte[] value) throws DBStoreException;
void delete(String table, byte[] key) throws DBStoreException;
void deleteSingle(String table, byte[] key) throws DBStoreException;
void deletePrefix(String table, byte[] key) throws DBStoreException;
void deleteRange(String table, byte[] keyFrom, byte[] keyTo) throws DBStoreException;
/**
* 删除所有cf指定范围的数据
*/
void deleteRange(byte[] keyFrom, byte[] keyTo) throws DBStoreException;
byte[] get(String table, byte[] key) throws DBStoreException;
void prepare();
Integer commit() throws DBStoreException;
void rollback();
RocksDBSession getDBSession();
}

419
hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/SessionOperatorImpl.java Normal file
View File

@ -0,0 +1,419 @@
/*
* 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.hugegraph.rocksdb.access;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Iterator;
import org.apache.hugegraph.rocksdb.access.RocksDBSession.CFHandleLock;
import org.apache.hugegraph.rocksdb.access.util.Asserts;
import org.apache.hugegraph.store.term.HgPair;
import org.apache.hugegraph.util.Bytes;
import org.rocksdb.ReadOptions;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;
import org.rocksdb.Slice;
import org.rocksdb.Snapshot;
import org.rocksdb.WriteBatch;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class SessionOperatorImpl implements SessionOperator {
private final RocksDB db;
private final RocksDBSession session;
private WriteBatch batch;
public SessionOperatorImpl(RocksDBSession session) {
this.session = session;
this.db = session.getDB();
}
public static final byte[] increaseOne(byte[] bytes) {
final byte BYTE_MAX_VALUE = (byte) 0xff;
assert bytes.length > 0;
byte last = bytes[bytes.length - 1];
if (last != BYTE_MAX_VALUE) {
bytes[bytes.length - 1] += (byte) 0x01;
} else {
// Process overflow (like [1, 255] => [2, 0])
int i = bytes.length - 1;
for (; i > 0 && bytes[i] == BYTE_MAX_VALUE; --i) {
bytes[i] += (byte) 0x01;
}
if (bytes[i] == BYTE_MAX_VALUE) {
assert i == 0;
throw new DBStoreException("Unable to increase bytes: %s", Bytes.toHex(bytes));
}
bytes[i] += (byte) 0x01;
}
return bytes;
}
public RocksDB rocksdb() {
return db;
}
@Override
public RocksDBSession getDBSession() {
return session;
}
private CFHandleLock getLock(String table) {
CFHandleLock cf = this.session.getCFHandleLock(table);
return cf;
}
@Override
public HgPair<byte[], byte[]> keyRange(String table) {
byte[] startKey, endKey;
try (CFHandleLock handle = this.getLock(table);
RocksIterator iter = rocksdb().newIterator(handle.get())) {
iter.seekToFirst();
if (!iter.isValid()) {
return null;
}
startKey = iter.key();
iter.seekToLast();
if (!iter.isValid()) {
return new HgPair<>(startKey, null);
}
endKey = iter.key();
}
return new HgPair<>(startKey, endKey);
}
@Override
public void compactRange(String table) throws DBStoreException {
try (CFHandleLock handle = this.getLock(table)) {
rocksdb().compactRange(handle.get());
} catch (RocksDBException e) {
throw new DBStoreException(e);
}
}
@Override
public void compactRange() throws DBStoreException {
for (String name : session.getTables().keySet()) {
compactRange(name);
}
}
@Override
public void put(String table, byte[] key, byte[] value) throws DBStoreException {
try {
this.getBatch().put(session.getCF(table), key, value);
} catch (RocksDBException e) {
throw new DBStoreException(e);
}
}
/*
* only support 'long data' operator
*/
@Override
public void merge(String table, byte[] key, byte[] value) throws DBStoreException {
try {
this.getBatch().merge(session.getCF(table), key, value);
} catch (RocksDBException e) {
throw new DBStoreException(e);
}
}
@Override
public void increase(String table, byte[] key, byte[] value) throws DBStoreException {
try (CFHandleLock cf = this.getLock(table)) {
rocksdb().merge(cf.get(), key, value);
} catch (RocksDBException e) {
throw new DBStoreException(e);
}
}
@Override
public void delete(String table, byte[] key) throws DBStoreException {
try {
this.getBatch().delete(session.getCF(table), key);
} catch (RocksDBException e) {
throw new DBStoreException(e);
}
}
@Override
public void deleteSingle(String table, byte[] key) throws DBStoreException {
try {
this.getBatch().singleDelete(session.getCF(table), key);
} catch (RocksDBException e) {
throw new DBStoreException(e);
}
}
@Override
public void deletePrefix(String table, byte[] key) throws DBStoreException {
byte[] keyFrom = key;
byte[] keyTo = Arrays.copyOf(key, key.length);
keyTo = increaseOne(keyTo);
try {
this.getBatch().deleteRange(session.getCF(table), keyFrom, keyTo);
} catch (RocksDBException e) {
throw new DBStoreException(e);
}
}
@Override
public void deleteRange(String table, byte[] keyFrom, byte[] keyTo) throws DBStoreException {
Asserts.isTrue(keyFrom != null, "KeyFrom is null");
Asserts.isTrue(keyTo != null, "KeyTo is null");
if (Bytes.compare(keyTo, keyFrom) < 0) {
throw new DBStoreException("[end key: %s ] is lower than [start key: %s]",
Arrays.toString(keyTo), Arrays.toString(keyFrom));
}
try {
this.prepare();
this.getBatch().deleteRange(session.getCF(table), keyFrom, keyTo);
this.commit();
} catch (RocksDBException e) {
this.rollback();
throw new DBStoreException(e);
}
}
@Override
public void deleteRange(byte[] keyFrom, byte[] keyTo) throws DBStoreException {
for (String name : session.getTables().keySet()) {
deleteRange(name, keyFrom, keyTo);
}
}
@Override
public byte[] get(String table, byte[] key) throws DBStoreException {
try (CFHandleLock cf = this.getLock(table)) {
return rocksdb().get(cf.get(), key);
} catch (RocksDBException e) {
throw new DBStoreException(e);
}
}
@Override
public void prepare() {
this.session.getCfHandleReadLock().lock();
}
/**
* commit抛出异常后一定要调用rollback，否则会造成cfHandleReadLock未释放
*/
@Override
public Integer commit() throws DBStoreException {
int count = this.getBatch().count();
if (count > 0) {
try {
rocksdb().write(session.getWriteOptions(), this.batch);
} catch (RocksDBException e) {
throw new DBStoreException(e);
}
// Clear batch if write() successfully (retained if failed)
this.batch.clear();
this.batch.close();
this.batch = null;
}
this.session.getCfHandleReadLock().unlock();
return count;
}
@Override
public void rollback() {
try {
if (this.batch != null) {
this.batch.clear();
this.batch.close();
}
this.batch = null;
} finally {
try {
this.session.getCfHandleReadLock().unlock();
} catch (Exception e) {
log.error("rollback {}", e);
}
}
}
@Override
public ScanIterator scan(String tableName) {
try (CFHandleLock handle = this.getLock(tableName)) {
if (handle == null) {
log.info("no find table : {}", tableName);
return null;
}
return new RocksDBScanIterator(this.rocksdb().newIterator(handle.get()), null, null,
ScanIterator.Trait.SCAN_ANY,
this.session.getRefCounter());
}
}
@Override
public ScanIterator scan(String tableName, byte[] prefix) {
return scan(tableName, prefix, 0);
}
@Override
public ScanIterator scan(String tableName, byte[] prefix, int scanType) {
try (CFHandleLock handle = this.getLock(tableName)) {
if (handle == null) {
log.info("no find table: {} for scanning with prefix: {}", tableName,
new String(prefix));
return null;
}
return new RocksDBScanIterator(this.rocksdb().newIterator(handle.get()), prefix, null,
ScanIterator.Trait.SCAN_PREFIX_BEGIN | scanType,
this.session.getRefCounter());
}
}
@Override
public ScanIterator scan(String tableName, byte[] keyFrom, byte[] keyTo, int scanType) {
try (CFHandleLock handle = this.getLock(tableName)) {
if (handle == null) {
log.info("no find table: {} for scantype: {}", tableName, scanType);
return null;
}
return new RocksDBScanIterator(this.rocksdb().newIterator(handle.get()), keyFrom, keyTo,
scanType,
this.session.getRefCounter());
}
}
/**
* 遍历所有cf指定范围的数据
* TODO: rocksdb7.x 不支持 setStartSeqNum，改为使用 Timestamp
* refer: https://github.com/facebook/rocksdb/wiki/User-defined-Timestamp
*/
@Override
public ScanIterator scanRaw(byte[] keyFrom, byte[] keyTo, long startSeqNum) {
int kNumInternalBytes = 8; //internal key 增加的8个字节后缀
Snapshot snapshot = rocksdb().getSnapshot();
Iterator<String> cfIterator = session.getTables().keySet().iterator();
return new ScanIterator() {
String cfName = null;
@Override
public boolean hasNext() {
return cfIterator.hasNext();
}
@Override
public boolean isValid() {
return cfIterator.hasNext();
}
@Override
public <T> T next() {
RocksIterator iterator = null;
ReadOptions readOptions = new ReadOptions()
.setSnapshot(snapshot);
if (keyFrom != null) {
readOptions.setIterateLowerBound(new Slice(keyFrom));
}
if (keyTo != null) {
readOptions.setIterateUpperBound(new Slice(keyTo));
}
while (iterator == null && cfIterator.hasNext()) {
cfName = cfIterator.next();
try (CFHandleLock handle = getLock(cfName)) {
iterator = rocksdb().newIterator(handle.get(), readOptions);
iterator.seekToFirst();
}
}
if (iterator == null) {
return null;
}
RocksIterator finalIterator = iterator;
return (T) new ScanIterator() {
private final ReadOptions holdReadOptions = readOptions;
@Override
public boolean hasNext() {
return finalIterator.isValid();
}
@Override
public boolean isValid() {
return finalIterator.isValid();
}
@Override
public <T> T next() {
byte[] key = finalIterator.key();
if (startSeqNum > 0) {
key = Arrays.copyOfRange(key, 0, key.length - kNumInternalBytes);
}
RocksDBSession.BackendColumn col =
RocksDBSession.BackendColumn.of(key, finalIterator.value());
finalIterator.next();
return (T) col;
}
@Override
public void close() {
finalIterator.close();
holdReadOptions.close();
}
};
}
@Override
public void close() {
rocksdb().releaseSnapshot(snapshot);
}
@Override
public byte[] position() {
return cfName.getBytes(StandardCharsets.UTF_8);
}
};
}
@Override
public long keyCount(byte[] start, byte[] end, String tableName) {
ScanIterator it = scan(tableName, start, end, ScanIterator.Trait.SCAN_LT_END);
return it.count();
}
@Override
public long estimatedKeyCount(String tableName) throws DBStoreException {
try {
return this.rocksdb()
.getLongProperty(session.getCF(tableName), "rocksdb.estimate-num-keys");
} catch (RocksDBException e) {
throw new DBStoreException(e);
}
}
private WriteBatch getBatch() {
if (this.batch == null) {
this.batch = new WriteBatch();
}
return this.batch;
}
}

42
hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/util/Asserts.java Normal file
View File

@ -0,0 +1,42 @@
/*
* 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.hugegraph.rocksdb.access.util;
import java.util.function.Supplier;
public final class Asserts {
public static void isTrue(boolean expression, String message) {
if (message == null) {
throw new IllegalArgumentException("message is null");
}
if (!expression) {
throw new IllegalArgumentException(message);
}
}
public static void isTrue(boolean expression, Supplier<RuntimeException> s) {
if (s == null) {
throw new IllegalArgumentException("Supplier<RuntimeException> is null");
}
if (!expression) {
throw s.get();
}
}
}

188
hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/util/CRC64.java Normal file
View File

@ -0,0 +1,188 @@
/*
* 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.hugegraph.rocksdb.access.util;
import java.util.zip.Checksum;
public class CRC64 implements Checksum {
private static final long[] CRC_TABLE = new long[]{0x0000000000000000L, 0x42F0E1EBA9EA3693L,
0x85E1C3D753D46D26L, 0xC711223CFA3E5BB5L,
0x493366450E42ECDFL, 0x0BC387AEA7A8DA4CL,
0xCCD2A5925D9681F9L, 0x8E224479F47CB76AL,
0x9266CC8A1C85D9BEL, 0xD0962D61B56FEF2DL,
0x17870F5D4F51B498L, 0x5577EEB6E6BB820BL,
0xDB55AACF12C73561L, 0x99A54B24BB2D03F2L,
0x5EB4691841135847L, 0x1C4488F3E8F96ED4L,
0x663D78FF90E185EFL, 0x24CD9914390BB37CL,
0xE3DCBB28C335E8C9L, 0xA12C5AC36ADFDE5AL,
0x2F0E1EBA9EA36930L, 0x6DFEFF5137495FA3L,
0xAAEFDD6DCD770416L, 0xE81F3C86649D3285L,
0xF45BB4758C645C51L, 0xB6AB559E258E6AC2L,
0x71BA77A2DFB03177L, 0x334A9649765A07E4L,
0xBD68D2308226B08EL, 0xFF9833DB2BCC861DL,
0x388911E7D1F2DDA8L, 0x7A79F00C7818EB3BL,
0xCC7AF1FF21C30BDEL, 0x8E8A101488293D4DL,
0x499B3228721766F8L, 0x0B6BD3C3DBFD506BL,
0x854997BA2F81E701L, 0xC7B97651866BD192L,
0x00A8546D7C558A27L, 0x4258B586D5BFBCB4L,
0x5E1C3D753D46D260L, 0x1CECDC9E94ACE4F3L,
0xDBFDFEA26E92BF46L, 0x990D1F49C77889D5L,
0x172F5B3033043EBFL, 0x55DFBADB9AEE082CL,
0x92CE98E760D05399L, 0xD03E790CC93A650AL,
0xAA478900B1228E31L, 0xE8B768EB18C8B8A2L,
0x2FA64AD7E2F6E317L, 0x6D56AB3C4B1CD584L,
0xE374EF45BF6062EEL, 0xA1840EAE168A547DL,
0x66952C92ECB40FC8L, 0x2465CD79455E395BL,
0x3821458AADA7578FL, 0x7AD1A461044D611CL,
0xBDC0865DFE733AA9L, 0xFF3067B657990C3AL,
0x711223CFA3E5BB50L, 0x33E2C2240A0F8DC3L,
0xF4F3E018F031D676L, 0xB60301F359DBE0E5L,
0xDA050215EA6C212FL, 0x98F5E3FE438617BCL,
0x5FE4C1C2B9B84C09L, 0x1D14202910527A9AL,
0x93366450E42ECDF0L, 0xD1C685BB4DC4FB63L,
0x16D7A787B7FAA0D6L, 0x5427466C1E109645L,
0x4863CE9FF6E9F891L, 0x0A932F745F03CE02L,
0xCD820D48A53D95B7L, 0x8F72ECA30CD7A324L,
0x0150A8DAF8AB144EL, 0x43A04931514122DDL,
0x84B16B0DAB7F7968L, 0xC6418AE602954FFBL,
0xBC387AEA7A8DA4C0L, 0xFEC89B01D3679253L,
0x39D9B93D2959C9E6L, 0x7B2958D680B3FF75L,
0xF50B1CAF74CF481FL, 0xB7FBFD44DD257E8CL,
0x70EADF78271B2539L, 0x321A3E938EF113AAL,
0x2E5EB66066087D7EL, 0x6CAE578BCFE24BEDL,
0xABBF75B735DC1058L, 0xE94F945C9C3626CBL,
0x676DD025684A91A1L, 0x259D31CEC1A0A732L,
0xE28C13F23B9EFC87L, 0xA07CF2199274CA14L,
0x167FF3EACBAF2AF1L, 0x548F120162451C62L,
0x939E303D987B47D7L, 0xD16ED1D631917144L,
0x5F4C95AFC5EDC62EL, 0x1DBC74446C07F0BDL,
0xDAAD56789639AB08L, 0x985DB7933FD39D9BL,
0x84193F60D72AF34FL, 0xC6E9DE8B7EC0C5DCL,
0x01F8FCB784FE9E69L, 0x43081D5C2D14A8FAL,
0xCD2A5925D9681F90L, 0x8FDAB8CE70822903L,
0x48CB9AF28ABC72B6L, 0x0A3B7B1923564425L,
0x70428B155B4EAF1EL, 0x32B26AFEF2A4998DL,
0xF5A348C2089AC238L, 0xB753A929A170F4ABL,
0x3971ED50550C43C1L, 0x7B810CBBFCE67552L,
0xBC902E8706D82EE7L, 0xFE60CF6CAF321874L,
0xE224479F47CB76A0L, 0xA0D4A674EE214033L,
0x67C58448141F1B86L, 0x253565A3BDF52D15L,
0xAB1721DA49899A7FL, 0xE9E7C031E063ACECL,
0x2EF6E20D1A5DF759L, 0x6C0603E6B3B7C1CAL,
0xF6FAE5C07D3274CDL, 0xB40A042BD4D8425EL,
0x731B26172EE619EBL, 0x31EBC7FC870C2F78L,
0xBFC9838573709812L, 0xFD39626EDA9AAE81L,
0x3A28405220A4F534L, 0x78D8A1B9894EC3A7L,
0x649C294A61B7AD73L, 0x266CC8A1C85D9BE0L,
0xE17DEA9D3263C055L, 0xA38D0B769B89F6C6L,
0x2DAF4F0F6FF541ACL, 0x6F5FAEE4C61F773FL,
0xA84E8CD83C212C8AL, 0xEABE6D3395CB1A19L,
0x90C79D3FEDD3F122L, 0xD2377CD44439C7B1L,
0x15265EE8BE079C04L, 0x57D6BF0317EDAA97L,
0xD9F4FB7AE3911DFDL, 0x9B041A914A7B2B6EL,
0x5C1538ADB04570DBL, 0x1EE5D94619AF4648L,
0x02A151B5F156289CL, 0x4051B05E58BC1E0FL,
0x87409262A28245BAL, 0xC5B073890B687329L,
0x4B9237F0FF14C443L, 0x0962D61B56FEF2D0L,
0xCE73F427ACC0A965L, 0x8C8315CC052A9FF6L,
0x3A80143F5CF17F13L, 0x7870F5D4F51B4980L,
0xBF61D7E80F251235L, 0xFD913603A6CF24A6L,
0x73B3727A52B393CCL, 0x31439391FB59A55FL,
0xF652B1AD0167FEEAL, 0xB4A25046A88DC879L,
0xA8E6D8B54074A6ADL, 0xEA16395EE99E903EL,
0x2D071B6213A0CB8BL, 0x6FF7FA89BA4AFD18L,
0xE1D5BEF04E364A72L, 0xA3255F1BE7DC7CE1L,
0x64347D271DE22754L, 0x26C49CCCB40811C7L,
0x5CBD6CC0CC10FAFCL, 0x1E4D8D2B65FACC6FL,
0xD95CAF179FC497DAL, 0x9BAC4EFC362EA149L,
0x158E0A85C2521623L, 0x577EEB6E6BB820B0L,
0x906FC95291867B05L, 0xD29F28B9386C4D96L,
0xCEDBA04AD0952342L, 0x8C2B41A1797F15D1L,
0x4B3A639D83414E64L, 0x09CA82762AAB78F7L,
0x87E8C60FDED7CF9DL, 0xC51827E4773DF90EL,
0x020905D88D03A2BBL, 0x40F9E43324E99428L,
0x2CFFE7D5975E55E2L, 0x6E0F063E3EB46371L,
0xA91E2402C48A38C4L, 0xEBEEC5E96D600E57L,
0x65CC8190991CB93DL, 0x273C607B30F68FAEL,
0xE02D4247CAC8D41BL, 0xA2DDA3AC6322E288L,
0xBE992B5F8BDB8C5CL, 0xFC69CAB42231BACFL,
0x3B78E888D80FE17AL, 0x7988096371E5D7E9L,
0xF7AA4D1A85996083L, 0xB55AACF12C735610L,
0x724B8ECDD64D0DA5L, 0x30BB6F267FA73B36L,
0x4AC29F2A07BFD00DL, 0x08327EC1AE55E69EL,
0xCF235CFD546BBD2BL, 0x8DD3BD16FD818BB8L,
0x03F1F96F09FD3CD2L, 0x41011884A0170A41L,
0x86103AB85A2951F4L, 0xC4E0DB53F3C36767L,
0xD8A453A01B3A09B3L, 0x9A54B24BB2D03F20L,
0x5D45907748EE6495L, 0x1FB5719CE1045206L,
0x919735E51578E56CL, 0xD367D40EBC92D3FFL,
0x1476F63246AC884AL, 0x568617D9EF46BED9L,
0xE085162AB69D5E3CL, 0xA275F7C11F7768AFL,
0x6564D5FDE549331AL, 0x279434164CA30589L,
0xA9B6706FB8DFB2E3L, 0xEB46918411358470L,
0x2C57B3B8EB0BDFC5L, 0x6EA7525342E1E956L,
0x72E3DAA0AA188782L, 0x30133B4B03F2B111L,
0xF7021977F9CCEAA4L, 0xB5F2F89C5026DC37L,
0x3BD0BCE5A45A6B5DL, 0x79205D0E0DB05DCEL,
0xBE317F32F78E067BL, 0xFCC19ED95E6430E8L,
0x86B86ED5267CDBD3L, 0xC4488F3E8F96ED40L,
0x0359AD0275A8B6F5L, 0x41A94CE9DC428066L,
0xCF8B0890283E370CL, 0x8D7BE97B81D4019FL,
0x4A6ACB477BEA5A2AL, 0x089A2AACD2006CB9L,
0x14DEA25F3AF9026DL, 0x562E43B4931334FEL,
0x913F6188692D6F4BL, 0xD3CF8063C0C759D8L,
0x5DEDC41A34BBEEB2L, 0x1F1D25F19D51D821L,
0xD80C07CD676F8394L, 0x9AFCE626CE85B507L};
private long crc = 0;
@Override
public void update(final int b) {
update((byte) (b & 0xFF));
}
public void update(final byte b) {
final int tab_index = ((int) (this.crc >> 56) ^ b) & 0xFF;
this.crc = CRC_TABLE[tab_index] ^ (this.crc << 8);
}
@Override
public void update(final byte[] buffer, final int offset, int length) {
for (int i = offset; length > 0; length--) {
update(buffer[i++]);
}
}
@Override
public void update(final byte[] buffer) {
for (int i = 0; i < buffer.length; i++) {
update(buffer[i]);
}
}
@Override
public long getValue() {
return this.crc;
}
@Override
public void reset() {
this.crc = 0;
}
}

100
hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/util/ZipUtils.java Normal file
View File

@ -0,0 +1,100 @@
/*
* 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.hugegraph.rocksdb.access.util;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.zip.CheckedInputStream;
import java.util.zip.CheckedOutputStream;
import java.util.zip.Checksum;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;
import java.util.zip.ZipOutputStream;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.io.output.NullOutputStream;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public final class ZipUtils {
public static void compress(final String rootDir, final String sourceDir,
final String outputFile, final Checksum checksum) throws
IOException {
try (final FileOutputStream fos = new FileOutputStream(outputFile);
final CheckedOutputStream cos = new CheckedOutputStream(fos, checksum);
final ZipOutputStream zos = new ZipOutputStream(new BufferedOutputStream(cos))) {
ZipUtils.compressDirectoryToZipFile(rootDir, sourceDir, zos);
zos.flush();
fos.getFD().sync();
}
}
private static void compressDirectoryToZipFile(final String rootDir, final String sourceDir,
final ZipOutputStream zos) throws IOException {
final String dir = Paths.get(rootDir, sourceDir).toString();
final File[] files = new File(dir).listFiles();
for (final File file : files) {
final String child = Paths.get(sourceDir, file.getName()).toString();
if (file.isDirectory()) {
compressDirectoryToZipFile(rootDir, child, zos);
} else {
zos.putNextEntry(new ZipEntry(child));
try (final FileInputStream fis = new FileInputStream(file);
final BufferedInputStream bis = new BufferedInputStream(fis)) {
IOUtils.copy(bis, zos);
}
}
}
}
public static void decompress(final String sourceFile, final String outputDir,
final Checksum checksum) throws IOException {
try (final FileInputStream fis = new FileInputStream(sourceFile);
final CheckedInputStream cis = new CheckedInputStream(fis, checksum);
final ZipInputStream zis = new ZipInputStream(new BufferedInputStream(cis))) {
ZipEntry entry;
while ((entry = zis.getNextEntry()) != null) {
final String fileName = entry.getName();
final Path entryPath = Paths.get(outputDir).resolve(fileName).normalize();
if (!entryPath.startsWith(Paths.get(outputDir).normalize())) {
// The file path is not in the expected directory. There may be a Zip Slip
// vulnerability. Ignore it or handle it accordingly.
continue;
}
final File entryFile = entryPath.toFile();
FileUtils.forceMkdir(entryFile.getParentFile());
try (final FileOutputStream fos = new FileOutputStream(entryFile);
final BufferedOutputStream bos = new BufferedOutputStream(fos)) {
IOUtils.copy(zis, bos);
bos.flush();
fos.getFD().sync();
}
}
IOUtils.copy(cis, NullOutputStream.NULL_OUTPUT_STREAM);
}
}
}

0
hugegraph-store/hg-store-rocksdb/src/main/java/place-holder.txt Normal file
View File

102
hugegraph-store/hg-store-rocksdb/src/test/java/org/apache/hugegraph/rocksdb/access/RocksDBFactoryTest.java Normal file
View File

@ -0,0 +1,102 @@
/*
* 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.hugegraph.rocksdb.access;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.configuration2.MapConfiguration;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.config.OptionSpace;
import org.junit.BeforeClass;
public class RocksDBFactoryTest {
@BeforeClass
public static void init() {
OptionSpace.register("rocksdb",
"org.apache.hugegraph.rocksdb.access.RocksDBOptions");
RocksDBOptions.instance();
Map<String, Object> configMap = new HashMap<>();
configMap.put("rocksdb.write_buffer_size", "1048576");
configMap.put("rocksdb.bloom_filter_bits_per_key", "10");
HugeConfig hConfig = new HugeConfig(new MapConfiguration(configMap));
RocksDBFactory rFactory = RocksDBFactory.getInstance();
rFactory.setHugeConfig(hConfig);
}
// @Test
public void testCreateSession() throws InterruptedException {
RocksDBFactory factory = RocksDBFactory.getInstance();
try (RocksDBSession dbSession = factory.createGraphDB("./tmp", "test1")) {
SessionOperator op = dbSession.sessionOp();
op.prepare();
try {
op.put("tbl", "k1".getBytes(), "v1".getBytes());
op.commit();
} catch (Exception e) {
op.rollback();
}
}
factory.destroyGraphDB("test1");
Thread.sleep(100000);
}
// @Test
public void testTotalKeys() {
RocksDBFactory dbFactory = RocksDBFactory.getInstance();
System.out.println(dbFactory.getTotalSize());
System.out.println(dbFactory.getTotalKey().entrySet()
.stream().map(e -> e.getValue()).reduce(0L, Long::sum));
}
// @Test
public void releaseAllGraphDB() {
System.out.println(RocksDBFactory.class);
RocksDBFactory rFactory = RocksDBFactory.getInstance();
if (rFactory.queryGraphDB("bj01") == null) {
rFactory.createGraphDB("./tmp", "bj01");
}
if (rFactory.queryGraphDB("bj02") == null) {
rFactory.createGraphDB("./tmp", "bj02");
}
if (rFactory.queryGraphDB("bj03") == null) {
rFactory.createGraphDB("./tmp", "bj03");
}
RocksDBSession dbSession = rFactory.queryGraphDB("bj01");
dbSession.checkTable("test");
SessionOperator sessionOp = dbSession.sessionOp();
sessionOp.prepare();
sessionOp.put("test", "hi".getBytes(), "byebye".getBytes());
sessionOp.commit();
rFactory.releaseAllGraphDB();
}
}

474
hugegraph-store/hg-store-rocksdb/src/test/java/org/apache/hugegraph/rocksdb/access/RocksDBSessionTest.java Normal file
View File

@ -0,0 +1,474 @@
/*
* 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.hugegraph.rocksdb.access;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.configuration2.MapConfiguration;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.config.OptionSpace;
import org.junit.BeforeClass;
public class RocksDBSessionTest {
private final String graphName = "testDummy";
@BeforeClass
public static void init() {
OptionSpace.register("rocksdb",
"org.apache.hugegraph.rocksdb.access.RocksDBOptions");
RocksDBOptions.instance();
Map<String, Object> configMap = new HashMap<>();
configMap.put("rocksdb.write_buffer_size", "1048576");
configMap.put("rocksdb.bloom_filter_bits_per_key", "10");
HugeConfig hConfig = new HugeConfig(new MapConfiguration(configMap));
RocksDBFactory rFactory = RocksDBFactory.getInstance();
rFactory.setHugeConfig(hConfig);
}
private static byte[] intToBytesForPartId(int v) {
short s = (short) v;
ByteBuffer buffer = ByteBuffer.allocate(Short.BYTES).order(ByteOrder.BIG_ENDIAN);
buffer.putShort(s);
return buffer.array();
}
public static byte[] byteCompose(byte[] b1, byte[] b2) {
byte[] b3 = new byte[b1.length + b2.length];
System.arraycopy(b1, 0, b3, 0, b1.length);
System.arraycopy(b2, 0, b3, b1.length, b2.length);
return b3;
}
private static byte[] keyAppendPartId(int partId, byte[] key) {
byte[] partBytes = intToBytesForPartId(partId);
byte[] targetKey = byteCompose(partBytes, key);
return targetKey;
}
private static byte[] getPartStartKey(int partId) {
return intToBytesForPartId(partId);
}
private static byte[] getPartEndKey(int partId) {
return intToBytesForPartId(partId + 1);
}
// @Test
public void put() {
System.out.println("RocksDBSessionTest::put test");
RocksDBFactory rFactory = RocksDBFactory.getInstance();
RocksDBSession rSession = rFactory.createGraphDB("./tmp", graphName);
String tName = "t1";
rSession.checkTable(tName);
SessionOperator sessionOp = rSession.sessionOp();
sessionOp.put(tName, "a1".getBytes(), "f1".getBytes());
sessionOp.commit();
String vRet = new String(rSession.sessionOp().get(tName, "a1".getBytes()));
assertEquals(vRet, "f1");
}
// @Test
public void selectTable() {
System.out.println("RocksDBSessionTest::selectTable test");
RocksDBFactory rFactory = RocksDBFactory.getInstance();
RocksDBSession rSession = rFactory.createGraphDB("./tmp", graphName);
String tName = "t2";
rSession.checkTable(tName);
assertTrue(rSession.tableIsExist(tName));
}
// @Test
public void batchPut() {
System.out.println("RocksDBSessionTest::batchPut test");
RocksDBFactory rFactory = RocksDBFactory.getInstance();
RocksDBSession rSession = rFactory.createGraphDB("./tmp", graphName);
String tName = "t3";
String tName2 = "t4";
SessionOperator sessionOp = rSession.sessionOp();
sessionOp.put(tName, "a1".getBytes(), "f1".getBytes());
sessionOp.put(tName, "a2".getBytes(), "f2".getBytes());
sessionOp.put(tName, "a3".getBytes(), "f3".getBytes());
sessionOp.put(tName, "a4".getBytes(), "f4".getBytes());
sessionOp.put(tName, "a5".getBytes(), "f5".getBytes());
sessionOp.put(tName, "a6".getBytes(), "f6".getBytes());
sessionOp.commit();
sessionOp.put(tName2, "m1".getBytes(), "k1".getBytes());
sessionOp.put(tName2, "m2".getBytes(), "k2".getBytes());
sessionOp.put(tName2, "m3".getBytes(), "k3".getBytes());
sessionOp.put(tName, "a7".getBytes(), "f7".getBytes());
sessionOp.commit();
String vRet1 = new String(rSession.sessionOp().get(tName, "a1".getBytes()));
String vRet2 = new String(rSession.sessionOp().get(tName2, "m1".getBytes()));
String vRet3 = new String(rSession.sessionOp().get(tName, "a7".getBytes()));
assertEquals(vRet1, "f1");
assertEquals(vRet2, "k1");
assertEquals(vRet3, "f7");
}
// @Test
public void get() {
System.out.println("RocksDBSessionTest::get test");
RocksDBFactory rFactory = RocksDBFactory.getInstance();
RocksDBSession rSession = rFactory.createGraphDB("./tmp", graphName);
String tName = "t5";
rSession.checkTable(tName);
SessionOperator sessionOp = rSession.sessionOp();
sessionOp.put(tName, "hash".getBytes(), "youareok".getBytes());
sessionOp.commit();
String vRet = new String(rSession.sessionOp().get(tName, "hash".getBytes()));
assertEquals(vRet, "youareok");
}
// @Test
public void createTables() {
System.out.println("RocksDBSessionTest::createTables test");
RocksDBFactory rFactory = RocksDBFactory.getInstance();
RocksDBSession rSession = rFactory.createGraphDB("./tmp", graphName);
rSession.createTables("fly1", "fly2", "fly3");
SessionOperator sessionOp = rSession.sessionOp();
sessionOp.put("fly1", "cat1".getBytes(), "hit1".getBytes());
sessionOp.put("fly2", "cat2".getBytes(), "hit2".getBytes());
sessionOp.put("fly3", "cat3".getBytes(), "hit3".getBytes());
sessionOp.commit();
assertTrue(rSession.tableIsExist("fly1"));
assertTrue(rSession.tableIsExist("fly2"));
assertTrue(rSession.tableIsExist("fly3"));
}
// @Test
public void dropTables() {
System.out.println("RocksDBSessionTest::dropTables test");
RocksDBFactory rFactory = RocksDBFactory.getInstance();
RocksDBSession rSession = rFactory.createGraphDB("./tmp", graphName);
rSession.createTables("dummy1");
assertTrue(rSession.tableIsExist("dummy1"));
rSession.dropTables("dummy1");
assertFalse(rSession.tableIsExist("dummy1"));
}
// @Test
public void scan() {
System.out.println("RocksDBSessionTest::scan test");
RocksDBFactory rFactory = RocksDBFactory.getInstance();
String graph1 = "tBatch";
rFactory.createGraphDB("./tmp", graphName);
RocksDBSession rSession = rFactory.createGraphDB("./tmp", graph1);
String tName = "t6";
rSession.checkTable(tName);
SessionOperator sessionOp = rSession.sessionOp();
sessionOp.put(tName, "box1".getBytes(), "gift1".getBytes());
sessionOp.put(tName, "box2".getBytes(), "gift2".getBytes());
sessionOp.put(tName, "box3".getBytes(), "gift3".getBytes());
sessionOp.put(tName, "box4".getBytes(), "gift4".getBytes());
sessionOp.put(tName, "box5".getBytes(), "gift5".getBytes());
sessionOp.put(tName, "box6".getBytes(), "gift6".getBytes());
sessionOp.put(tName, "box7".getBytes(), "gift7".getBytes());
sessionOp.put(tName, "box8".getBytes(), "gift8".getBytes());
sessionOp.commit();
//scan table
ScanIterator it = sessionOp.scan(tName);
while (it.hasNext()) {
RocksDBSession.BackendColumn col = it.next();
System.out.println(new String(col.name) + " : " + new String(col.value));
}
long c = rSession.getApproximateDataSize();
System.out.println(c);
long c1 = RocksDBFactory.getInstance().getTotalSize();
System.out.println(c1);
}
// @Test
public void testScan() {
System.out.println("RocksDBSessionTest::testScan test");
RocksDBFactory rFactory = RocksDBFactory.getInstance();
RocksDBSession rSession = rFactory.createGraphDB("./tmp", graphName);
String tName = "t7";
rSession.checkTable(tName);
SessionOperator sessionOp = rSession.sessionOp();
sessionOp.put(tName, "box1".getBytes(), "gift1".getBytes());
sessionOp.put(tName, "box2".getBytes(), "gift2".getBytes());
sessionOp.put(tName, "box3".getBytes(), "gift3".getBytes());
sessionOp.put(tName, "room1".getBytes(), "killer1".getBytes());
sessionOp.put(tName, "box4".getBytes(), "gift4".getBytes());
sessionOp.put(tName, "box5".getBytes(), "gift5".getBytes());
sessionOp.put(tName, "room2".getBytes(), "killer2".getBytes());
sessionOp.put(tName, "box6".getBytes(), "gift6".getBytes());
sessionOp.put(tName, "boat1".getBytes(), "girl1".getBytes());
sessionOp.put(tName, "box7".getBytes(), "gift7".getBytes());
sessionOp.put(tName, "box8".getBytes(), "gift8".getBytes());
sessionOp.commit();
//prefix scan
ScanIterator it = sessionOp.scan(tName, "bo".getBytes());
while (it.hasNext()) {
RocksDBSession.BackendColumn col = it.next();
System.out.println(new String(col.name) + " : " + new String(col.value));
}
}
// @Test
public void testScan1() {
System.out.println("RocksDBSessionTest::testScan1 test");
RocksDBFactory rFactory = RocksDBFactory.getInstance();
RocksDBSession rSession = rFactory.createGraphDB("./tmp", graphName);
String tName = "t8";
rSession.checkTable(tName);
SessionOperator sessionOp = rSession.sessionOp();
sessionOp.put(tName, "box1".getBytes(), "gift1".getBytes());
sessionOp.put(tName, "box2".getBytes(), "gift2".getBytes());
sessionOp.put(tName, "box3".getBytes(), "gift3".getBytes());
sessionOp.put(tName, "room1".getBytes(), "killer1".getBytes());
sessionOp.put(tName, "box4".getBytes(), "gift4".getBytes());
sessionOp.put(tName, "box5".getBytes(), "gift5".getBytes());
sessionOp.put(tName, "room2".getBytes(), "killer2".getBytes());
sessionOp.put(tName, "box6".getBytes(), "gift6".getBytes());
sessionOp.put(tName, "boat1".getBytes(), "girl1".getBytes());
sessionOp.put(tName, "box7".getBytes(), "gift7".getBytes());
sessionOp.put(tName, "box8".getBytes(), "gift8".getBytes());
sessionOp.commit();
//range scan
ScanIterator it = sessionOp.scan(tName, "box2".getBytes(), "box5".getBytes(),
ScanIterator.Trait.SCAN_GTE_BEGIN |
ScanIterator.Trait.SCAN_LTE_END);
while (it.hasNext()) {
RocksDBSession.BackendColumn col = it.next();
System.out.println(new String(col.name) + " : " + new String(col.value));
}
}
// @Test
public void testCount() {
System.out.println("RocksDBSessionTest::testCount test");
RocksDBFactory rFactory = RocksDBFactory.getInstance();
RocksDBSession rSession = rFactory.createGraphDB("./tmp", graphName);
String tName = "c1";
rSession.checkTable(tName);
int part1 = 1;
int part2 = 2;
int part3 = 3;
SessionOperator sessionOp = rSession.sessionOp();
sessionOp.prepare();
sessionOp.put(tName, keyAppendPartId(part1, "k1".getBytes()), "v1".getBytes());
sessionOp.put(tName, keyAppendPartId(part1, "k2".getBytes()), "v2".getBytes());
sessionOp.put(tName, keyAppendPartId(part1, "k3".getBytes()), "v3".getBytes());
sessionOp.put(tName, intToBytesForPartId(part2), "v".getBytes());
sessionOp.put(tName, keyAppendPartId(part2, "k4".getBytes()), "v4".getBytes());
sessionOp.put(tName, keyAppendPartId(part2, "k5".getBytes()), "v5".getBytes());
sessionOp.put(tName, keyAppendPartId(part2, "k6".getBytes()), "v6".getBytes());
sessionOp.put(tName, intToBytesForPartId(part3), "v".getBytes());
sessionOp.put(tName, keyAppendPartId(part3, "k8".getBytes()), "v8".getBytes());
sessionOp.put(tName, keyAppendPartId(part3, "k9".getBytes()), "v9".getBytes());
sessionOp.put(tName, keyAppendPartId(part3, "k10".getBytes()), "v10".getBytes());
sessionOp.put(tName, keyAppendPartId(part3, "k11".getBytes()), "v11".getBytes());
sessionOp.commit();
// assertEquals(3, rSession.sessionOp().keyCount(tName, getPartStartKey(part1),
// getPartEndKey(part1)));
// assertEquals(4, rSession.sessionOp().keyCount(tName, getPartStartKey(part2),
// getPartEndKey(part2)));
// assertEquals(5, rSession.sessionOp().keyCount(tName, getPartStartKey(part3),
// getPartEndKey(part3)));
assertEquals(12, rSession.sessionOp().keyCount("\0".getBytes(StandardCharsets.UTF_8),
"\255".getBytes(StandardCharsets.UTF_8),
tName));
assertEquals(12, rSession.sessionOp().estimatedKeyCount(tName));
}
// @Test
public void merge() {
System.out.println("RocksDBSessionTest::merge test");
RocksDBFactory rFactory = RocksDBFactory.getInstance();
RocksDBSession rSession = rFactory.createGraphDB("./tmp", graphName);
String tName = "room1";
rSession.checkTable(tName);
byte[] x2 = longToByteArray(19);
System.out.println(x2);
SessionOperator sessionOp = rSession.sessionOp();
byte[] x1 = sessionOp.get(tName, "p3".getBytes());
String str1 = new String(x1);
if (x1 != null) {
long curValue = longFromByteArray(x1);
System.out.println("current value:" + curValue);
sessionOp.merge(tName, "p3".getBytes(), longToByteArray(10));
sessionOp.commit();
}
byte[] value = rSession.sessionOp().get(tName, "p3".getBytes());
final long longValue = longFromByteArray(value);
System.out.println("after merge value: " + longValue);
sessionOp.put(tName, "p3".getBytes(), "19".getBytes());
sessionOp.commit();
ScanIterator it = sessionOp.scan(tName, "p".getBytes());
System.out.println("after put ------>");
while (it.hasNext()) {
RocksDBSession.BackendColumn col = it.next();
System.out.println(new String(col.name) + ":" + new String(col.value));
}
}
// @Test
public void truncate() {
System.out.println("RocksDBSessionTest::truncate test");
RocksDBFactory rFactory = RocksDBFactory.getInstance();
RocksDBSession rSession = rFactory.createGraphDB("./tmp", "gg1");
rSession.checkTable("f1");
SessionOperator op = rSession.sessionOp();
op.put("f1", "m1".getBytes(), "n1".getBytes());
op.put("f1", "m2".getBytes(), "n2".getBytes());
op.put("f1", "m3".getBytes(), "n3".getBytes());
op.commit();
rSession.checkTable("f2");
RocksDBSession rocksDBSession2 = rFactory.createGraphDB("./tmp", "gg2");
rocksDBSession2.checkTable("txt");
rSession.truncate();
op = rSession.sessionOp();
op.put("f1", "beijing".getBytes(), "renkou".getBytes());
op.commit();
rFactory.releaseAllGraphDB();
}
// @Test
public void batchPut2() {
RocksDBFactory rFactory = RocksDBFactory.getInstance();
RocksDBSession rSession = rFactory.createGraphDB("./tmp", "gg1");
rSession.checkTable("f1");
SessionOperator op = rSession.sessionOp();
try {
op.put("f1", "m2".getBytes(), "xx2".getBytes());
op.put("f1", "m1".getBytes(), "xx1".getBytes());
op.put("f1", "m3".getBytes(), "xx3".getBytes());
op.put("f1", "m4".getBytes(), "xx5".getBytes());
op.commit();
op.deleteRange("f1", "m5".getBytes(), "m2".getBytes());
op.put("f2", new byte[]{1, 1}, new byte[]{1, -3});
op.put("f2", new byte[]{1, -2}, new byte[]{2, 0});
op.put("f2", new byte[]{1, 10}, new byte[]{1, 5});
op.put("f2", new byte[]{1, 9}, new byte[]{1, 3});
op.commit();
op.deleteRange("f2", new byte[]{1, 1}, new byte[]{1, -1});
op.commit();
ScanIterator it = op.scan("f2");
while (it.hasNext()) {
RocksDBSession.BackendColumn col = it.next();
System.out.println(Arrays.toString(col.name) + "=>" + Arrays.toString(col.value));
}
} catch (DBStoreException e) {
System.out.println(e);
}
}
// @Test
public void doStuff() {
byte[] stuff = new byte[]{1, 3, 1, -3};
System.out.println(Arrays.toString(stuff));
}
private byte[] longToByteArray(long l) {
ByteBuffer buf = ByteBuffer.allocate(Long.BYTES).order(ByteOrder.LITTLE_ENDIAN);
buf.putLong(l);
return buf.array();
}
private long longFromByteArray(byte[] a) {
ByteBuffer buf = ByteBuffer.allocate(Long.BYTES).order(ByteOrder.LITTLE_ENDIAN);
buf.put(a);
buf.flip();
return buf.getLong();
}
}

65
hugegraph-store/hg-store-rocksdb/src/test/java/org/apache/hugegraph/rocksdb/access/RocksdbDump.java Normal file
View File

@ -0,0 +1,65 @@
/*
* 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.hugegraph.rocksdb.access;
import java.util.ArrayList;
import java.util.List;
import org.rocksdb.ColumnFamilyDescriptor;
import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.DBOptions;
import org.rocksdb.Options;
import org.rocksdb.ReadOptions;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;
public class RocksdbDump {
// @Test
public void dump() throws RocksDBException {
String dbPath =
"D:\\Workspaces\\baidu\\hugegraph\\hugegraph-store\\tmp\\8500\\db\\default" +
"\\hugegraph\\g";
List<ColumnFamilyDescriptor> cfDescriptors = new ArrayList<>();
List<byte[]> columnFamilyBytes = RocksDB.listColumnFamilies(new Options(), dbPath);
ColumnFamilyOptions cfOptions = new ColumnFamilyOptions();
if (columnFamilyBytes.size() > 0) {
for (byte[] columnFamilyByte : columnFamilyBytes) {
cfDescriptors.add(new ColumnFamilyDescriptor(columnFamilyByte, cfOptions));
}
}
final List<ColumnFamilyHandle> columnFamilyHandles = new ArrayList<>();
try (final DBOptions options = new DBOptions();
final RocksDB db = RocksDB.open(options, dbPath, cfDescriptors, columnFamilyHandles)) {
for (ColumnFamilyHandle handle : columnFamilyHandles) {
System.out.println(new String(handle.getName()) + "---------------");
try (RocksIterator iterator = db.newIterator(handle, new ReadOptions())) {
iterator.seekToFirst();
while (iterator.isValid()) {
byte[] key = iterator.key();
// System.out.println(new String(key) + " -- " + Bytes.toHex(key));
iterator.next();
}
}
}
}
}
}

430
hugegraph-store/hg-store-rocksdb/src/test/java/org/apache/hugegraph/rocksdb/access/SnapshotManagerTest.java Normal file
View File

@ -0,0 +1,430 @@
/*
* 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.hugegraph.rocksdb.access;
import java.io.File;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import org.apache.hugegraph.store.term.HgPair;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class SnapshotManagerTest {
final String tempSuffix = "_temp_";
final String graphName = "test_graph";
// @Test
public void testRetrieveIndex() throws DBStoreException {
String snapshotPath =
"D:\\Code\\baidu\\starhugegraph\\hugegraph-store\\tmp\\8501\\raft\\1/snapshot" +
"\\snapshot_40\\graph_1_p1";
long lastIndex = 0L;
File file = new File(snapshotPath);
File parentFile = new File(file.getParent());
String[] arr = parentFile.getName().split("_");
if (arr.length >= 2) {
lastIndex = Long.parseLong(arr[arr.length - 1]);
} else {
throw new DBStoreException(String.format("Invalid Snapshot path %s", snapshotPath));
}
System.out.println(lastIndex);
}
// @Test
public void testFoundMaxDir() {
// String path = "D:\\Code\\baidu\\starhugegraph\\hugegraph-store\\tmp\\8503\\db\\default
// \\hugegraph\\g\\0_123";
// String path = "D:\\Code\\baidu\\starhugegraph\\hugegraph-store\\tmp\\8503\\db\\default
// \\hugegraph\\g\\0";
// String path = "D:\\Code\\baidu\\starhugegraph\\hugegraph-store\\tmp\\8501\\db\\default
// \\hugegraph\\g\\0";
// String path = "D:\\tmp\\db\\0";
// String path = "D:\\tmp\\db\\0_111";
String path = "D:\\tmp\\db\\0_111_22222";
File file = new File(path);
int strIndex = file.getName().indexOf("_");
String defaultName;
if (strIndex < 0) {
defaultName = file.getName();
} else {
defaultName = file.getName().substring(0, strIndex);
}
String prefix = defaultName + "_";
File parentFile = new File(file.getParent());
final List<HgPair<Long, Long>> dbs = new ArrayList<>();
final File[] files = parentFile.listFiles();
if (files != null) {
// search all db path
for (final File sFile : files) {
final String name = sFile.getName();
if (!name.startsWith(prefix) && !name.equals(defaultName)) {
continue;
}
if (name.endsWith(tempSuffix)) {
continue;
}
long v1 = -1L;
long v2 = -1L;
if (name.length() > defaultName.length()) {
String[] versions = name.substring(prefix.length()).split("_");
if (versions.length == 1) {
v1 = Long.parseLong(versions[0]);
} else if (versions.length == 2) {
v1 = Long.parseLong(versions[0]);
v2 = Long.parseLong(versions[1]);
} else {
continue;
}
}
dbs.add(new HgPair<>(v1, v2));
}
}
// get last index db path
String latestDBPath = "";
if (!dbs.isEmpty()) {
dbs.sort((o1, o2) -> o1.getKey().equals(o2.getKey()) ?
o1.getValue().compareTo(o2.getValue()) :
o1.getKey().compareTo(o2.getKey()));
final int dbCount = dbs.size();
// delete old db
for (int i = 0; i < dbCount; i++) {
final HgPair<Long, Long> pair = dbs.get(i);
String curDBName;
if (pair.getKey() == -1L) {
curDBName = defaultName;
} else if (pair.getValue() == -1L) {
curDBName = String.format("%s_%d", defaultName, pair.getKey());
} else {
curDBName =
String.format("%s_%d_%d", defaultName, pair.getKey(), pair.getValue());
}
String curDBPath = Paths.get(parentFile.getPath(), curDBName).toString();
if (i == dbCount - 1) {
latestDBPath = curDBPath;
} else {
log.info("delete old dbpath {}", curDBPath);
}
}
} else {
latestDBPath = Paths.get(parentFile.getPath(), defaultName).toString();
}
log.info("{} latest db path {}", this.graphName, latestDBPath);
}
// @Test
public void testDefaultPath() {
// String latestDBPath = "D:\\Code\\baidu\\starhugegraph\\hugegraph-store\\tmp\\8501\\db
// \\default\\hugegraph\\g\\0_123";
String latestDBPath =
"D:\\Code\\baidu\\starhugegraph\\hugegraph-store\\tmp\\8501\\db\\default" +
"\\hugegraph\\g\\0";
File file = new File(latestDBPath);
String parent = file.getParent();
String defaultName = file.getName().split("_")[0];
String defaultPath = Paths.get(parent, defaultName).toString();
System.out.println(defaultPath);
}
// private static final String graphName = "unit-test-graph";
// private static final String graphName2 = "unit-test-graph-2";
// private static final int partition_1 = 1;
// private static final int partition_2 = 2;
// private static final int partition_3 = 3;
// private static final int partition_4 = 4;
// private static ArrayList<String> tableList = new ArrayList<String>();
// private static ArrayList<Integer> partitionList = new ArrayList<Integer>();
// private static RocksDBFactory factory = RocksDBFactory.getInstance();
// private static HugeConfig hConfig;
// private static HugeConfig hConfig2;
// private int index = 0;
//
// private static byte[] values = new byte[1024];
//
//
// // @BeforeClass
// public static void beforeClass() {
// // Register config
// OptionSpace.register("rocksdb",
// "org.apache.hugegraph.rocksdb.access.RocksDBOptions");
// // Register backend
// URL configPath = SnapshotManagerTest.class.getResource("/hugegraph.properties");
// URL configPath2 = SnapshotManagerTest.class.getResource("/hugegraph-2.properties");
// hConfig = new HugeConfig(configPath.getPath());
// hConfig2 = new HugeConfig(configPath2.getPath());
// factory.setHugeConfig(hConfig);
// // init tables
// tableList.add("t1");
// tableList.add("t2");
// tableList.add("t3");
// tableList.add("t4");
// tableList.add("abc/t5");
// tableList.add("../t6");
// // init partitons
// partitionList.add(partition_1);
// partitionList.add(partition_2);
// partitionList.add(partition_3);
// partitionList.add(partition_4);
//
// for (int i = 0; i < 1024; i++)
// values[i] = (byte) (i % 0x7F);
//
// }
//
// @AfterClass
// public static void afterClass() {
// factory.releaseAllGraphDB();
// }
// // @Test
// public void test(){
// Map<String, Object> configMap = new HashMap<>();
// configMap.put("rocksdb.data_path", "tmp/test2/huge/");
// configMap.put("rocksdb.wal_path", "tmp/test2/wal");
// configMap.put("rocksdb.snapshot_path", "tmp/test2/snapshot");
// configMap.put("rocksdb.write_buffer_size","1048576");
// HugeConfig config = new HugeConfig(configMap);
// factory.setHugeConfig(config);
// RocksDBSession session = factory.createGraphDB(graphName, false);
// SessionOperator sessionOp = session.sessionOp();
//
// String table = "t1";
//
// String startKey = String.format("%d_%08d", partition_1, 0);
// String lastKey = String.format("%d_%08d", partition_1, 99);
// String endKey = String.format("%d_%08d", partition_1, 100);
//
//
// printKV(sessionOp, table, startKey);
// printKV(sessionOp, table, lastKey);
// }
//
// private void printKV(SessionOperator sessionOp, String table, String key) {
// byte[] value = sessionOp.get(table, key.getBytes());
// if (value != null)
// System.out.printf("table=%s, key=%s, value=%s\n", table, key, new String(value));
// else
// System.out.printf("table=%s, key=%s, does not found\n", table, key);
// }
//
// private void insertData(int count) {
// RocksDBSession session = factory.createGraphDB(graphName, false);
// SessionOperator sessionOp = session.sessionOp();
// sessionOp.prepare();
// sessionOp.prepare();
// final int begin = index;
// final int end = index + count;
// System.out.printf("begin to insert data from index %d to %d\n", begin, end);
//
// for (; index < end; index++) {
// for (String table : tableList) {
// for (Integer partition : partitionList) {
// String key = String.format("%d_%08d", partition, index);
// String value = String.format("%s_%08d", table, index);
// sessionOp.put(table, key.getBytes(), value.getBytes());
// }
// }
// if (index % 1000 == 999) {
// System.out.printf("insertData, commit index %d\n", index+1);
// sessionOp.commit();
// sessionOp.prepare();
// }
// }
// sessionOp.commit();
// session.flush();
// System.out.printf("inserted %d pairs data\n", count);
// }
//
// // @Test
// public void testSnapshotMetadata() {
// DBSnapshotMeta metadata = new DBSnapshotMeta();
// metadata.setGraphName(graphName);
// metadata.setPartitionId(partition_1);
// metadata.setStartKey("k1".getBytes());
// metadata.setEndKey("k9".getBytes());
// HashMap<String, String> sstFiles = new HashMap<String, String>();
// sstFiles.put("cf1", "cf1.sst");
// sstFiles.put("cf2", "cf2.sst");
// sstFiles.put("cf3", "cf3.sst");
// sstFiles.put("cf4", "cf4.sst");
// metadata.setCreatedDate(new Date());
// metadata.setSstFiles(sstFiles);
//
// System.out.println("metadata object:");
// System.out.println(metadata.toString());
//
// String jsonStr = JSON.toJSONString(metadata);
// System.out.println("metadata json:");
// System.out.println(jsonStr);
//
// DBSnapshotMeta metadata2 = JSON.parseObject(jsonStr, DBSnapshotMeta.class);
// System.out.println("metadata2 object:");
// System.out.println(metadata2.toString());
//
// assertEquals(metadata, metadata2);
// }
//
// // @Test
// public void testCompress() throws IOException {
// final String rootDir = "./tmp/fileutils";
// final File dir1 = Paths.get(rootDir, "dir1").toFile();
// final File dir2 = Paths.get(rootDir, "dir2").toFile();
// final File dir3 = Paths.get(rootDir, "dir3").toFile();
// final File dir4 = Paths.get(rootDir, "dir4").toFile();
// final File file0 = Paths.get(rootDir, "file0").toFile();
// final File file1 = Paths.get(dir1.getAbsolutePath(), "file1").toFile();
// FileUtils.forceMkdirParent(dir1);
// FileUtils.forceMkdirParent(dir2);
// FileUtils.forceMkdirParent(dir3);
// FileUtils.deleteDirectory(dir2);
// FileUtils.forceMkdirParent(dir4);
// FileUtils.moveDirectory(dir3, dir1);
// FileUtils.touch(file0);
// FileUtils.touch(file1);
// final Checksum c1 = new CRC64();
// ZipUtils.compress("./tmp", "fileutils", "./tmp/fileutils.zip", c1);
//
// FileUtils.deleteDirectory(new File("./tmp/fileutils"));
//
// final Checksum c2 = new CRC64();
// ZipUtils.decompress("./tmp/fileutils.zip", "./tmp", c2);
// Assert.assertEquals(c1.getValue(), c2.getValue());
// }
//
//
// // @Test
// public void testDecompress() throws IOException {
// final Checksum checksum = new CRC64();
// String dir = "./tmp/server-00/snapshot/test/import/1";
// String source = Paths.get(dir, "snapshot.zip").toString();
// FileUtils.deleteDirectory(Paths.get(dir, "snapshot").toFile());
// ZipUtils.decompress(source, dir, checksum);
// }
//
//
// // @Test
// public void testExportSnapshot() throws DBStoreException, IOException {
// final String exportDir = hConfig.getString("rocksdb.snapshot_path");
// File exportDirFile = new File(exportDir);
// if (exportDirFile.exists()) {
// FileUtils.forceDelete(new File(exportDir));
// }
// FileUtils.forceMkdir(new File(exportDir));
//
// RocksDBSession session = factory.createGraphDB(graphName, false);
// insertData(1000);
// String startKey = String.format("%d_%08d", partition_1, 0);
// String endKey = String.format("%d_%08d", partition_1, 100);
// session.exportSnapshot(exportDir, 1, startKey.getBytes(), endKey.getBytes());
// }
//
// // @Test
// public void testImportSnapshot() throws DBStoreException, IOException {
// final String importDir = hConfig.getString("rocksdb.snapshot_path");
// FileUtils.forceMkdir(new File(importDir));
//
// factory.setHugeConfig(hConfig2);
// RocksDBSession session = factory.createGraphDB(graphName, false);
// SessionOperator sessionOp = session.sessionOp();
//
// String table = "t1";
//
// String startKey = String.format("%d_%08d", partition_1, 0);
// String lastKey = String.format("%d_%08d", partition_1, 99);
// String endKey = String.format("%d_%08d", partition_1, 100);
//
// session.importSnapshot(importDir, partition_1);
// long keyCount = session.sessionOp().keyCount(table, startKey.getBytes(), endKey
// .getBytes());
// System.out.printf("key count of %s is %d\n", graphName, keyCount);
//
//
// System.out.printf("after load snapshot\n");
// printKV(sessionOp, table, startKey);
// printKV(sessionOp, table, lastKey);
// }
//
// // @Test
// public void testExportSnapshot2() throws DBStoreException, IOException {
// final String exportDir = hConfig.getString("rocksdb.snapshot_path");
// deleteDir(new File(hConfig.getString("rocksdb.data_path")));
// deleteDir(new File(exportDir));
// FileUtils.forceMkdir(new File(exportDir));
//
// RocksDBSession session = factory.createGraphDB(graphName, false);
//
// long seqNum = session.getLatestSequenceNumber();
// System.out.println(seqNum);
// insertData(500);
// System.out.println(session.getLatestSequenceNumber());
// String startKey = String.format("%d_%08d", partition_1, index - 200);
// String endKey = String.format("%d_%08d", partition_1, index - 100);
//
// ScanIterator cfIterator = session.sessionOp().scanRaw(startKey.getBytes(), endKey
// .getBytes(), seqNum);
// while (cfIterator.hasNext()) {
// ScanIterator iterator = cfIterator.next();
// System.out.println("cf name is " + new String(cfIterator.position()));
// while (iterator.hasNext()) {
// RocksDBSession.BackendColumn col = iterator.next();
// System.out.println(new String(col.name));
// }
// System.out.println("-----------------------------------------");
// iterator.close();
// }
// cfIterator.close();
// }
//
// // @Test
// public void testImportSnapshot2() throws DBStoreException, IOException {
// final String importDir = "/tmp/snapshot";
// deleteDir(new File("/tmp/rocksdb/data" + "/" + graphName2));
// RocksDBSession session = factory.createGraphDB(graphName2, false);
// SessionOperator sessionOp = session.sessionOp();
//
//
// session.flush();
//
// String table = "t1";
//
// String k1 = String.format("%d_%08d", partition_1, 100);
// String k2 = String.format("%d_%08d", partition_1, 109);
// String k3 = String.format("%d_%08d", partition_1, 110);
//
// printKV(sessionOp, table, k1);
// printKV(sessionOp, table, k2);
// printKV(sessionOp, table, k3);
//
// }
//
//
// private static boolean deleteDir(File dir) {
// if (dir.isDirectory())
// for (File file : dir.listFiles())
// deleteDir(file);
// return dir.delete();
// }
}

0
hugegraph-store/hg-store-rocksdb/src/test/java/place-holder.txt Normal file
View File

29
hugegraph-store/hg-store-rocksdb/src/test/resources/hugegraph-2.properties Normal file
View File

@ -0,0 +1,29 @@
#
# 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.
#
# rocksdb backend config
rocksdb.data_path=./tmp/rocksdb-2/data
rocksdb.wal_path=./tmp/rocksdb-2/data
rocksdb.snapshot_path=./tmp/rocksdb-2/snapshot
rocksdb.write_buffer_size=1048576
rocksdb.min_write_buffer_number_to_merge=2
rocksdb.max_write_buffer_number=4
rocksdb.level0_file_num_compaction_trigger=2
rocksdb.target_file_size_base=1048576
rocksdb.target_file_size_multiplier=2
rocksdb.max_bytes_for_level_base=1048576
rocksdb.max_bytes_for_level_multiplier=2

29
hugegraph-store/hg-store-rocksdb/src/test/resources/hugegraph.properties Normal file
View File

@ -0,0 +1,29 @@
#
# 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.
#
# rocksdb backend config
rocksdb.data_path=./tmp/rocksdb/data
rocksdb.wal_path=./tmp/rocksdb/data
rocksdb.snapshot_path=./tmp/rocksdb/snapshot
rocksdb.write_buffer_size=1048576
rocksdb.min_write_buffer_number_to_merge=2
rocksdb.max_write_buffer_number=4
rocksdb.level0_file_num_compaction_trigger=2
rocksdb.target_file_size_base=1048576
rocksdb.target_file_size_multiplier=2
rocksdb.max_bytes_for_level_base=1048576
rocksdb.max_bytes_for_level_multiplier=2

30
hugegraph-store/hg-store-rocksdb/src/test/resources/log4j2.xml Normal file
View File

@ -0,0 +1,30 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<Configuration status="WARN">
<Appenders>
<Console name="Console" target="SYSTEM_OUT">
<PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS}: [%-5level] %msg%n" />
</Console>
</Appenders>
<Loggers>
<Root level="debug">
<AppenderRef ref="Console" />
</Root>
</Loggers>
</Configuration>

14
hugegraph-store/pom.xml
View File

@ -38,8 +38,8 @@
<module>hg-store-grpc</module>
<module>hg-store-client</module>
<module>hg-store-test</module>
<module>hg-store-rocksdb</module>
<!-- TODO: uncomment later-->
<!-- <module>hg-store-rocksdb</module>-->
<!-- <module>hg-store-core</module>-->
<!-- <module>hg-store-node</module>-->
<!-- <module>hg-store-dist</module>-->
@ -67,11 +67,11 @@
<artifactId>hg-store-grpc</artifactId>
<version>${project.version}</version>
</dependency>
<!-- <dependency>-->
<!-- <groupId>org.apache.hugegraph</groupId>-->
<!-- <artifactId>hg-store-rocksdb</artifactId>-->
<!-- <version>${project.version}</version>-->
<!-- </dependency>-->
<dependency>
<groupId>org.apache.hugegraph</groupId>
<artifactId>hg-store-rocksdb</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hugegraph</groupId>
<artifactId>hg-store-client</artifactId>
@ -254,7 +254,7 @@
<!-- <version>2.20</version>-->
<!-- <executions>-->
<!-- <execution>-->
<!-- <id>store-rocksdb-test</id>-->
<!-- <id>sit -test</id>-->
<!-- <goals>-->
<!-- <goal>test</goal>-->
<!-- </goals>-->