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

support tikv 1st version 

Change-Id: I23fd71ec977a1944a02ea726717fc8b3c8659a15
This commit is contained in:
zhangyi51 2021-02-22 20:25:02 +08:00
parent e7a6e62922
commit e12290cabb
15 changed files with 2059 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
hugegraph-dist/pom.xml
View File

@ -64,6 +64,11 @@
<artifactId>hugegraph-postgresql</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.baidu.hugegraph</groupId>
<artifactId>hugegraph-tikv</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.tinkerpop</groupId>

3
hugegraph-dist/src/assembly/static/conf/log4j2.xml
View File

@ -98,5 +98,8 @@
<AsyncLogger name="com.baidu.hugegraph.api.filter.AuthenticationFilter" level="INFO" additivity="false">
<appender-ref ref="audit"/>
</AsyncLogger>
<AsyncLogger name="org.tikv.common.operation.KVErrorHandler" level="ERROR" additivity="false">
<appender-ref ref="file"/>
</AsyncLogger>
</loggers>
</configuration>

12
hugegraph-dist/src/main/java/com/baidu/hugegraph/dist/RegisterUtil.java vendored
View File

@ -93,6 +93,9 @@ public class RegisterUtil {
case "postgresql":
registerPostgresql();
break;
case "tikv":
registerTikv();
break;
default:
throw new HugeException("Unsupported backend type '%s'",
backend);
@ -182,6 +185,15 @@ public class RegisterUtil {
"com.baidu.hugegraph.backend.store.postgresql.PostgresqlStoreProvider");
}
public static void registerTikv() {
// Register config
OptionSpace.register("tikv",
"com.baidu.hugegraph.backend.store.tikv.TikvOptions");
// Register backend
BackendProviderFactory.register("tikv",
"com.baidu.hugegraph.backend.store.tikv.TikvStoreProvider");
}
public static void registerServer() {
// Register ServerOptions (rest-server)
OptionSpace.register("server", "com.baidu.hugegraph.config.ServerOptions");

2
hugegraph-dist/src/main/resources/backend.properties
View File

@ -1 +1 @@
backends=[cassandra, scylladb, rocksdb, mysql, palo, hbase, postgresql]
backends=[cassandra, scylladb, rocksdb, mysql, palo, hbase, postgresql, tikv]

26
hugegraph-tikv/pom.xml Normal file
View File

@ -0,0 +1,26 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>hugegraph</artifactId>
<groupId>com.baidu.hugegraph</groupId>
<version>0.12.0</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>hugegraph-tikv</artifactId>
<dependencies>
<dependency>
<groupId>com.baidu.hugegraph</groupId>
<artifactId>hugegraph-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.tikv</groupId>
<artifactId>tikv-client-java</artifactId>
<version>3.1.0-SNAPSHOT</version>
</dependency>
</dependencies>
</project>

127
hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvFeatures.java Normal file
View File

@ -0,0 +1,127 @@
/*
* Copyright 2017 HugeGraph Authors
*
* 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 com.baidu.hugegraph.backend.store.tikv;
import com.baidu.hugegraph.backend.store.BackendFeatures;
public class TikvFeatures implements BackendFeatures {
@Override
public boolean supportsScanToken() {
return false;
}
@Override
public boolean supportsScanKeyPrefix() {
return true;
}
@Override
public boolean supportsScanKeyRange() {
return true;
}
@Override
public boolean supportsQuerySchemaByName() {
return false;
}
@Override
public boolean supportsQueryByLabel() {
return false;
}
@Override
public boolean supportsQueryWithRangeCondition() {
return true;
}
@Override
public boolean supportsQuerySortByInputIds() {
return true;
}
@Override
public boolean supportsQueryWithOrderBy() {
return true;
}
@Override
public boolean supportsQueryWithContains() {
return false;
}
@Override
public boolean supportsQueryWithContainsKey() {
return false;
}
@Override
public boolean supportsQueryByPage() {
return true;
}
@Override
public boolean supportsDeleteEdgeByLabel() {
return false;
}
@Override
public boolean supportsUpdateVertexProperty() {
// Vertex properties are stored in a cell(column value)
return false;
}
@Override
public boolean supportsMergeVertexProperty() {
return false;
}
@Override
public boolean supportsUpdateEdgeProperty() {
// Edge properties are stored in a cell(column value)
return false;
}
@Override
public boolean supportsTransaction() {
return false;
}
@Override
public boolean supportsNumberType() {
return false;
}
@Override
public boolean supportsAggregateProperty() {
return false;
}
@Override
public boolean supportsTtl() {
return true;
}
@Override
public boolean supportsOlapProperties() {
return false;
}
}

37
hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvMetrics.java Normal file
View File

@ -0,0 +1,37 @@
/*
* Copyright 2017 HugeGraph Authors
*
* 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 com.baidu.hugegraph.backend.store.tikv;
import java.util.Map;
import com.baidu.hugegraph.backend.store.BackendMetrics;
import com.baidu.hugegraph.util.E;
public class TikvMetrics implements BackendMetrics {
public TikvMetrics(TikvSessions tikv) {
E.checkArgumentNotNull(tikv, "Tikv connection is not opened");
}
@Override
public Map<String, Object> metrics() {
return null;
}
}

90
hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvOptions.java Normal file
View File

@ -0,0 +1,90 @@
/*
* Copyright 2017 HugeGraph Authors
*
* 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 com.baidu.hugegraph.backend.store.tikv;
import static com.baidu.hugegraph.config.OptionChecker.*;
import com.baidu.hugegraph.config.ConfigOption;
import com.baidu.hugegraph.config.OptionHolder;
public class TikvOptions extends OptionHolder {
private TikvOptions() {
super();
}
private static volatile TikvOptions instance;
public static synchronized TikvOptions instance() {
if (instance == null) {
instance = new TikvOptions();
instance.registerOptions();
}
return instance;
}
public static final ConfigOption<String> TIKV_PDS =
new ConfigOption<>(
"tikv.pds",
"The addresses of Tikv pds, separated with commas.",
disallowEmpty(),
"localhost"
);
public static final ConfigOption<Integer> TIKV_BATCH_GET_CONCURRENCY =
new ConfigOption<>(
"tikv.batch_get_concurrency",
"The number of thread pool size for batch get of tikv client.",
disallowEmpty(),
20
);
public static final ConfigOption<Integer> TIKV_BATCH_PUT_CONCURRENCY =
new ConfigOption<>(
"tikv.batch_put_concurrency",
"The number of thread pool size for batch put of tikv client.",
disallowEmpty(),
20
);
public static final ConfigOption<Integer> TIKV_BATCH_DELETE_CONCURRENCY =
new ConfigOption<>(
"tikv.batch_delete_concurrency",
"The number of thread pool size for batch delete of tikv client.",
disallowEmpty(),
20
);
public static final ConfigOption<Integer> TIKV_BATCH_SCAN_CONCURRENCY =
new ConfigOption<>(
"tikv.batch_scan_concurrency",
"The number of thread pool size for batch scan of tikv client.",
disallowEmpty(),
5
);
public static final ConfigOption<Integer> TIKV_DELETE_RANGE_CONCURRENCY =
new ConfigOption<>(
"tikv.delete_range_concurrency",
"The number of thread pool size for delete range of tikv client.",
disallowEmpty(),
20
);
}

94
hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvSessions.java Normal file
View File

@ -0,0 +1,94 @@
/*
* Copyright 2017 HugeGraph Authors
*
* 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 com.baidu.hugegraph.backend.store.tikv;
import java.util.Set;
import org.apache.commons.lang3.tuple.Pair;
import com.baidu.hugegraph.backend.store.BackendEntry.BackendColumnIterator;
import com.baidu.hugegraph.backend.store.BackendSession.AbstractBackendSession;
import com.baidu.hugegraph.backend.store.BackendSessionPool;
import com.baidu.hugegraph.config.HugeConfig;
public abstract class TikvSessions extends BackendSessionPool {
public TikvSessions(HugeConfig config, String database, String store) {
super(config, database + "/" + store);
}
public abstract Set<String> openedTables();
public abstract void createTable(String... tables);
public abstract void dropTable(String... tables);
public abstract boolean existsTable(String table);
@Override
public abstract Session session();
/**
* Session for RocksDB
*/
public static abstract class Session extends AbstractBackendSession {
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 abstract Pair<byte[], byte[]> keyRange(String table);
public abstract void put(String table, byte[] key, byte[] value);
public abstract void increase(String table, byte[] key, byte[] value);
public abstract void delete(String table, byte[] key);
public abstract void deletePrefix(String table, byte[] key);
public abstract void deleteRange(String table,
byte[] keyFrom, byte[] keyTo);
public abstract byte[] get(String table, byte[] key);
public abstract BackendColumnIterator scan(String table);
public abstract BackendColumnIterator scan(String table,
byte[] prefix);
public BackendColumnIterator scan(String table,
byte[] keyFrom,
byte[] keyTo) {
return this.scan(table, keyFrom, keyTo, SCAN_LT_END);
}
public abstract BackendColumnIterator scan(String table,
byte[] keyFrom,
byte[] keyTo,
int scanType);
public static boolean matchScanType(int expected, int actual) {
return (expected & actual) == expected;
}
}
public interface Countable {
public long count();
}
}

