diff --git a/hugegraph-dist/pom.xml b/hugegraph-dist/pom.xml index 60e770b3f..11db58083 100644 --- a/hugegraph-dist/pom.xml +++ b/hugegraph-dist/pom.xml @@ -64,6 +64,11 @@ hugegraph-postgresql ${project.version} + + com.baidu.hugegraph + hugegraph-tikv + ${project.version} + org.apache.tinkerpop diff --git a/hugegraph-dist/src/assembly/static/conf/log4j2.xml b/hugegraph-dist/src/assembly/static/conf/log4j2.xml index fae6721d8..fb97ddb3c 100644 --- a/hugegraph-dist/src/assembly/static/conf/log4j2.xml +++ b/hugegraph-dist/src/assembly/static/conf/log4j2.xml @@ -98,5 +98,8 @@ + + + diff --git a/hugegraph-dist/src/main/java/com/baidu/hugegraph/dist/RegisterUtil.java b/hugegraph-dist/src/main/java/com/baidu/hugegraph/dist/RegisterUtil.java index 8ce07f7b2..94e0dca99 100644 --- a/hugegraph-dist/src/main/java/com/baidu/hugegraph/dist/RegisterUtil.java +++ b/hugegraph-dist/src/main/java/com/baidu/hugegraph/dist/RegisterUtil.java @@ -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"); diff --git a/hugegraph-dist/src/main/resources/backend.properties b/hugegraph-dist/src/main/resources/backend.properties index 45f95d2f1..3786cf132 100644 --- a/hugegraph-dist/src/main/resources/backend.properties +++ b/hugegraph-dist/src/main/resources/backend.properties @@ -1 +1 @@ -backends=[cassandra, scylladb, rocksdb, mysql, palo, hbase, postgresql] +backends=[cassandra, scylladb, rocksdb, mysql, palo, hbase, postgresql, tikv] diff --git a/hugegraph-tikv/pom.xml b/hugegraph-tikv/pom.xml new file mode 100644 index 000000000..e6919feb0 --- /dev/null +++ b/hugegraph-tikv/pom.xml @@ -0,0 +1,26 @@ + + + + hugegraph + com.baidu.hugegraph + 0.12.0 + + 4.0.0 + + hugegraph-tikv + + + + com.baidu.hugegraph + hugegraph-core + ${project.version} + + + org.tikv + tikv-client-java + 3.1.0-SNAPSHOT + + + diff --git a/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvFeatures.java b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvFeatures.java new file mode 100644 index 000000000..a6e39f100 --- /dev/null +++ b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvFeatures.java @@ -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; + } +} diff --git a/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvMetrics.java b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvMetrics.java new file mode 100644 index 000000000..030bdbbba --- /dev/null +++ b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvMetrics.java @@ -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 metrics() { + return null; + } +} diff --git a/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvOptions.java b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvOptions.java new file mode 100644 index 000000000..3e4dfff41 --- /dev/null +++ b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvOptions.java @@ -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 TIKV_PDS = + new ConfigOption<>( + "tikv.pds", + "The addresses of Tikv pds, separated with commas.", + disallowEmpty(), + "localhost" + ); + + public static final ConfigOption 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 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 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 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 TIKV_DELETE_RANGE_CONCURRENCY = + new ConfigOption<>( + "tikv.delete_range_concurrency", + "The number of thread pool size for delete range of tikv client.", + disallowEmpty(), + 20 + ); +} diff --git a/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvSessions.java b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvSessions.java new file mode 100644 index 000000000..1d592009b --- /dev/null +++ b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvSessions.java @@ -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 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 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(); + } +} diff --git a/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvStdSessions.java b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvStdSessions.java new file mode 100644 index 000000000..c3290e165 --- /dev/null +++ b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvStdSessions.java @@ -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 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 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 putBatch; + private List deleteBatch; + private Set deletePrefixBatch; + private Map 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 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 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 results = tikv().scanPrefix0(this.key(table)); + return new ColumnIterator(table, results); + } + + @Override + public BackendColumnIterator scan(String table, byte[] prefix) { + assert !this.hasChanges(); + Iterator 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 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 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 results) { + this(table, results, null, null, 0); + } + + public ColumnIterator(String table, Iterator 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() { + } + } +} diff --git a/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvStore.java b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvStore.java new file mode 100644 index 000000000..176bc1c87 --- /dev/null +++ b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvStore.java @@ -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 { + + 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 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 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 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 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 tableNames() { + List 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()"); + } + } +} diff --git a/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvStoreProvider.java b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvStoreProvider.java new file mode 100644 index 000000000..d7f7c76d2 --- /dev/null +++ b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvStoreProvider.java @@ -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"; + } +} diff --git a/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvTable.java b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvTable.java new file mode 100644 index 000000000..b169b2a15 --- /dev/null +++ b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvTable.java @@ -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 { + + 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 results = this.queryBy(session, query); + if (results instanceof Countable) { + return ((Countable) results).count(); + } + return IteratorUtils.count(results); + } + + @Override + public Iterator 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 { + + 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 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 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 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); + } + } +} diff --git a/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvTables.java b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvTables.java new file mode 100644 index 000000000..0186047da --- /dev/null +++ b/hugegraph-tikv/src/main/java/com/baidu/hugegraph/backend/store/tikv/TikvTables.java @@ -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 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); + } + } +} diff --git a/pom.xml b/pom.xml index bbddbf6d0..5476258ed 100644 --- a/pom.xml +++ b/pom.xml @@ -120,6 +120,7 @@ hugegraph-palo hugegraph-hbase hugegraph-postgresql + hugegraph-tikv