support tikv 1st version

Change-Id: I23fd71ec977a1944a02ea726717fc8b3c8659a15

hugegraph-dist/pom.xml

@@ -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>

hugegraph-dist/src/assembly/static/conf/log4j2.xml

@@ -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>

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");

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]

hugegraph-tikv/pom.xml

@@ -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>

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;
+    }
+}

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<String, Object> metrics() {
+        return null;
+    }
+}

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<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
+            );
+}

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<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();
+    }
+}

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<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() {
+        }
+    }
+}

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<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()");
+        }
+    }
+}

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";
+    }
+}

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<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);
+        }
+    }
+}

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<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);
+        }
+    }
+}

pom.xml

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