562
hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvStdSessions.java Normal file
View File

@ -0,0 +1,562 @@
/*
* Copyright 2017 HugeGraph Authors
*
* 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 com.baidu.hugegraph.backend.store.tikv;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.lang3.tuple.Pair;
import org.tikv.common.TiConfiguration;
import org.tikv.common.TiSession;
import org.tikv.common.key.Key;
import org.tikv.kvproto.Kvrpcpb;
import org.tikv.raw.RawKVClient;
import org.tikv.shade.com.google.protobuf.ByteString;
import com.baidu.hugegraph.backend.store.BackendEntry;
import com.baidu.hugegraph.backend.store.BackendEntry.BackendColumnIterator;
import com.baidu.hugegraph.backend.store.BackendEntryIterator;
import com.baidu.hugegraph.config.HugeConfig;
import com.baidu.hugegraph.util.Bytes;
import com.baidu.hugegraph.util.E;
import com.baidu.hugegraph.util.StringEncoding;
public class TikvStdSessions extends TikvSessions {
private final HugeConfig config;
private volatile RawKVClient tikvClient;
private final Map<String, Integer> tables;
private final AtomicInteger refCount;
private static int tableCode = 0;
public TikvStdSessions(HugeConfig config, String database, String store) {
super(config, database, store);
this.config = config;
TiConfiguration conf = TiConfiguration.createRawDefault(
this.config.get(TikvOptions.TIKV_PDS));
conf.setBatchGetConcurrency(
this.config.get(TikvOptions.TIKV_BATCH_GET_CONCURRENCY));
conf.setBatchPutConcurrency(
this.config.get(TikvOptions.TIKV_BATCH_PUT_CONCURRENCY));
conf.setBatchDeleteConcurrency(
this.config.get(TikvOptions.TIKV_BATCH_DELETE_CONCURRENCY));
conf.setBatchScanConcurrency(
this.config.get(TikvOptions.TIKV_BATCH_SCAN_CONCURRENCY));
conf.setDeleteRangeConcurrency(
this.config.get(TikvOptions.TIKV_DELETE_RANGE_CONCURRENCY));
TiSession session = TiSession.create(conf);
this.tikvClient = session.createRawClient();
this.tables = new ConcurrentHashMap<>();
this.refCount = new AtomicInteger(1);
}
@Override
public void open() throws Exception {
// pass
}
@Override
protected boolean opened() {
return this.tikvClient != null;
}
@Override
public Set<String> openedTables() {
return this.tables.keySet();
}
@Override
public synchronized void createTable(String... tables) {
for (String table : tables) {
this.tables.put(table, tableCode++);
}
}
@Override
public synchronized void dropTable(String... tables) {
for (String table : tables) {
this.tables.remove(table);
}
}
@Override
public boolean existsTable(String table) {
return this.tables.containsKey(table);
}
@Override
public final Session session() {
return (Session) super.getOrNewSession();
}
@Override
protected final Session newSession() {
return new StdSession(this.config());
}
@Override
protected synchronized void doClose() {
this.checkValid();
if (this.refCount.decrementAndGet() > 0) {
return;
}
assert this.refCount.get() == 0;
this.tables.clear();
this.tikvClient.close();
}
private void checkValid() {
}
private RawKVClient tikv() {
this.checkValid();
return this.tikvClient;
}
public static final byte[] encode(String string) {
return StringEncoding.encode(string);
}
public static final String decode(byte[] bytes) {
return StringEncoding.decode(bytes);
}
/**
* StdSession implement for tikv
*/
private final class StdSession extends Session {
private Map<ByteString, ByteString> putBatch;
private List<ByteString> deleteBatch;
private Set<ByteString> deletePrefixBatch;
private Map<ByteString, ByteString> deleteRangeBatch;
public StdSession(HugeConfig conf) {
this.putBatch = new HashMap<>();
this.deleteBatch = new ArrayList<>();
this.deletePrefixBatch = new HashSet<>();
this.deleteRangeBatch = new HashMap<>();
}
@Override
public void open() {
this.opened = true;
}
@Override
public void close() {
assert this.closeable();
this.opened = false;
}
@Override
public boolean closed() {
return !this.opened;
}
@Override
public void reset() {
this.putBatch = new HashMap<>();
this.deleteBatch = new ArrayList<>();
this.deletePrefixBatch = new HashSet<>();
this.deleteRangeBatch = new HashMap<>();
}
/**
* Any change in the session
*/
@Override
public boolean hasChanges() {
return this.size() > 0;
}
/**
* Commit all updates(put/delete) to DB
*/
@Override
public Integer commit() {
int count = this.size();
if (count <= 0) {
return 0;
}
if (this.putBatch.size() > 0) {
tikv().batchPutAtomic(this.putBatch);
this.putBatch.clear();
}
if (this.deleteBatch.size() > 0) {
tikv().batchDeleteAtomic(this.deleteBatch);
this.deleteBatch.clear();
}
if (this.deletePrefixBatch.size() > 0) {
for (ByteString key : this.deletePrefixBatch) {
tikv().deletePrefix(key);
}
}
if (this.deleteRangeBatch.size() > 0) {
for (Map.Entry<ByteString, ByteString> entry :
this.deleteRangeBatch.entrySet()) {
tikv().deleteRange(entry.getKey(), entry.getValue());
}
}
return count;
}
/**
* Rollback all updates(put/delete) not committed
*/
@Override
public void rollback() {
this.putBatch.clear();
this.deleteBatch.clear();
this.deletePrefixBatch.clear();
this.deleteRangeBatch.clear();
}
@Override
public Pair<byte[], byte[]> keyRange(String table) {
// TODO: get first key and lastkey of fake tikv table
return null;
}
/**
* Add a KV record to a table
*/
@Override
public void put(String table, byte[] key, byte[] value) {
this.putBatch.put(this.key(table, key), ByteString.copyFrom(value));
}
@Override
public synchronized void increase(String table, byte[] key, byte[] value) {
long old = 0L;
byte[] oldValue = this.get(table, key);
if (oldValue.length != 0) {
old = this.l(oldValue);
}
ByteString newValue = ByteString.copyFrom(
this.b(old + this.l(value)));
tikv().put(this.key(table, key), newValue);
}
@Override
public void delete(String table, byte[] key) {
this.deleteBatch.add(this.key(table, key));
}
@Override
public void deletePrefix(String table, byte[] key) {
ByteString deleteKey = this.key(table, key);
this.deletePrefixBatch.add(deleteKey);
}
/**
* Delete a range of keys from a table
*/
@Override
public void deleteRange(String table, byte[] keyFrom, byte[] keyTo) {
ByteString startKey = this.key(table, keyFrom);
ByteString endKey = this.key(table, keyTo);
this.deleteRangeBatch.put(startKey, endKey);
}
@Override
public byte[] get(String table, byte[] key) {
return tikv().get(this.key(table, key)).toByteArray();
}
@Override
public BackendColumnIterator scan(String table) {
assert !this.hasChanges();
Iterator<Kvrpcpb.KvPair> results = tikv().scanPrefix0(this.key(table));
return new ColumnIterator(table, results);
}
@Override
public BackendColumnIterator scan(String table, byte[] prefix) {
assert !this.hasChanges();
Iterator<Kvrpcpb.KvPair> results = tikv().scanPrefix0(
this.key(table, prefix));
return new ColumnIterator(table, results);
}
@Override
public BackendColumnIterator scan(String table, byte[] keyFrom,
byte[] keyTo, int scanType) {
assert !this.hasChanges();
Iterator<Kvrpcpb.KvPair> results;
if (keyFrom == null) {
results = tikv().scanPrefix0(this.key(table));
} else {
if (keyTo == null) {
results = tikv().scan0(this.key(table, keyFrom), Key.toRawKey(this.key(table)).nextPrefix().toByteString());
} else {
results = tikv().scan0(this.key(table, keyFrom), Key.toRawKey(this.key(table, keyTo)).nextPrefix().toByteString());
}
}
return new ColumnIterator(table, results, keyFrom, keyTo, scanType);
}
private ByteString key(String table) {
byte[] prefix = table.getBytes();
byte[] actualKey = new byte[prefix.length + 1];
System.arraycopy(prefix, 0, actualKey, 0, prefix.length);
actualKey[prefix.length] = (byte) 0xff;
return ByteString.copyFrom(actualKey);
}
private ByteString key(String table, byte[] key) {
byte[] prefix = table.getBytes();
byte[] actualKey = new byte[prefix.length + 1 + key.length];
System.arraycopy(prefix, 0, actualKey, 0, prefix.length);
actualKey[prefix.length] = (byte) 0xff;
System.arraycopy(key, 0, actualKey, prefix.length + 1, key.length);
return ByteString.copyFrom(actualKey);
}
private byte[] b(long value) {
return ByteBuffer.allocate(Long.BYTES)
.order(ByteOrder.nativeOrder())
.putLong(value).array();
}
private long l(byte[] bytes) {
assert bytes.length == Long.BYTES;
return ByteBuffer.wrap(bytes)
.order(ByteOrder.nativeOrder())
.getLong();
}
private int size() {
return this.putBatch.size() + this.deleteBatch.size() +
this.deletePrefixBatch.size() + this.deleteRangeBatch.size();
}
}
private static class ColumnIterator implements BackendColumnIterator,
Countable {
private final String table;
private final Iterator<Kvrpcpb.KvPair> iter;
private final byte[] keyBegin;
private final byte[] keyEnd;
private final int scanType;
private byte[] position;
private byte[] value;
private boolean matched;
public ColumnIterator(String table, Iterator<Kvrpcpb.KvPair> results) {
this(table, results, null, null, 0);
}
public ColumnIterator(String table, Iterator<Kvrpcpb.KvPair> results,
byte[] keyBegin, byte[] keyEnd, int scanType) {
E.checkNotNull(results, "results");
this.table = table;
this.iter = results;
this.keyBegin = keyBegin;
this.keyEnd = keyEnd;
this.scanType = scanType;
this.position = keyBegin;
this.value = null;
this.matched = false;
this.checkArguments();
}
private void checkArguments() {
E.checkArgument(!(this.match(Session.SCAN_PREFIX_BEGIN) &&
this.match(Session.SCAN_PREFIX_END)),
"Can't set SCAN_PREFIX_WITH_BEGIN and " +
"SCAN_PREFIX_WITH_END at the same time");
E.checkArgument(!(this.match(Session.SCAN_PREFIX_BEGIN) &&
this.match(Session.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(Session.SCAN_PREFIX_END) &&
this.match(Session.SCAN_LT_END)),
"Can't set SCAN_PREFIX_WITH_END and " +
"SCAN_LT_END/SCAN_LTE_END at the same time");
if (this.match(Session.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(Session.SCAN_PREFIX_END)) {
E.checkArgument(this.keyEnd != null,
"Parameter `keyEnd` can't be null " +
"if set SCAN_PREFIX_WITH_END");
}
if (this.match(Session.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(Session.SCAN_LT_END)) {
E.checkArgument(this.keyEnd != null,
"Parameter `keyEnd` can't be null " +
"if set SCAN_LT_END or SCAN_LTE_END");
}
}
private boolean match(int expected) {
return Session.matchScanType(expected, this.scanType);
}
@Override
public boolean hasNext() {
// Update position for paging
if (!this.iter.hasNext()) {
return false;
}
Kvrpcpb.KvPair next = this.iter.next();
byte[] tikvKey = next.getKey().toByteArray();
byte[] prefix = this.table.getBytes();
int length = tikvKey.length - prefix.length - 1;
byte[] key = new byte[length];
System.arraycopy(tikvKey, prefix.length + 1, key, 0, length);
this.position = key;
this.value = next.getValue().toByteArray();
// Do filter if not SCAN_ANY
if (!this.match(Session.SCAN_ANY)) {
this.matched = this.filter(this.position);
}
if (!this.matched) {
// The end
this.position = null;
// Free the iterator if finished
this.close();
}
return this.matched;
}
private boolean filter(byte[] key) {
if (this.match(Session.SCAN_PREFIX_BEGIN)) {
/*
* Prefix with `keyBegin`?
* TODO: use custom prefix_extractor instead
* or use ReadOptions.prefix_same_as_start
*/
return Bytes.prefixWith(key, this.keyBegin);
} else if (this.match(Session.SCAN_PREFIX_END)) {
/*
* Prefix with `keyEnd`?
* like the following query for range index:
* key > 'age:20' and prefix with 'age'
*/
assert this.keyEnd != null;
return Bytes.prefixWith(key, this.keyEnd);
} else if (this.match(Session.SCAN_LT_END)) {
/*
* Less (equal) than `keyEnd`?
* NOTE: don't use BytewiseComparator due to signed byte
*/
assert this.keyEnd != null;
if (this.match(Session.SCAN_LTE_END)) {
// Just compare the prefix, can be there are excess tail
key = Arrays.copyOfRange(key, 0, this.keyEnd.length);
return Bytes.compare(key, this.keyEnd) <= 0;
} else {
return Bytes.compare(key, this.keyEnd) < 0;
}
} else {
assert this.match(Session.SCAN_ANY) ||
this.match(Session.SCAN_GT_BEGIN) ||
this.match(Session.SCAN_GTE_BEGIN) :
"Unknow scan type";
return true;
}
}
@Override
public BackendEntry.BackendColumn next() {
if (!this.matched) {
if (!this.hasNext()) {
throw new NoSuchElementException();
}
}
BackendEntry.BackendColumn col = BackendEntry.BackendColumn.of(
this.position, this.value);
this.matched = false;
return col;
}
@Override
public long count() {
long count = 0L;
while (this.hasNext()) {
this.next();
count++;
this.matched = false;
BackendEntryIterator.checkInterrupted();
}
return count;
}
@Override
public byte[] position() {
return this.position;
}
@Override
public void close() {
}
}
}

385
hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvStore.java Normal file
View File

@ -0,0 +1,385 @@
/*
* Copyright 2017 HugeGraph Authors
*
* 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 com.baidu.hugegraph.backend.store.tikv;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import com.baidu.hugegraph.backend.BackendException;
import com.baidu.hugegraph.backend.id.Id;
import com.baidu.hugegraph.backend.query.Query;
import com.baidu.hugegraph.backend.store.AbstractBackendStore;
import com.baidu.hugegraph.backend.store.BackendAction;
import com.baidu.hugegraph.backend.store.BackendEntry;
import com.baidu.hugegraph.backend.store.BackendFeatures;
import com.baidu.hugegraph.backend.store.BackendMutation;
import com.baidu.hugegraph.backend.store.BackendStoreProvider;
import com.baidu.hugegraph.backend.store.tikv.TikvSessions.Session;
import com.baidu.hugegraph.config.HugeConfig;
import com.baidu.hugegraph.exception.ConnectionException;
import com.baidu.hugegraph.type.HugeType;
import com.baidu.hugegraph.util.E;
import com.baidu.hugegraph.util.Log;
public abstract class TikvStore extends AbstractBackendStore<Session> {
private static final Logger LOG = Log.logger(TikvStore.class);
private static final BackendFeatures FEATURES = new TikvFeatures();
private final String store;
private final String namespace;
private final BackendStoreProvider provider;
private final Map<HugeType, TikvTable> tables;
private TikvSessions sessions;
public TikvStore(final BackendStoreProvider provider,
final String namespace, final String store) {
this.tables = new HashMap<>();
this.provider = provider;
this.namespace = namespace;
this.store = store;
this.sessions = null;
this.registerMetaHandlers();
LOG.debug("Store loaded: {}", store);
}
private void registerMetaHandlers() {
this.registerMetaHandler("metrics", (session, meta, args) -> {
TikvMetrics metrics = new TikvMetrics(this.sessions);
return metrics.metrics();
});
}
protected void registerTableManager(HugeType type, TikvTable table) {
this.tables.put(type, table);
}
@Override
protected final TikvTable table(HugeType type) {
assert type != null;
TikvTable table = this.tables.get(type);
if (table == null) {
throw new BackendException("Unsupported table type: %s", type);
}
return table;
}
@Override
protected Session session(HugeType type) {
this.checkOpened();
return this.sessions.session();
}
protected List<String> tableNames() {
return this.tables.values().stream().map(t -> t.table())
.collect(Collectors.toList());
}
public String namespace() {
return this.namespace;
}
@Override
public String store() {
return this.store;
}
@Override
public String database() {
return this.namespace;
}
@Override
public BackendStoreProvider provider() {
return this.provider;
}
@Override
public BackendFeatures features() {
return FEATURES;
}
@Override
public synchronized void open(HugeConfig config) {
E.checkNotNull(config, "config");
if (this.sessions == null) {
this.sessions = new TikvStdSessions(config, this.namespace, this.store);
}
assert this.sessions != null;
if (!this.sessions.closed()) {
LOG.debug("Store {} has been opened before", this.store);
this.sessions.useSession();
return;
}
try {
// NOTE: won't throw error even if connection refused
this.sessions.open();
} catch (Exception e) {
if (!e.getMessage().contains("Column family not found")) {
LOG.error("Failed to open HBase '{}'", this.store, e);
throw new ConnectionException("Failed to connect to HBase", e);
}
if (this.isSchemaStore()) {
LOG.info("Failed to open HBase '{}' with database '{}', " +
"try to init CF later", this.store, this.namespace);
}
}
this.sessions.session();
LOG.debug("Store opened: {}", this.store);
}
@Override
public void close() {
this.checkOpened();
this.sessions.close();
LOG.debug("Store closed: {}", this.store);
}
@Override
public boolean opened() {
this.checkConnectionOpened();
return this.sessions.session().opened();
}
@Override
public void mutate(BackendMutation mutation) {
if (LOG.isDebugEnabled()) {
LOG.debug("Store {} mutation: {}", this.store, mutation);
}
this.checkOpened();
Session session = this.sessions.session();
for (Iterator<BackendAction> it = mutation.mutation(); it.hasNext();) {
this.mutate(session, it.next());
}
}
private void mutate(Session session, BackendAction item) {
BackendEntry entry = item.entry();
TikvTable table = this.table(entry.type());
switch (item.action()) {
case INSERT:
table.insert(session, entry);
break;
case DELETE:
table.delete(session, entry);
break;
case APPEND:
table.append(session, entry);
break;
case ELIMINATE:
table.eliminate(session, entry);
break;
default:
throw new AssertionError(String.format(
"Unsupported mutate action: %s", item.action()));
}
}
@Override
public Iterator<BackendEntry> query(Query query) {
this.checkOpened();
Session session = this.sessions.session();
TikvTable table = this.table(TikvTable.tableType(query));
return table.query(session, query);
}
@Override
public Number queryNumber(Query query) {
this.checkOpened();
Session session = this.sessions.session();
TikvTable table = this.table(TikvTable.tableType(query));
return table.queryNumber(session, query);
}
@Override
public void init() {
LOG.debug("Store initialized: {}", this.store);
}
@Override
public void clear(boolean clearSpace) {
LOG.debug("Store cleared: {}", this.store);
}
@Override
public boolean initialized() {
return true;
}
@Override
public void truncate() {
LOG.debug("Store truncated: {}", this.store);
}
@Override
public void beginTx() {
// pass
}
@Override
public void commitTx() {
this.checkOpened();
Session session = this.sessions.session();
session.commit();
}
@Override
public void rollbackTx() {
this.checkOpened();
Session session = this.sessions.session();
session.rollback();
}
private final void checkConnectionOpened() {
}
/***************************** Store defines *****************************/
public static class TikvSchemaStore extends TikvStore {
private final TikvTables.Counters counters;
public TikvSchemaStore(BackendStoreProvider provider,
String namespace, String store) {
super(provider, namespace, store);
this.counters = new TikvTables.Counters(namespace);
registerTableManager(HugeType.VERTEX_LABEL,
new TikvTables.VertexLabel(namespace));
registerTableManager(HugeType.EDGE_LABEL,
new TikvTables.EdgeLabel(namespace));
registerTableManager(HugeType.PROPERTY_KEY,
new TikvTables.PropertyKey(namespace));
registerTableManager(HugeType.INDEX_LABEL,
new TikvTables.IndexLabel(namespace));
registerTableManager(HugeType.SECONDARY_INDEX,
new TikvTables.SecondaryIndex(store));
}
@Override
protected List<String> tableNames() {
List<String> tableNames = super.tableNames();
tableNames.add(this.counters.table());
return tableNames;
}
@Override
public void increaseCounter(HugeType type, long increment) {
super.checkOpened();
this.counters.increaseCounter(super.sessions.session(),
type, increment);
}
@Override
public long getCounter(HugeType type) {
super.checkOpened();
return this.counters.getCounter(super.sessions.session(), type);
}
@Override
public boolean isSchemaStore() {
return true;
}
}
public static class TikvGraphStore extends TikvStore {
public TikvGraphStore(BackendStoreProvider provider,
String namespace, String store) {
super(provider, namespace, store);
registerTableManager(HugeType.VERTEX,
new TikvTables.Vertex(store));
registerTableManager(HugeType.EDGE_OUT,
TikvTables.Edge.out(store));
registerTableManager(HugeType.EDGE_IN,
TikvTables.Edge.in(store));
registerTableManager(HugeType.SECONDARY_INDEX,
new TikvTables.SecondaryIndex(store));
registerTableManager(HugeType.VERTEX_LABEL_INDEX,
new TikvTables.VertexLabelIndex(store));
registerTableManager(HugeType.EDGE_LABEL_INDEX,
new TikvTables.EdgeLabelIndex(store));
registerTableManager(HugeType.RANGE_INT_INDEX,
new TikvTables.RangeIntIndex(store));
registerTableManager(HugeType.RANGE_FLOAT_INDEX,
new TikvTables.RangeFloatIndex(store));
registerTableManager(HugeType.RANGE_LONG_INDEX,
new TikvTables.RangeLongIndex(store));
registerTableManager(HugeType.RANGE_DOUBLE_INDEX,
new TikvTables.RangeDoubleIndex(store));
registerTableManager(HugeType.SEARCH_INDEX,
new TikvTables.SearchIndex(store));
registerTableManager(HugeType.SHARD_INDEX,
new TikvTables.ShardIndex(store));
registerTableManager(HugeType.UNIQUE_INDEX,
new TikvTables.UniqueIndex(store));
}
@Override
public boolean isSchemaStore() {
return false;
}
@Override
public Id nextId(HugeType type) {
throw new UnsupportedOperationException(
"TikvGraphStore.nextId()");
}
@Override
public void increaseCounter(HugeType type, long num) {
throw new UnsupportedOperationException(
"TikvGraphStore.increaseCounter()");
}
@Override
public long getCounter(HugeType type) {
throw new UnsupportedOperationException(
"TikvGraphStore.getCounter()");
}
}
}

56
hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvStoreProvider.java Normal file
View File

@ -0,0 +1,56 @@
/*
* Copyright 2017 HugeGraph Authors
*
* 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 com.baidu.hugegraph.backend.store.tikv;
import com.baidu.hugegraph.backend.store.AbstractBackendStoreProvider;
import com.baidu.hugegraph.backend.store.BackendStore;
import com.baidu.hugegraph.backend.store.tikv.TikvStore.TikvGraphStore;
import com.baidu.hugegraph.backend.store.tikv.TikvStore.TikvSchemaStore;
public class TikvStoreProvider extends AbstractBackendStoreProvider {
protected String namespace() {
return this.graph().toLowerCase();
}
@Override
protected BackendStore newSchemaStore(String store) {
return new TikvSchemaStore(this, this.namespace(), store);
}
@Override
protected BackendStore newGraphStore(String store) {
return new TikvGraphStore(this, this.namespace(), store);
}
@Override
public String type() {
return "tikv";
}
@Override
public String version() {
/*
* Versions history:
* [1.0] HugeGraph-1328: supports tikv
*/
return "1.0";
}
}

334
hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvTable.java Normal file
View File

@ -0,0 +1,334 @@
/*
* Copyright 2017 HugeGraph Authors
*
* 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 com.baidu.hugegraph.backend.store.tikv;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import org.slf4j.Logger;
import com.baidu.hugegraph.backend.id.Id;
import com.baidu.hugegraph.backend.page.PageState;
import com.baidu.hugegraph.backend.query.Aggregate;
import com.baidu.hugegraph.backend.query.Aggregate.AggregateFunc;
import com.baidu.hugegraph.backend.query.Condition.Relation;
import com.baidu.hugegraph.backend.query.ConditionQuery;
import com.baidu.hugegraph.backend.query.IdPrefixQuery;
import com.baidu.hugegraph.backend.query.IdRangeQuery;
import com.baidu.hugegraph.backend.query.Query;
import com.baidu.hugegraph.backend.serializer.BinaryBackendEntry;
import com.baidu.hugegraph.backend.serializer.BinaryEntryIterator;
import com.baidu.hugegraph.backend.store.BackendEntry;
import com.baidu.hugegraph.backend.store.BackendEntry.BackendColumn;
import com.baidu.hugegraph.backend.store.BackendEntry.BackendColumnIterator;
import com.baidu.hugegraph.backend.store.BackendEntry.BackendColumnIteratorWrapper;
import com.baidu.hugegraph.backend.store.BackendEntryIterator;
import com.baidu.hugegraph.backend.store.BackendTable;
import com.baidu.hugegraph.backend.store.Shard;
import com.baidu.hugegraph.backend.store.tikv.TikvSessions.Countable;
import com.baidu.hugegraph.backend.store.tikv.TikvSessions.Session;
import com.baidu.hugegraph.exception.NotSupportException;
import com.baidu.hugegraph.iterator.FlatMapperIterator;
import com.baidu.hugegraph.type.HugeType;
import com.baidu.hugegraph.util.Bytes;
import com.baidu.hugegraph.util.E;
import com.baidu.hugegraph.util.Log;
import com.baidu.hugegraph.util.StringEncoding;
public class TikvTable extends BackendTable<Session, BackendEntry> {
private static final Logger LOG = Log.logger(TikvStore.class);
private final RocksDBShardSpliter shardSpliter;
public TikvTable(String database, String table) {
super(String.format("%s+%s", database, table));
this.shardSpliter = new RocksDBShardSpliter(this.table());
}
@Override
protected void registerMetaHandlers() {
this.registerMetaHandler("splits", (session, meta, args) -> {
E.checkArgument(args.length == 1,
"The args count of %s must be 1", meta);
long splitSize = (long) args[0];
return this.shardSpliter.getSplits(session, splitSize);
});
}
@Override
public void init(Session session) {
// pass
}
@Override
public void clear(Session session) {
// pass
}
@Override
public void insert(Session session, BackendEntry entry) {
assert !entry.columns().isEmpty();
for (BackendColumn col : entry.columns()) {
assert entry.belongToMe(col) : entry;
session.put(this.table(), col.name, col.value);
}
}
@Override
public void delete(Session session, BackendEntry entry) {
if (entry.columns().isEmpty()) {
session.delete(this.table(), entry.id().asBytes());
} else {
for (BackendColumn col : entry.columns()) {
assert entry.belongToMe(col) : entry;
session.delete(this.table(), col.name);
}
}
}
@Override
public void append(Session session, BackendEntry entry) {
assert entry.columns().size() == 1;
this.insert(session, entry);
}
@Override
public void eliminate(Session session, BackendEntry entry) {
assert entry.columns().size() == 1;
this.delete(session, entry);
}
@Override
public Number queryNumber(Session session, Query query) {
Aggregate aggregate = query.aggregateNotNull();
if (aggregate.func() != AggregateFunc.COUNT) {
throw new NotSupportException(aggregate.toString());
}
assert aggregate.func() == AggregateFunc.COUNT;
assert query.noLimit();
Iterator<BackendColumn> results = this.queryBy(session, query);
if (results instanceof Countable) {
return ((Countable) results).count();
}
return IteratorUtils.count(results);
}
@Override
public Iterator<BackendEntry> query(Session session, Query query) {
if (query.limit() == 0L && !query.noLimit()) {
LOG.debug("Return empty result(limit=0) for query {}", query);
return Collections.emptyIterator();
}
return newEntryIterator(this.queryBy(session, query), query);
}
protected BackendColumnIterator queryBy(Session session, Query query) {
// Query all
if (query.empty()) {
return this.queryAll(session, query);
}
// Query by prefix
if (query instanceof IdPrefixQuery) {
IdPrefixQuery pq = (IdPrefixQuery) query;
return this.queryByPrefix(session, pq);
}
// Query by range
if (query instanceof IdRangeQuery) {
IdRangeQuery rq = (IdRangeQuery) query;
return this.queryByRange(session, rq);
}
// Query by id
if (query.conditions().isEmpty()) {
assert !query.ids().isEmpty();
// NOTE: this will lead to lazy create rocksdb iterator
return new BackendColumnIteratorWrapper(new FlatMapperIterator<>(
query.ids().iterator(), id -> this.queryById(session, id)
));
}
// Query by condition (or condition + id)
ConditionQuery cq = (ConditionQuery) query;
return this.queryByCond(session, cq);
}
protected BackendColumnIterator queryAll(Session session, Query query) {
if (query.paging()) {
PageState page = PageState.fromString(query.page());
byte[] begin = page.position();
return session.scan(this.table(), begin, null, Session.SCAN_ANY);
} else {
return session.scan(this.table());
}
}
protected BackendColumnIterator queryById(Session session, Id id) {
// TODO: change to get() after vertex and schema don't use id prefix
return session.scan(this.table(), id.asBytes());
}
protected BackendColumnIterator getById(Session session, Id id) {
byte[] value = session.get(this.table(), id.asBytes());
if (value.length == 0) {
return BackendColumnIterator.empty();
}
BackendColumn col = BackendColumn.of(id.asBytes(), value);
return new BackendEntry.BackendColumnIteratorWrapper(col);
}
protected BackendColumnIterator queryByPrefix(Session session,
IdPrefixQuery query) {
int type = query.inclusiveStart() ?
Session.SCAN_GTE_BEGIN : Session.SCAN_GT_BEGIN;
type |= Session.SCAN_PREFIX_END;
return session.scan(this.table(), query.start().asBytes(),
query.prefix().asBytes(), type);
}
protected BackendColumnIterator queryByRange(Session session,
IdRangeQuery query) {
byte[] start = query.start().asBytes();
byte[] end = query.end() == null ? null : query.end().asBytes();
int type = query.inclusiveStart() ?
Session.SCAN_GTE_BEGIN : Session.SCAN_GT_BEGIN;
if (end != null) {
type |= query.inclusiveEnd() ?
Session.SCAN_LTE_END : Session.SCAN_LT_END;
}
return session.scan(this.table(), start, end, type);
}
protected BackendColumnIterator queryByCond(Session session,
ConditionQuery query) {
if (query.containsScanCondition()) {
E.checkArgument(query.relations().size() == 1,
"Invalid scan with multi conditions: %s", query);
Relation scan = query.relations().iterator().next();
Shard shard = (Shard) scan.value();
return this.queryByRange(session, shard, query.page());
}
throw new NotSupportException("query: %s", query);
}
protected BackendColumnIterator queryByRange(Session session, Shard shard,
String page) {
byte[] start = this.shardSpliter.position(shard.start());
byte[] end = this.shardSpliter.position(shard.end());
if (page != null && !page.isEmpty()) {
byte[] position = PageState.fromString(page).position();
E.checkArgument(start == null ||
Bytes.compare(position, start) >= 0,
"Invalid page out of lower bound");
start = position;
}
if (start == null) {
start = ShardSpliter.START_BYTES;
}
int type = Session.SCAN_GTE_BEGIN;
if (end != null) {
type |= Session.SCAN_LT_END;
}
return session.scan(this.table(), start, end, type);
}
protected static final BackendEntryIterator newEntryIterator(
BackendColumnIterator cols,
Query query) {
return new BinaryEntryIterator<>(cols, query, (entry, col) -> {
if (entry == null || !entry.belongToMe(col)) {
HugeType type = query.resultType();
// NOTE: only support BinaryBackendEntry currently
entry = new BinaryBackendEntry(type, col.name);
}
entry.columns(col);
return entry;
});
}
protected static final long sizeOfBackendEntry(BackendEntry entry) {
return BinaryEntryIterator.sizeOfEntry(entry);
}
private static class RocksDBShardSpliter extends ShardSpliter<Session> {
private static final String MEM_SIZE = "rocksdb.size-all-mem-tables";
private static final String SST_SIZE = "rocksdb.total-sst-files-size";
private static final String NUM_KEYS = "rocksdb.estimate-num-keys";
public RocksDBShardSpliter(String table) {
super(table);
}
@Override
public List<Shard> getSplits(Session session, long splitSize) {
E.checkArgument(splitSize >= MIN_SHARD_SIZE,
"The split-size must be >= %s bytes, but got %s",
MIN_SHARD_SIZE, splitSize);
Pair<byte[], byte[]> keyRange = session.keyRange(this.table());
if (keyRange == null || keyRange.getRight() == null) {
return super.getSplits(session, splitSize);
}
long size = this.estimateDataSize(session);
if (size <= 0) {
size = this.estimateNumKeys(session) * ESTIMATE_BYTES_PER_KV;
}
double count = Math.ceil(size / (double) splitSize);
if (count <= 0) {
count = 1;
}
Range range = new Range(keyRange.getLeft(),
Range.increase(keyRange.getRight()));
List<Shard> splits = new ArrayList<>((int) count);
splits.addAll(range.splitEven((int) count));
return splits;
}
@Override
public long estimateDataSize(Session session) {
return 1L;
}
@Override
public long estimateNumKeys(Session session) {
return 1L;
}
@Override
public byte[] position(String position) {
if (END.equals(position)) {
return null;
}
return StringEncoding.decodeBase64(position);
}
}
}

326
hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvTables.java Normal file
View File

@ -0,0 +1,326 @@
/*
* Copyright 2017 HugeGraph Authors
*
* 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 com.baidu.hugegraph.backend.store.tikv;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.List;
import com.baidu.hugegraph.backend.id.Id;
import com.baidu.hugegraph.backend.query.Condition;
import com.baidu.hugegraph.backend.query.Condition.Relation;
import com.baidu.hugegraph.backend.query.ConditionQuery;
import com.baidu.hugegraph.backend.serializer.BinarySerializer;
import com.baidu.hugegraph.backend.store.BackendEntry;
import com.baidu.hugegraph.backend.store.BackendEntry.BackendColumnIterator;
import com.baidu.hugegraph.backend.store.tikv.TikvSessions.Session;
import com.baidu.hugegraph.type.HugeType;
import com.baidu.hugegraph.type.define.HugeKeys;
import com.baidu.hugegraph.util.E;
public class TikvTables {
public static class Counters extends TikvTable {
private static final String TABLE = HugeType.COUNTER.string();
public Counters(String database) {
super(database, TABLE);
}
public long getCounter(Session session, HugeType type) {
byte[] key = new byte[]{type.code()};
byte[] value = session.get(this.table(), key);
if (value.length != 0) {
return l(value);
} else {
return 0L;
}
}
public void increaseCounter(Session session, HugeType type,
long increment) {
byte[] key = new byte[]{type.code()};
session.increase(this.table(), key, b(increment));
}
private static byte[] b(long value) {
return ByteBuffer.allocate(Long.BYTES)
.order(ByteOrder.nativeOrder())
.putLong(value).array();
}
private static long l(byte[] bytes) {
assert bytes.length == Long.BYTES;
return ByteBuffer.wrap(bytes)
.order(ByteOrder.nativeOrder())
.getLong();
}
}
public static class VertexLabel extends TikvTable {
public static final String TABLE = HugeType.VERTEX_LABEL.string();
public VertexLabel(String database) {
super(database, TABLE);
}
}
public static class EdgeLabel extends TikvTable {
public static final String TABLE = HugeType.EDGE_LABEL.string();
public EdgeLabel(String database) {
super(database, TABLE);
}
}
public static class PropertyKey extends TikvTable {
public static final String TABLE = HugeType.PROPERTY_KEY.string();
public PropertyKey(String database) {
super(database, TABLE);
}
}
public static class IndexLabel extends TikvTable {
public static final String TABLE = HugeType.INDEX_LABEL.string();
public IndexLabel(String database) {
super(database, TABLE);
}
}
public static class Vertex extends TikvTable {
public static final String TABLE = HugeType.VERTEX.string();
public Vertex(String database) {
super(database, TABLE);
}
@Override
protected BackendColumnIterator queryById(Session session, Id id) {
return this.getById(session, id);
}
}
public static class Edge extends TikvTable {
public static final String TABLE_SUFFIX = HugeType.EDGE.string();
public Edge(boolean out, String database) {
// Edge out/in table
super(database, (out ? 'o' : 'i') + TABLE_SUFFIX);
}
public static Edge out(String database) {
return new Edge(true, database);
}
public static Edge in(String database) {
return new Edge(false, database);
}
@Override
protected BackendColumnIterator queryById(Session session, Id id) {
return this.getById(session, id);
}
}
public static class IndexTable extends TikvTable {
public IndexTable(String database, String table) {
super(database, table);
}
@Override
public void eliminate(Session session, BackendEntry entry) {
assert entry.columns().size() == 1;
super.delete(session, entry);
}
@Override
public void delete(Session session, BackendEntry entry) {
/*
* Only delete index by label will come here
* Regular index delete will call eliminate()
*/
for (BackendEntry.BackendColumn column : entry.columns()) {
// Don't assert entry.belongToMe(column), length-prefix is 1*
session.delete(this.table(), column.name);
}
}
}
public static class SecondaryIndex extends IndexTable {
public static final String TABLE = HugeType.SECONDARY_INDEX.string();
public SecondaryIndex(String database) {
super(database, TABLE);
}
}
public static class VertexLabelIndex extends IndexTable {
public static final String TABLE = HugeType.VERTEX_LABEL_INDEX.string();
public VertexLabelIndex(String database) {
super(database, TABLE);
}
}
public static class EdgeLabelIndex extends IndexTable {
public static final String TABLE = HugeType.EDGE_LABEL_INDEX.string();
public EdgeLabelIndex(String database) {
super(database, TABLE);
}
}
public static class SearchIndex extends IndexTable {
public static final String TABLE = HugeType.SEARCH_INDEX.string();
public SearchIndex(String database) {
super(database, TABLE);
}
}
public static class UniqueIndex extends IndexTable {
public static final String TABLE = HugeType.UNIQUE_INDEX.string();
public UniqueIndex(String database) {
super(database, TABLE);
}
}
public static class RangeIndex extends IndexTable {
public RangeIndex(String database, String table) {
super(database, table);
}
@Override
protected BackendColumnIterator queryByCond(Session session,
ConditionQuery query) {
assert !query.conditions().isEmpty();
List<Condition> conds = query.syspropConditions(HugeKeys.ID);
E.checkArgument(!conds.isEmpty(),
"Please specify the index conditions");
Id prefix = null;
Id min = null;
boolean minEq = false;
Id max = null;
boolean maxEq = false;
for (Condition c : conds) {
Relation r = (Relation) c;
switch (r.relation()) {
case PREFIX:
prefix = (Id) r.value();
break;
case GTE:
minEq = true;
case GT:
min = (Id) r.value();
break;
case LTE:
maxEq = true;
case LT:
max = (Id) r.value();
break;
default:
E.checkArgument(false, "Unsupported relation '%s'",
r.relation());
}
}
E.checkArgumentNotNull(min, "Range index begin key is missing");
byte[] begin = min.asBytes();
if (!minEq) {
begin = BinarySerializer.increaseOne(begin);
}
if (max == null) {
E.checkArgumentNotNull(prefix, "Range index prefix is missing");
return session.scan(this.table(), begin, prefix.asBytes(),
Session.SCAN_PREFIX_END);
} else {
byte[] end = max.asBytes();
int type = maxEq ? Session.SCAN_LTE_END : Session.SCAN_LT_END;
return session.scan(this.table(), begin, end, type);
}
}
}
public static class RangeIntIndex extends RangeIndex {
public static final String TABLE = HugeType.RANGE_INT_INDEX.string();
public RangeIntIndex(String store) {
super(store, TABLE);
}
}
public static class RangeFloatIndex extends RangeIndex{
public static final String TABLE = HugeType.RANGE_FLOAT_INDEX.string();
public RangeFloatIndex(String store) {
super(store, TABLE);
}
}
public static class RangeLongIndex extends RangeIndex {
public static final String TABLE = HugeType.RANGE_LONG_INDEX.string();
public RangeLongIndex(String store) {
super(store, TABLE);
}
}
public static class RangeDoubleIndex extends RangeIndex{
public static final String TABLE = HugeType.RANGE_DOUBLE_INDEX.string();
public RangeDoubleIndex(String store) {
super(store, TABLE);
}
}
public static class ShardIndex extends RangeIndex {
public static final String TABLE = HugeType.SHARD_INDEX.string();
public ShardIndex(String database) {
super(database, TABLE);
}
}
}

1
pom.xml
View File

@ -120,6 +120,7 @@
<module>hugegraph-palo</module>
<module>hugegraph-hbase</module>
<module>hugegraph-postgresql</module>
<module>hugegraph-tikv</module>
</modules>
<dependencyManagement>