From c50c34a9042f115d1a8d5e53b27fc52aa6df5f41 Mon Sep 17 00:00:00 2001
From: Linary <liningrui@vip.qq.com>
Date: Wed, 27 Mar 2019 16:22:34 +0800
Subject: [PATCH] Support pagination when single index query (#328)

Support pagination when single index query

Change-Id: Iab355a9f68cebd8ebc666cf126842eb19fa40240
---
 hugegraph-api/pom.xml                         |   4 +-
 .../baidu/hugegraph/api/graph/EdgeAPI.java    |  13 +-
 .../baidu/hugegraph/api/graph/VertexAPI.java  |  13 +-
 .../baidu/hugegraph/api/job/GremlinAPI.java   |   3 +-
 .../baidu/hugegraph/version/ApiVersion.java   |   3 +-
 hugegraph-cassandra/pom.xml                   |   2 +-
 .../store/cassandra/CassandraTable.java       |  47 +-
 hugegraph-core/pom.xml                        |   4 +-
 .../hugegraph/backend/page/IdHolder.java      | 103 ++++
 .../hugegraph/backend/page/IdHolderList.java  |  71 +++
 .../backend/page/PageEntryIterator.java       | 107 ++++
 .../baidu/hugegraph/backend/page/PageIds.java |  50 ++
 .../hugegraph/backend/page/PageState.java     |  97 +++
 .../hugegraph/backend/page/QueryList.java     | 262 ++++++++
 .../backend/query/ConditionQuery.java         |   1 -
 .../backend/query/ConditionQueryFlatten.java  |  14 +-
 .../backend/query/IdPrefixQuery.java          |   4 +
 .../baidu/hugegraph/backend/query/Query.java  |   3 +-
 .../serializer/BinaryInlineSerializer.java    |   2 +-
 .../backend/serializer/BinarySerializer.java  |  70 ++-
 .../backend/store/BackendStoreSystemInfo.java |   5 +-
 .../backend/tx/AbstractTransaction.java       |   1 +
 .../backend/tx/GraphIndexTransaction.java     | 189 +++---
 .../backend/tx/GraphTransaction.java          |  42 +-
 .../baidu/hugegraph/config/CoreOptions.java   |   8 +
 .../baidu/hugegraph/version/CoreVersion.java  |   2 +-
 hugegraph-dist/pom.xml                        |   2 +-
 hugegraph-example/pom.xml                     |   2 +-
 hugegraph-hbase/pom.xml                       |   2 +-
 .../backend/store/hbase/HbaseSerializer.java  |   2 +-
 .../backend/store/hbase/HbaseSessions.java    |  11 +
 .../store/hbase/HbaseStoreProvider.java       |   3 +-
 .../backend/store/hbase/HbaseTable.java       |  12 +-
 .../backend/store/hbase/HbaseTables.java      |  39 +-
 hugegraph-mysql/pom.xml                       |   2 +-
 .../backend/store/mysql/MysqlTable.java       |   3 +
 .../backend/store/mysql/MysqlTables.java      |  21 -
 hugegraph-palo/pom.xml                        |   2 +-
 hugegraph-rocksdb/pom.xml                     |   2 +-
 .../backend/store/rocksdb/RocksDBTable.java   |   3 +-
 hugegraph-scylladb/pom.xml                    |   2 +-
 hugegraph-test/pom.xml                        |   2 +-
 .../baidu/hugegraph/core/VertexCoreTest.java  | 570 +++++++++++++++++-
 .../hugegraph/core/VertexLabelCoreTest.java   |   4 +-
 .../src/main/resources/hugegraph.properties   |   2 +
 pom.xml                                       |   2 +-
 46 files changed, 1584 insertions(+), 224 deletions(-)
 create mode 100644 hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/IdHolder.java
 create mode 100644 hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/IdHolderList.java
 create mode 100644 hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/PageEntryIterator.java
 create mode 100644 hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/PageIds.java
 create mode 100644 hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/PageState.java
 create mode 100644 hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/QueryList.java

diff --git a/hugegraph-api/pom.xml b/hugegraph-api/pom.xml
index 74ac2d4195..7082c4ad44 100644
--- a/hugegraph-api/pom.xml
+++ b/hugegraph-api/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hugegraph</artifactId>
         <groupId>com.baidu.hugegraph</groupId>
-        <version>0.9.1</version>
+        <version>0.9.2</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
@@ -86,7 +86,7 @@
                             </addDefaultSpecificationEntries>
                         </manifest>
                         <manifestEntries>
-                            <Implementation-Version>0.34.0.0</Implementation-Version>
+                            <Implementation-Version>0.35.0.0</Implementation-Version>
                         </manifestEntries>
                     </archive>
                 </configuration>
diff --git a/hugegraph-api/src/main/java/com/baidu/hugegraph/api/graph/EdgeAPI.java b/hugegraph-api/src/main/java/com/baidu/hugegraph/api/graph/EdgeAPI.java
index 75ed3a4761..ba54eba04d 100644
--- a/hugegraph-api/src/main/java/com/baidu/hugegraph/api/graph/EdgeAPI.java
+++ b/hugegraph-api/src/main/java/com/baidu/hugegraph/api/graph/EdgeAPI.java
@@ -222,17 +222,20 @@ public String list(@Context GraphManager manager,
         LOG.debug("Graph [{}] query edges by vertex: {}, direction: {}, " +
                   "label: {}, properties: {}, offset: {}, page: {}, limit: {}",
                   vertexId, direction, label, properties, offset, page, limit);
+
+        Map<String, Object> props = parseProperties(properties);
         if (page != null) {
             E.checkArgument(vertexId == null && direction == null &&
-                            label == null && properties == null && offset == 0,
-                            "Not support quering edges based on paging and " +
-                            "[vertex, direction, label, properties, offset] " +
-                            "together");
+                            offset == 0,
+                            "Not support querying edges based on paging " +
+                            "and [vertex, direction, offset] together");
+            E.checkArgument(props.size() <= 1,
+                            "Not support querying edges based on paging " +
+                            "and more than one property");
         }
 
         Id vertex = VertexAPI.checkAndParseVertexId(vertexId);
         Direction dir = parseDirection(direction);
-        Map<String, Object> props = parseProperties(properties);
 
         HugeGraph g = graph(manager, graph);
 
diff --git a/hugegraph-api/src/main/java/com/baidu/hugegraph/api/graph/VertexAPI.java b/hugegraph-api/src/main/java/com/baidu/hugegraph/api/graph/VertexAPI.java
index 8479f46712..ee0823c01e 100644
--- a/hugegraph-api/src/main/java/com/baidu/hugegraph/api/graph/VertexAPI.java
+++ b/hugegraph-api/src/main/java/com/baidu/hugegraph/api/graph/VertexAPI.java
@@ -173,13 +173,16 @@ public String list(@Context GraphManager manager,
         LOG.debug("Graph [{}] query vertices by label: {}, properties: {}, " +
                   "offset: {}, page: {}, limit: {}",
                   graph, label, properties, offset, page, limit);
-        if (page != null) {
-            E.checkArgument(label == null && properties == null && offset == 0,
-                            "Not support quering vertices based on paging " +
-                            "and [label, properties, offset] together");
-        }
 
         Map<String, Object> props = parseProperties(properties);
+        if (page != null) {
+            E.checkArgument(offset == 0,
+                            "Not support querying vertices based on paging " +
+                            "and offset together");
+            E.checkArgument(props.size() <= 1,
+                            "Not support querying vertices based on paging " +
+                            "and more than one property");
+        }
 
         HugeGraph g = graph(manager, graph);
 
diff --git a/hugegraph-api/src/main/java/com/baidu/hugegraph/api/job/GremlinAPI.java b/hugegraph-api/src/main/java/com/baidu/hugegraph/api/job/GremlinAPI.java
index 5f76c7f7e3..46c3988cd5 100644
--- a/hugegraph-api/src/main/java/com/baidu/hugegraph/api/job/GremlinAPI.java
+++ b/hugegraph-api/src/main/java/com/baidu/hugegraph/api/job/GremlinAPI.java
@@ -59,8 +59,7 @@
 import com.codahale.metrics.Histogram;
 import com.codahale.metrics.annotation.Timed;
 import com.fasterxml.jackson.annotation.JsonProperty;
-
-import jersey.repackaged.com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMap;
 
 @Path("graphs/{graph}/jobs/gremlin")
 @Singleton
diff --git a/hugegraph-api/src/main/java/com/baidu/hugegraph/version/ApiVersion.java b/hugegraph-api/src/main/java/com/baidu/hugegraph/version/ApiVersion.java
index 9ef7222e22..9a627212a3 100644
--- a/hugegraph-api/src/main/java/com/baidu/hugegraph/version/ApiVersion.java
+++ b/hugegraph-api/src/main/java/com/baidu/hugegraph/version/ApiVersion.java
@@ -78,10 +78,11 @@ public final class ApiVersion {
      * [0.32] Issue-250: Keep depth and degree consistent for traverser api
      * [0.33] Issue-305: Implement customized paths and crosspoints RESTful API
      * [0.34] Issue-307: Let VertexAPI use simplified property serializer
+     * [0.35] Issue-287: Support pagination when do index query
      */
 
     // The second parameter of Version.of() is for IDE running without JAR
-    public static final Version VERSION = Version.of(ApiVersion.class, "0.34");
+    public static final Version VERSION = Version.of(ApiVersion.class, "0.35");
 
     public static final void check() {
         // Check version of hugegraph-core. Firstly do check from version 0.3
diff --git a/hugegraph-cassandra/pom.xml b/hugegraph-cassandra/pom.xml
index 5f98a01252..fa8352fe2b 100644
--- a/hugegraph-cassandra/pom.xml
+++ b/hugegraph-cassandra/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hugegraph</artifactId>
         <groupId>com.baidu.hugegraph</groupId>
-        <version>0.9.1</version>
+        <version>0.9.2</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
diff --git a/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTable.java b/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTable.java
index ae8c80b974..41b45df32e 100644
--- a/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTable.java
+++ b/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTable.java
@@ -127,25 +127,6 @@ protected List<Select> query2Select(String table, Query query) {
                       "currently, it will be replaced by [0, offset + limit)");
         }
 
-        // Set limit
-        if (query.limit() != Query.NO_LIMIT) {
-            long total = query.total();
-            String page = query.page();
-            if (page == null) {
-                select.limit((int) total);
-            } else {
-                select.setFetchSize((int) total);
-                // It's the first time if page is empty
-                if (!page.isEmpty()) {
-                    try {
-                        select.setPagingState(PagingState.fromString(page));
-                    } catch (PagingStateException e) {
-                        throw new BackendException(e.getMessage());
-                    }
-                }
-            }
-        }
-
         // Set order-by
         for (Map.Entry<HugeKeys, Order> order : query.orders().entrySet()) {
             String name = formatKey(order.getKey());
@@ -162,19 +143,45 @@ protected List<Select> query2Select(String table, Query query) {
 
         if (query.conditions().isEmpty()) {
             // Query only by id
+            this.setPageState(query, ids);
             LOG.debug("Query only by id(s): {}", ids);
             return ids;
         } else {
-            List<Select> conds = new ArrayList<Select>(ids.size());
+            List<Select> conds = new ArrayList<>(ids.size());
             for (Select selection : ids) {
                 // Query by condition
                 conds.addAll(this.queryCondition2Select(query, selection));
             }
+            this.setPageState(query, conds);
             LOG.debug("Query by conditions: {}", conds);
             return conds;
         }
     }
 
+    protected void setPageState(Query query, List<Select> selects) {
+        if (query.limit() == Query.NO_LIMIT) {
+            return;
+        }
+        for (Select select : selects) {
+            long total = query.total();
+            String page = query.page();
+            if (page == null) {
+                // Set limit
+                select.limit((int) total);
+            } else {
+                select.setFetchSize((int) total);
+                // It's the first time if page is empty
+                if (!page.isEmpty()) {
+                    try {
+                        select.setPagingState(PagingState.fromString(page));
+                    } catch (PagingStateException e) {
+                        throw new BackendException(e);
+                    }
+                }
+            }
+        }
+    }
+
     protected List<Select> queryId2Select(Query query, Select select) {
         // Query by id(s)
         if (query.ids().isEmpty()) {
diff --git a/hugegraph-core/pom.xml b/hugegraph-core/pom.xml
index a1d53b1847..d2e3dc69aa 100644
--- a/hugegraph-core/pom.xml
+++ b/hugegraph-core/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <groupId>com.baidu.hugegraph</groupId>
         <artifactId>hugegraph</artifactId>
-        <version>0.9.1</version>
+        <version>0.9.2</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <artifactId>hugegraph-core</artifactId>
@@ -157,7 +157,7 @@
                             </addDefaultSpecificationEntries>
                         </manifest>
                         <manifestEntries>
-                            <Implementation-Version>0.9.1.0</Implementation-Version>
+                            <Implementation-Version>0.9.2.0</Implementation-Version>
                         </manifestEntries>
                     </archive>
                 </configuration>
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/IdHolder.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/IdHolder.java
new file mode 100644
index 0000000000..41bcebabda
--- /dev/null
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/IdHolder.java
@@ -0,0 +1,103 @@
+/*
+ * 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.page;
+
+import java.util.Set;
+import java.util.function.Function;
+
+import com.baidu.hugegraph.backend.id.Id;
+import com.baidu.hugegraph.backend.query.ConditionQuery;
+import com.baidu.hugegraph.util.E;
+import com.baidu.hugegraph.util.InsertionOrderUtil;
+import com.google.common.collect.ImmutableSet;
+
+public final class IdHolder {
+
+    private final ConditionQuery query;
+    private final Function<ConditionQuery, PageIds> idsFetcher;
+    private boolean exhausted;
+
+    private Set<Id> ids;
+
+    /**
+     * For non-paging situation
+     */
+    public IdHolder(Set<Id> ids) {
+        this.query = null;
+        this.idsFetcher = null;
+        this.exhausted = false;
+        if (ids instanceof ImmutableSet) {
+            this.ids = InsertionOrderUtil.newSet(ids);
+        } else {
+            this.ids = ids;
+        }
+    }
+
+    /**
+     * For paging situation
+     */
+    public IdHolder(ConditionQuery query,
+                    Function<ConditionQuery, PageIds> idsFetcher) {
+        E.checkArgument(query.paging(),
+                        "Query '%s' must include page info", query);
+        this.query = query.copy();
+        this.idsFetcher = idsFetcher;
+        this.exhausted = false;
+        this.ids = null;
+    }
+
+    public void merge(Set<Id> ids) {
+        this.ids.addAll(ids);
+    }
+
+    public Set<Id> ids() {
+        E.checkNotNull(this.ids, "ids");
+        return this.ids;
+    }
+
+    public int size() {
+        if (this.ids == null) {
+            return 0;
+        }
+        return this.ids.size();
+    }
+
+    public boolean paging() {
+        return this.idsFetcher != null;
+    }
+
+    public PageIds fetchNext(String page, long pageSize) {
+        if (this.exhausted) {
+            return PageIds.EMPTY;
+        }
+
+        this.query.page(page);
+        this.query.limit(pageSize);
+
+        PageIds result = this.idsFetcher.apply(this.query);
+
+        assert result != null;
+        this.ids = result.ids();
+        if (this.ids.size() != this.query.limit() || result.page() == null) {
+            this.exhausted = true;
+        }
+        return result;
+    }
+}
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/IdHolderList.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/IdHolderList.java
new file mode 100644
index 0000000000..98c8428233
--- /dev/null
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/IdHolderList.java
@@ -0,0 +1,71 @@
+/*
+ * 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.page;
+
+import java.util.ArrayList;
+import java.util.Collection;
+
+import com.baidu.hugegraph.util.E;
+
+public final class IdHolderList extends ArrayList<IdHolder> {
+
+    private final boolean paging;
+
+    public IdHolderList(boolean paging) {
+        this.paging = paging;
+    }
+
+    public boolean paging() {
+        return this.paging;
+    }
+
+    @Override
+    public boolean add(IdHolder holder) {
+        E.checkArgument(this.paging == holder.paging(),
+                        "The IdHolder to be linked must be " +
+                        "in same paging mode");
+        if (this.paging || this.isEmpty()) {
+            super.add(holder);
+        } else {
+            assert this.size() == 1;
+            IdHolder self = this.get(0);
+            assert !self.paging();
+            self.merge(holder.ids());
+        }
+        return true;
+    }
+
+    @Override
+    public boolean addAll(Collection<? extends IdHolder> idHolders) {
+        for (IdHolder idHolder : idHolders) {
+            this.add(idHolder);
+        }
+        return true;
+    }
+
+    public int idsSize() {
+        if (this.paging || this.isEmpty()) {
+            return 0;
+        } else {
+            assert this.size() == 1;
+            return this.get(0).size();
+        }
+    }
+}
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/PageEntryIterator.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/PageEntryIterator.java
new file mode 100644
index 0000000000..f02828299c
--- /dev/null
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/PageEntryIterator.java
@@ -0,0 +1,107 @@
+/*
+ * 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.page;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+import com.baidu.hugegraph.backend.store.BackendEntry;
+import com.baidu.hugegraph.exception.NotSupportException;
+import com.baidu.hugegraph.iterator.Metadatable;
+import com.baidu.hugegraph.util.E;
+
+public class PageEntryIterator implements Iterator<BackendEntry>, Metadatable {
+
+    private final QueryList queries;
+    private final long pageSize;
+    private QueryList.PageIterator results;
+    private PageState pageState;
+    private long remaining;
+
+    public PageEntryIterator(QueryList queries, long pageSize) {
+        this.queries = queries;
+        this.pageSize = pageSize;
+        this.results = QueryList.PageIterator.EMPTY;
+        this.pageState = this.parsePageState();
+        this.remaining = queries.parent().limit();
+    }
+
+    private PageState parsePageState() {
+        String page = this.queries.parent().page();
+        PageState pageState = PageState.fromString(page);
+        E.checkState(pageState.offset() < this.queries.total(),
+                     "Invalid page '%s' with an offset '%s' exceeds " +
+                     "the size of IdHolderList", page, pageState.offset());
+        return pageState;
+    }
+
+    @Override
+    public boolean hasNext() {
+        if (this.results.iterator().hasNext()) {
+            return true;
+        }
+        return this.fetch();
+    }
+
+    private boolean fetch() {
+        if (this.remaining <= 0 ||
+            this.pageState.offset() >= this.queries.total()) {
+            return false;
+        }
+
+        long pageSize = this.remaining < this.pageSize ?
+                        this.remaining : this.pageSize;
+        this.results = this.queries.fetchNext(this.pageState, pageSize);
+        assert this.results != null;
+
+        if (this.results.iterator().hasNext()) {
+            if (results.page() == null) {
+                this.pageState.increase();
+            } else {
+                this.pageState.page(this.results.page());
+            }
+            return true;
+        } else {
+            this.pageState.increase();
+            return this.fetch();
+        }
+    }
+
+    @Override
+    public BackendEntry next() {
+        if (!this.hasNext()) {
+            throw new NoSuchElementException();
+        }
+        BackendEntry entry = this.results.iterator().next();
+        this.remaining--;
+        return entry;
+    }
+
+    @Override
+    public Object metadata(String meta, Object... args) {
+        if ("page".equals(meta)) {
+            if (this.pageState.offset() >= this.queries.total()) {
+                return null;
+            }
+            return this.pageState.toString();
+        }
+        throw new NotSupportException("Invalid meta '%s'", meta);
+    }
+}
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/PageIds.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/PageIds.java
new file mode 100644
index 0000000000..d79da290a7
--- /dev/null
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/PageIds.java
@@ -0,0 +1,50 @@
+/*
+ * 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.page;
+
+import java.util.Set;
+
+import com.baidu.hugegraph.backend.id.Id;
+import com.google.common.collect.ImmutableSet;
+
+public final class PageIds {
+
+    public static final PageIds EMPTY = new PageIds(ImmutableSet.of(), null);
+
+    private final Set<Id> ids;
+    private final String page;
+
+    public PageIds(Set<Id> ids, String page) {
+        this.ids = ids;
+        this.page = page;
+    }
+
+    public Set<Id> ids() {
+        return this.ids;
+    }
+
+    public String page() {
+        return this.page;
+    }
+
+    public boolean empty() {
+        return this.ids.isEmpty();
+    }
+}
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/PageState.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/PageState.java
new file mode 100644
index 0000000000..10264c5f00
--- /dev/null
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/PageState.java
@@ -0,0 +1,97 @@
+/*
+ * 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.page;
+
+import java.util.Base64;
+
+import com.baidu.hugegraph.HugeException;
+import com.baidu.hugegraph.backend.serializer.BytesBuffer;
+import com.baidu.hugegraph.util.Bytes;
+import com.baidu.hugegraph.util.E;
+
+public final class PageState {
+
+    public static final String PAGE_NONE = "";
+
+    private int offset;
+    private String page;
+
+    public PageState(int offset, String page) {
+        E.checkArgument(offset >= 0, "The offset must be >= 0");
+        E.checkNotNull(page, "page");
+        this.offset = offset;
+        this.page = page;
+    }
+
+    public void increase() {
+        this.offset++;
+        this.page = PAGE_NONE;
+    }
+
+    public int offset() {
+        return this.offset;
+    }
+
+    public void page(String page) {
+        this.page = page;
+    }
+
+    public String page() {
+        return this.page;
+    }
+
+    @Override
+    public String toString() {
+        return Base64.getEncoder().encodeToString(this.toBytes());
+    }
+
+    public byte[] toBytes() {
+        BytesBuffer buffer = BytesBuffer.allocate(256);
+        buffer.writeInt(this.offset);
+        buffer.writeString(this.page);
+        return buffer.bytes();
+    }
+
+    public static PageState fromString(String page) {
+        byte[] bytes;
+        try {
+            bytes = Base64.getDecoder().decode(page);
+        } catch (Exception e) {
+            throw new HugeException("Invalid page: '%s'", e, page);
+        }
+        return fromBytes(bytes);
+    }
+
+    public static PageState fromBytes(byte[] bytes) {
+        if (bytes.length == 0) {
+            // The first page
+            return new PageState(0, "");
+        }
+        try {
+            BytesBuffer buffer = BytesBuffer.wrap(bytes);
+            int offset = buffer.readInt();
+            String page = buffer.readString();
+            return new PageState(offset, page);
+        } catch (Exception e) {
+            throw new HugeException("Invalid page: '0x%s'",
+                                    e, Bytes.toHex(bytes));
+        }
+    }
+}
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/QueryList.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/QueryList.java
new file mode 100644
index 0000000000..d7f6ba78e6
--- /dev/null
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/page/QueryList.java
@@ -0,0 +1,262 @@
+/*
+ * 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.page;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
+
+import com.baidu.hugegraph.HugeGraph;
+import com.baidu.hugegraph.backend.id.Id;
+import com.baidu.hugegraph.backend.query.IdQuery;
+import com.baidu.hugegraph.backend.query.Query;
+import com.baidu.hugegraph.backend.store.BackendEntry;
+import com.baidu.hugegraph.config.CoreOptions;
+import com.baidu.hugegraph.iterator.FlatMapperIterator;
+import com.baidu.hugegraph.iterator.Metadatable;
+import com.baidu.hugegraph.util.CollectionUtil;
+import com.baidu.hugegraph.util.E;
+import com.google.common.collect.ImmutableSet;
+
+public final class QueryList {
+
+    private final HugeGraph graph;
+    private final Query parent;
+    // The size of each page fetched by the inner page
+    private final Function<Query, Iterator<BackendEntry>> fetcher;
+    private final List<QueryHolder> queries;
+
+    public QueryList(HugeGraph graph, Query parent,
+                     Function<Query, Iterator<BackendEntry>> fetcher) {
+        this.graph = graph;
+        this.parent = parent;
+        this.fetcher = fetcher;
+        this.queries = new ArrayList<>();
+    }
+
+    public Query parent() {
+        return this.parent;
+    }
+
+    public void add(List<IdHolder> holders) {
+        if (!this.parent.paging()) {
+            for (QueryHolder q : this.queries) {
+                if (q instanceof IndexQuery) {
+                    ((IndexQuery) q).holders.addAll(holders);
+                    return;
+                }
+            }
+        }
+        this.queries.add(new IndexQuery(holders));
+    }
+
+    public void add(Query query) {
+        // TODO: maybe need do deduplicate(for -> flatten)
+        this.queries.add(new OptimizedQuery(query));
+    }
+
+    public int total() {
+        int total = 0;
+        for (QueryHolder q : this.queries) {
+            total += q.total();
+        }
+        return total;
+    }
+
+    public boolean empty() {
+        return this.queries.isEmpty();
+    }
+
+    public Iterator<BackendEntry> fetch() {
+        assert !queries.isEmpty();
+        if (this.parent.paging()) {
+            int pageSize = this.graph.configuration()
+                                     .get(CoreOptions.QUERY_PAGE_SIZE);
+            return new PageEntryIterator(this, pageSize);
+        } else {
+            return this.fetchAll();
+        }
+    }
+
+    protected Iterator<BackendEntry> fetchAll() {
+        return new FlatMapperIterator<>(this.queries.iterator(), q -> {
+            return q.iterator();
+        });
+    }
+
+    protected PageIterator fetchNext(PageState pageState, long pageSize) {
+        QueryHolder query = null;
+        int offset = pageState.offset();
+        int current = 0;
+        for (QueryHolder q : this.queries) {
+            if (current + q.total() > offset) {
+                query = q;
+                break;
+            }
+            current += q.total();
+        }
+        E.checkNotNull(query, "query");
+        assert offset >= current;
+        return query.iterator(offset - current, pageState.page(), pageSize);
+    }
+
+    private static Set<Id> limit(Set<Id> ids, Query query) {
+        long fromIndex = query.offset();
+        E.checkArgument(fromIndex <= Integer.MAX_VALUE,
+                        "Offset must be <= 0x7fffffff, but got '%s'",
+                        fromIndex);
+
+        if (query.offset() >= ids.size()) {
+            return ImmutableSet.of();
+        }
+        if (query.limit() == Query.NO_LIMIT && query.offset() == 0) {
+            return ids;
+        }
+        long toIndex = query.offset() + query.limit();
+        if (query.limit() == Query.NO_LIMIT || toIndex > ids.size()) {
+            toIndex = ids.size();
+        }
+        assert fromIndex < ids.size();
+        assert toIndex <= ids.size();
+        return CollectionUtil.subSet(ids, (int) fromIndex, (int) toIndex);
+    }
+
+    /**
+     * A container that can generate queries
+     */
+    private interface QueryHolder {
+
+        /**
+         * For non-paging situation
+         * @return          BackendEntry iterator
+         */
+        public Iterator<BackendEntry> iterator();
+
+        /**
+         * For paging situation
+         * @param index     position IdHolder(Query)
+         * @param page      set query page
+         * @param pageSize  set query page size
+         * @return          BackendEntry iterator with page
+         */
+        public PageIterator iterator(int index, String page, long pageSize);
+
+        public int total();
+    }
+
+    /**
+     * Generate queries from index.optimizeQuery()
+     */
+    private class OptimizedQuery implements QueryHolder {
+
+        private final Query query;
+
+        public OptimizedQuery(Query query) {
+            this.query = query;
+        }
+
+        public Iterator<BackendEntry> iterator() {
+            return fetcher.apply(this.query);
+        }
+
+        public PageIterator iterator(int index, String page, long pageSize) {
+            assert index == 0;
+            this.query.page(page);
+            Iterator<BackendEntry> iterator = fetcher.apply(this.query);
+            // Must iterate all entries before get the next page
+            List<BackendEntry> results = IteratorUtils.list(iterator);
+            page = (String) ((Metadatable) iterator).metadata("page");
+            return new PageIterator(results.iterator(), page);
+        }
+
+        public int total() {
+            return 1;
+        }
+    }
+
+    /**
+     * Generate queries from index.indexQuery()
+     */
+    private class IndexQuery implements QueryHolder {
+
+        // Actual is an instance of IdHolderList
+        private final List<IdHolder> holders;
+
+        public IndexQuery(List<IdHolder> holders) {
+            this.holders = holders;
+        }
+
+        public Iterator<BackendEntry> iterator() {
+            return new FlatMapperIterator<>(this.holders.iterator(), holder -> {
+                if (holder.ids().isEmpty()) {
+                    return null;
+                }
+                Set<Id> ids = holder.ids();
+                if (ids.size() > parent.limit()) {
+                    ids = limit(ids, parent);
+                }
+                IdQuery query = new IdQuery(parent.resultType(), ids);
+                return fetcher.apply(query);
+            });
+        }
+
+        public PageIterator iterator(int index, String page, long pageSize) {
+            IdHolder holder = this.holders.get(index);
+            PageIds pageIds = holder.fetchNext(page, pageSize);
+            if (pageIds.empty()) {
+                return PageIterator.EMPTY;
+            }
+            IdQuery query = new IdQuery(parent.resultType(), pageIds.ids());
+            return new PageIterator(fetcher.apply(query), pageIds.page());
+        }
+
+        public int total() {
+            return this.holders.size();
+        }
+    }
+
+    public static class PageIterator {
+
+        public static final PageIterator EMPTY = new PageIterator(
+                                                 Collections.emptyIterator(),
+                                                 PageState.PAGE_NONE);
+
+        private final Iterator<BackendEntry> iterator;
+        private final String page;
+
+        public PageIterator(Iterator<BackendEntry> iterator, String page) {
+            this.iterator = iterator;
+            this.page = page;
+        }
+
+        public Iterator<BackendEntry> iterator() {
+            return this.iterator;
+        }
+
+        public String page() {
+            return this.page;
+        }
+    }
+}
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/ConditionQuery.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/ConditionQuery.java
index 8d7600eb7a..d94695e3b0 100644
--- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/ConditionQuery.java
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/ConditionQuery.java
@@ -423,7 +423,6 @@ public int optimized() {
         return this.optimizedType;
     }
 
-
     public void registerResultsFilter(Function<HugeElement, Boolean> filter) {
         this.resultsFilter = filter;
 
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/ConditionQueryFlatten.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/ConditionQueryFlatten.java
index 2c228fe238..9d9ec1988a 100644
--- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/ConditionQueryFlatten.java
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/ConditionQueryFlatten.java
@@ -23,7 +23,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Date;
-import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -31,6 +31,7 @@
 import com.baidu.hugegraph.backend.id.Id;
 import com.baidu.hugegraph.backend.query.Condition.Relation;
 import com.baidu.hugegraph.type.define.HugeKeys;
+import com.baidu.hugegraph.util.InsertionOrderUtil;
 import com.baidu.hugegraph.util.NumericUtil;
 import com.google.common.collect.ImmutableList;
 
@@ -44,7 +45,7 @@ public static List<ConditionQuery> flatten(ConditionQuery query) {
         List<ConditionQuery> queries = new ArrayList<>();
 
         // Flatten IN/NOT_IN if needed
-        Set<Condition> conditions = new HashSet<>();
+        Set<Condition> conditions = InsertionOrderUtil.newSet();
         for (Condition condition : query.conditions()) {
             Condition cond = flattenIn(condition);
             if (cond == null) {
@@ -166,7 +167,7 @@ private static Condition convTextContainsAny2Or(Relation relation) {
     }
 
     private static Set<Relations> flattenAndOr(Condition condition) {
-        Set<Relations> result = new HashSet<>();
+        Set<Relations> result = InsertionOrderUtil.newSet();
         switch (condition.type()) {
             case RELATION:
                 Relation relation = (Relation) condition;
@@ -191,7 +192,7 @@ private static Set<Relations> flattenAndOr(Condition condition) {
 
     private static Set<Relations> and(Set<Relations> left,
                                       Set<Relations> right) {
-        Set<Relations> result = new HashSet<>();
+        Set<Relations> result = InsertionOrderUtil.newSet();
         for (Relations leftRelations : left) {
             for (Relations rightRelations : right) {
                 Relations relations = new Relations();
@@ -205,7 +206,7 @@ private static Set<Relations> and(Set<Relations> left,
 
     private static Set<Relations> or(Set<Relations> left,
                                      Set<Relations> right) {
-        Set<Relations> result = new HashSet<>(left);
+        Set<Relations> result = InsertionOrderUtil.newSet(left);
         result.addAll(right);
         return result;
     }
@@ -467,8 +468,9 @@ private static int compare(Relation first, Relation second) {
 
     /**
      * Rename Relation Set to Relations to make code more readable
+     * Use `LinkedHashSet` to keep relation in order with insertion
      */
-    private static class Relations extends HashSet<Relation> {
+    private static class Relations extends LinkedHashSet<Relation> {
 
         private static final long serialVersionUID = -2110811280408887334L;
 
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/IdPrefixQuery.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/IdPrefixQuery.java
index a221f3ae77..a6e7ca3101 100644
--- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/IdPrefixQuery.java
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/IdPrefixQuery.java
@@ -39,6 +39,10 @@ public IdPrefixQuery(Query originQuery, Id prefix) {
         this(originQuery.resultType(), originQuery, prefix, true, prefix);
     }
 
+    public IdPrefixQuery(Query originQuery, Id start, Id prefix) {
+        this(originQuery.resultType(), originQuery, start, true, prefix);
+    }
+
     public IdPrefixQuery(Query originQuery,
                          Id start, boolean inclusive, Id prefix) {
         this(originQuery.resultType(), originQuery, start, inclusive, prefix);
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/Query.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/Query.java
index 1ae2715438..406ccb5a31 100644
--- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/Query.java
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/query/Query.java
@@ -265,8 +265,9 @@ public int hashCode() {
 
     @Override
     public String toString() {
-        return String.format("Query for %s offset=%d, limit=%d, order by %s",
+        return String.format("Query for %s page=%s, offset=%d, limit=%d, order by %s",
                              this.resultType,
+                             this.page,
                              this.offset,
                              this.limit,
                              this.orders.toString());
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/serializer/BinaryInlineSerializer.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/serializer/BinaryInlineSerializer.java
index aebfe1e6fd..1b8a540e39 100644
--- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/serializer/BinaryInlineSerializer.java
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/serializer/BinaryInlineSerializer.java
@@ -33,7 +33,7 @@
 public class BinaryInlineSerializer extends BinarySerializer {
 
     public BinaryInlineSerializer() {
-        super(true);
+        super(true, true);
     }
 
     @Override
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/serializer/BinarySerializer.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/serializer/BinarySerializer.java
index 7d6f342fef..be4a27fd1f 100644
--- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/serializer/BinarySerializer.java
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/serializer/BinarySerializer.java
@@ -34,10 +34,10 @@
 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.IdQuery;
 import com.baidu.hugegraph.backend.query.IdRangeQuery;
 import com.baidu.hugegraph.backend.query.Query;
 import com.baidu.hugegraph.backend.serializer.BinaryBackendEntry.BinaryId;
+import com.baidu.hugegraph.backend.serializer.BinaryEntryIterator.PageState;
 import com.baidu.hugegraph.backend.store.BackendEntry;
 import com.baidu.hugegraph.backend.store.BackendEntry.BackendColumn;
 import com.baidu.hugegraph.schema.EdgeLabel;
@@ -71,20 +71,23 @@
 
 public class BinarySerializer extends AbstractSerializer {
 
-    private static final byte[] EMPTY_BYTES = new byte[0];
+    public static final byte[] EMPTY_BYTES = new byte[0];
 
     /*
      * Id is stored in column name if keyWithIdPrefix=true like RocksDB,
      * else stored in rowkey like HBase.
      */
     private final boolean keyWithIdPrefix;
+    private final boolean indexWithIdPrefix;
 
     public BinarySerializer() {
-        this(true);
+        this(true, true);
     }
 
-    public BinarySerializer(boolean keyWithIdPrefix) {
+    public BinarySerializer(boolean keyWithIdPrefix,
+                            boolean indexWithIdPrefix) {
         this.keyWithIdPrefix = keyWithIdPrefix;
+        this.indexWithIdPrefix = indexWithIdPrefix;
     }
 
     @Override
@@ -323,7 +326,7 @@ protected byte[] formatIndexName(HugeIndex index) {
         int idLen = 1 + elemId.length();
 
         BytesBuffer buffer;
-        if (!this.keyWithIdPrefix) {
+        if (!this.indexWithIdPrefix) {
             buffer = BytesBuffer.allocate(idLen);
         } else {
             Id indexId = index.id();
@@ -350,7 +353,7 @@ protected void parseIndexName(BinaryBackendEntry entry, HugeIndex index,
                 continue;
             }
             BytesBuffer buffer = BytesBuffer.wrap(col.name);
-            if (this.keyWithIdPrefix) {
+            if (this.indexWithIdPrefix) {
                 buffer.readId();
             }
             index.elementIds(buffer.readId(true));
@@ -642,11 +645,26 @@ private Query writeStringIndexQuery(ConditionQuery query) {
         E.checkArgument(index != null, "Please specify the index label");
         E.checkArgument(key != null, "Please specify the index key");
 
-        Id id = formatIndexId(query.resultType(), index, key);
-        IdQuery idQuery = new IdQuery(query, id);
-        idQuery.limit(query.limit());
-        idQuery.offset(query.offset());
-        return idQuery;
+        Id prefix = formatIndexId(query.resultType(), index, key);
+
+        /*
+         * If used paging and the page number is not empty, deserialize
+         * the page to id and use it as the starting row for this query
+         */
+        Query newQuery;
+        if (query.paging() && !query.page().isEmpty()) {
+            byte[] position = PageState.fromString(query.page()).position();
+            E.checkArgument(Bytes.compare(position, prefix.asBytes()) >= 0,
+                            "Invalid page out of lower bound");
+            BinaryId start = new BinaryId(position, null);
+            newQuery = new IdPrefixQuery(query, start, prefix);
+        } else {
+            newQuery = new IdPrefixQuery(query, prefix);
+        }
+        newQuery.page(query.page());
+        newQuery.limit(query.limit());
+        newQuery.offset(query.offset());
+        return newQuery;
     }
 
     private Query writeRangeIndexQuery(ConditionQuery query) {
@@ -689,10 +707,11 @@ private Query writeRangeIndexQuery(ConditionQuery query) {
         HugeType type = query.resultType();
         if (keyEq != null) {
             Id id = formatIndexId(type, index, keyEq);
-            IdQuery idQuery = new IdQuery(query, id);
-            idQuery.limit(query.limit());
-            idQuery.offset(query.offset());
-            return idQuery;
+            Query newQuery = new IdPrefixQuery(query, id);
+            newQuery.page(query.page());
+            newQuery.limit(query.limit());
+            newQuery.offset(query.offset());
+            return newQuery;
         }
 
         if (keyMin == null) {
@@ -713,15 +732,32 @@ private Query writeRangeIndexQuery(ConditionQuery query) {
             keyMinEq = true;
         }
 
+        Id start = min;
+        if (query.paging() && !query.page().isEmpty()) {
+            byte[] position = PageState.fromString(query.page()).position();
+            E.checkArgument(Bytes.compare(position, start.asBytes()) >= 0,
+                            "Invalid page out of lower bound");
+            start = new BinaryId(position, null);
+        }
+
+        Query newQuery;
         if (keyMax == null) {
             Id prefix = formatIndexId(type, index, null);
             // Reset the first byte to make same length-prefix
             prefix.asBytes()[0] = min.asBytes()[0];
-            return new IdPrefixQuery(query, min, keyMinEq, prefix);
+            newQuery = new IdPrefixQuery(query, start, keyMinEq, prefix);
         } else {
             Id max = formatIndexId(type, index, keyMax);
-            return new IdRangeQuery(query, min, keyMinEq, max, keyMaxEq);
+            if (keyMaxEq) {
+                keyMaxEq = false;
+                increaseOne(max.asBytes());
+            }
+            newQuery = new IdRangeQuery(query, start, keyMinEq, max, keyMaxEq);
         }
+        newQuery.page(query.page());
+        newQuery.limit(query.limit());
+        newQuery.offset(query.offset());
+        return newQuery;
     }
 
     private BinaryBackendEntry formatILDeletion(HugeIndex index) {
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendStoreSystemInfo.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendStoreSystemInfo.java
index 0b409a73fe..ce3a493f07 100644
--- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendStoreSystemInfo.java
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendStoreSystemInfo.java
@@ -19,6 +19,8 @@
 
 package com.baidu.hugegraph.backend.store;
 
+import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.Map;
 
 import org.apache.tinkerpop.gremlin.structure.Graph.Hidden;
@@ -66,7 +68,8 @@ private Map<String, Object> info() {
         PropertyKey pkey = null;
         try {
             pkey = schema.getPropertyKey(PK_BACKEND_INFO);
-        } catch (BackendException | IllegalStateException ignored) {
+        } catch (BackendException | IllegalStateException |
+                 UncheckedIOException ignored) {
             // pass
         }
         return pkey != null ? pkey.userdata() : null;
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/AbstractTransaction.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/AbstractTransaction.java
index eb5c6c12e0..d38386c5f2 100644
--- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/AbstractTransaction.java
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/AbstractTransaction.java
@@ -33,6 +33,7 @@
 import com.baidu.hugegraph.backend.store.BackendEntry;
 import com.baidu.hugegraph.backend.store.BackendMutation;
 import com.baidu.hugegraph.backend.store.BackendStore;
+import com.baidu.hugegraph.config.HugeConfig;
 import com.baidu.hugegraph.exception.NotFoundException;
 import com.baidu.hugegraph.perf.PerfUtil.Watched;
 import com.baidu.hugegraph.type.HugeType;
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/GraphIndexTransaction.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/GraphIndexTransaction.java
index 3973536be4..640ffe581d 100644
--- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/GraphIndexTransaction.java
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/GraphIndexTransaction.java
@@ -40,16 +40,19 @@
 import com.baidu.hugegraph.backend.BackendException;
 import com.baidu.hugegraph.backend.id.Id;
 import com.baidu.hugegraph.backend.id.SplicingIdGenerator;
+import com.baidu.hugegraph.backend.page.IdHolder;
+import com.baidu.hugegraph.backend.page.IdHolderList;
+import com.baidu.hugegraph.backend.page.PageIds;
 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.query.ConditionQueryFlatten;
-import com.baidu.hugegraph.backend.query.IdQuery;
-import com.baidu.hugegraph.backend.query.Query;
 import com.baidu.hugegraph.backend.serializer.AbstractSerializer;
 import com.baidu.hugegraph.backend.store.BackendEntry;
 import com.baidu.hugegraph.backend.store.BackendStore;
 import com.baidu.hugegraph.exception.NoIndexException;
+import com.baidu.hugegraph.exception.NotSupportException;
+import com.baidu.hugegraph.iterator.Metadatable;
 import com.baidu.hugegraph.job.EphemeralJob;
 import com.baidu.hugegraph.job.EphemeralJobBuilder;
 import com.baidu.hugegraph.perf.PerfUtil.Watched;
@@ -76,7 +79,6 @@
 public class GraphIndexTransaction extends AbstractTransaction {
 
     private static final String INDEX_EMPTY_SYM = "\u0000";
-    private static final Query EMPTY_QUERY = new ConditionQuery(null);
 
     private final Analyzer textAnalyzer;
 
@@ -226,7 +228,7 @@ private void updateIndex(IndexLabel indexLabel, Object propValue,
      * @return      converted id query
      */
     @Watched(prefix = "index")
-    public Query indexQuery(ConditionQuery query) {
+    public List<IdHolder> indexQuery(ConditionQuery query) {
         // Index query must have been flattened in Graph tx
         query.checkFlattened();
 
@@ -245,29 +247,21 @@ public Query indexQuery(ConditionQuery query) {
 
         // Query by index
         query.optimized(OptimizedType.INDEX.ordinal());
-        Set<Id> ids;
         if (query.allSysprop() && conds.size() == 1 &&
             query.containsCondition(HugeKeys.LABEL)) {
             // Query only by label
-            ids = this.queryByLabel(query);
+            return this.queryByLabel(query);
         } else {
             // Query by userprops (or userprops + label)
-            ids = this.queryByUserprop(query);
+            return this.queryByUserprop(query);
         }
-
-        if (ids.isEmpty()) {
-            return EMPTY_QUERY;
-        }
-
-        // Wrap id(s) by IdQuery
-        return new IdQuery(query, ids);
     }
 
     @Watched(prefix = "index")
-    private Set<Id> queryByLabel(ConditionQuery query) {
+    private List<IdHolder> queryByLabel(ConditionQuery query) {
         HugeType queryType = query.resultType();
         IndexLabel il = IndexLabel.label(queryType);
-        Id label = (Id) query.condition(HugeKeys.LABEL);
+        Id label = query.condition(HugeKeys.LABEL);
         assert label != null;
 
         SchemaLabel schemaLabel;
@@ -290,50 +284,111 @@ private Set<Id> queryByLabel(ConditionQuery query) {
         indexQuery.eq(HugeKeys.INDEX_LABEL_ID, il.id());
         indexQuery.eq(HugeKeys.FIELD_VALUES, label);
         // Set offset and limit to avoid redundant element ids
+        indexQuery.page(query.page());
         indexQuery.limit(query.limit());
         indexQuery.offset(query.offset());
         indexQuery.capacity(query.capacity());
 
-        return this.doIndexQuery(il, indexQuery);
+        IdHolder idHolder = this.doIndexQuery(il, indexQuery);
+        List<IdHolder> holders = new IdHolderList(query.paging());
+        holders.add(idHolder);
+        return holders;
     }
 
     @Watched(prefix = "index")
-    private Set<Id> queryByUserprop(ConditionQuery query) {
+    private List<IdHolder> queryByUserprop(ConditionQuery query) {
         // Get user applied label or collect all qualified labels with
         // related index labels
         Set<MatchedIndex> indexes = this.collectMatchedIndexes(query);
         if (indexes.isEmpty()) {
-            Id label = (Id) query.condition(HugeKeys.LABEL);
+            Id label = query.condition(HugeKeys.LABEL);
             throw noIndexException(this.graph(), query, label);
         }
 
         // Value type of Condition not matched
         if (!validQueryConditionValues(this.graph(), query)) {
-            return ImmutableSet.of();
+            return ImmutableList.of();
         }
 
+        boolean paging = query.paging();
         // Do index query
-        Set<Id> ids = InsertionOrderUtil.newSet();
+        IdHolderList holders = new IdHolderList(paging);
+        long idsSize = 0;
         for (MatchedIndex index : indexes) {
+            if (paging && index.indexLabels().size() > 1) {
+                throw new NotSupportException("joint index query in paging");
+            }
+
             if (index.containsSearchIndex()) {
                 // Do search-index query
-                ids.addAll(this.queryByUserpropWithSearchIndex(query, index));
+                holders.addAll(this.doSearchIndex(query, index));
             } else {
                 // Do secondary-index or range-index query
                 IndexQueries queries = index.constructIndexQueries(query);
-                ids.addAll(this.intersectIndexQueries(queries));
+                assert !paging || queries.size() <= 1;
+                IdHolder holder = this.doIndexQueries(queries);
+                holders.add(holder);
             }
 
-            if (query.reachLimit(ids.size())) {
+            idsSize += holders.idsSize();
+            if (query.reachLimit(idsSize)) {
                 break;
             }
         }
-        return limit(ids, query);
+        return holders;
+    }
+
+    private List<IdHolder> doSearchIndex(ConditionQuery query,
+                                         MatchedIndex index) {
+        query = this.constructSearchQuery(query, index);
+        List<IdHolder> holders = new IdHolderList(query.paging());
+        for (ConditionQuery q : ConditionQueryFlatten.flatten(query)) {
+            IndexQueries queries = index.constructIndexQueries(q);
+            assert !query.paging() || queries.size() <= 1;
+            IdHolder holder = this.doIndexQueries(queries);
+            // NOTE: ids will be merged into one IdHolder if not in paging
+            holders.add(holder);
+        }
+        return holders;
+    }
+
+    @Watched(prefix = "index")
+    private IdHolder doIndexQueries(IndexQueries queries) {
+        if (queries.size() == 1) {
+            return this.doSingleOrCompositeIndex(queries);
+        } else {
+            return this.doJointIndex(queries);
+        }
     }
 
     @Watched(prefix = "index")
-    private Set<Id> queryByUserpropWithSearchIndex(ConditionQuery query,
-                                                   MatchedIndex index) {
+    private IdHolder doSingleOrCompositeIndex(IndexQueries queries) {
+        assert queries.size() == 1;
+        Map.Entry<IndexLabel, ConditionQuery> e = queries.one();
+        IndexLabel indexLabel = e.getKey();
+        ConditionQuery query = e.getValue();
+        return this.doIndexQuery(indexLabel, query);
+    }
+
+    @Watched(prefix = "index")
+    private IdHolder doJointIndex(IndexQueries queries) {
+        Set<Id> intersectIds = null;
+        for (Map.Entry<IndexLabel, ConditionQuery> e : queries.entrySet()) {
+            Set<Id> ids = this.doIndexQuery(e.getKey(), e.getValue()).ids();
+            if (intersectIds == null) {
+                intersectIds = ids;
+            } else {
+                CollectionUtil.intersectWithModify(intersectIds, ids);
+            }
+            if (intersectIds.isEmpty()) {
+                break;
+            }
+        }
+        return new IdHolder(intersectIds);
+    }
+
+    private ConditionQuery constructSearchQuery(ConditionQuery query,
+                                                MatchedIndex index) {
         ConditionQuery originQuery = query;
         Set<Id> indexFields = new HashSet<>();
         // Convert has(key, text) to has(key, textContainsAny(word1, word2))
@@ -372,13 +427,7 @@ private Set<Id> queryByUserpropWithSearchIndex(ConditionQuery query,
             return true;
         });
 
-        // Do query
-        Set<Id> ids = InsertionOrderUtil.newSet();
-        for (ConditionQuery cq : ConditionQueryFlatten.flatten(query)) {
-            IndexQueries queries = index.constructIndexQueries(cq);
-            ids.addAll(this.intersectIndexQueries(queries));
-        }
-        return ids;
+        return query;
     }
 
     private boolean matchSearchIndexWords(String propValue, String fieldValue) {
@@ -396,31 +445,26 @@ private boolean needIndexForLabel() {
     }
 
     @Watched(prefix = "index")
-    private Collection<Id> intersectIndexQueries(IndexQueries queries) {
-        Collection<Id> intersectIds = null;
-
-        for (Map.Entry<IndexLabel, ConditionQuery> entry : queries.entrySet()) {
-            Set<Id> ids = this.doIndexQuery(entry.getKey(), entry.getValue());
-            if (intersectIds == null) {
-                intersectIds = ids;
-            } else {
-                CollectionUtil.intersectWithModify(intersectIds, ids);
-            }
-            if (intersectIds.isEmpty()) {
-                break;
-            }
+    private IdHolder doIndexQuery(IndexLabel indexLabel, ConditionQuery query) {
+        if (!query.paging()) {
+            PageIds pageIds = this.doIndexQueryOnce(indexLabel, query);
+            return new IdHolder(pageIds.ids());
+        } else {
+            return new IdHolder(query, (q) -> {
+                return this.doIndexQueryOnce(indexLabel, q);
+            });
         }
-        return intersectIds;
     }
 
     @Watched(prefix = "index")
-    private Set<Id> doIndexQuery(IndexLabel indexLabel, ConditionQuery query) {
-        Set<Id> ids = InsertionOrderUtil.newSet();
+    private PageIds doIndexQueryOnce(IndexLabel indexLabel,
+                                     ConditionQuery query) {
         LockUtil.Locks locks = new LockUtil.Locks(this.graph().name());
         try {
             locks.lockReads(LockUtil.INDEX_LABEL_DELETE, indexLabel.id());
             locks.lockReads(LockUtil.INDEX_LABEL_REBUILD, indexLabel.id());
 
+            Set<Id> ids = InsertionOrderUtil.newSet();
             Iterator<BackendEntry> entries = super.query(query);
             while(entries.hasNext()) {
                 HugeIndex index = this.serializer.readIndex(graph(), query,
@@ -430,16 +474,29 @@ private Set<Id> doIndexQuery(IndexLabel indexLabel, ConditionQuery query) {
                     break;
                 }
             }
+            // If there is no data, the entries is not a Metadatable object
+            if (ids.isEmpty()) {
+                return PageIds.EMPTY;
+            }
+            // NOTE: Memory backend's iterator is not Metadatable
+            if (!query.paging()) {
+                return new PageIds(ids, null);
+            }
+            E.checkState(entries instanceof Metadatable,
+                         "The entries must be Metadatable when query " +
+                         "in paging, but got '%s'",
+                         entries.getClass().getName());
+            Object page = ((Metadatable) entries).metadata("page");
+            return new PageIds(ids, (String) page);
         } finally {
             locks.unlock();
         }
-        return ids;
     }
 
     @Watched(prefix = "index")
     private Set<MatchedIndex> collectMatchedIndexes(ConditionQuery query) {
         SchemaTransaction schema = this.graph().schemaTransaction();
-        Id label = (Id) query.condition(HugeKeys.LABEL);
+        Id label = query.condition(HugeKeys.LABEL);
 
         List<? extends SchemaLabel> schemaLabels;
         if (label != null) {
@@ -908,27 +965,6 @@ private static Set<IndexLabel> relatedIndexLabels(HugeElement element) {
         return indexLabels;
     }
 
-    private static Set<Id> limit(Set<Id> ids, Query query) {
-        long fromIndex = query.offset();
-        E.checkArgument(fromIndex <= Integer.MAX_VALUE,
-                        "Offset must be <= 0x7fffffff, but got '%s'",
-                        fromIndex);
-
-        if (query.offset() >= ids.size()) {
-            return ImmutableSet.of();
-        }
-        if (query.limit() == Query.NO_LIMIT && query.offset() == 0) {
-            return ids;
-        }
-        long toIndex = query.offset() + query.limit();
-        if (query.limit() == Query.NO_LIMIT || toIndex > ids.size()) {
-            toIndex = ids.size();
-        }
-        assert fromIndex < ids.size();
-        assert toIndex <= ids.size();
-        return CollectionUtil.subSet(ids, (int) fromIndex, (int) toIndex);
-    }
-
     public void removeIndex(IndexLabel indexLabel) {
         HugeIndex index = new HugeIndex(indexLabel);
         this.doRemove(this.serializer.writeIndex(index));
@@ -968,6 +1004,7 @@ public IndexQueries constructIndexQueries(ConditionQuery query) {
                  * Not set offset because this query might be a sub-query,
                  * see queryByUserprop()
                  */
+                indexQuery.page(query.page());
                 indexQuery.limit(query.total());
 
                 return IndexQueries.of(il, indexQuery);
@@ -1001,6 +1038,12 @@ public static IndexQueries of(IndexLabel il, ConditionQuery query) {
             indexQueries.put(il, query);
             return indexQueries;
         }
+
+        public Map.Entry<IndexLabel, ConditionQuery> one() {
+            E.checkState(this.size() == 1,
+                         "Please ensure index queries only contains one entry");
+            return this.entrySet().iterator().next();
+        }
     }
 
     public enum OptimizedType {
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/GraphTransaction.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/GraphTransaction.java
index fbe14beb9f..c3b178379b 100644
--- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/GraphTransaction.java
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/GraphTransaction.java
@@ -45,6 +45,8 @@
 import com.baidu.hugegraph.backend.id.EdgeId;
 import com.baidu.hugegraph.backend.id.Id;
 import com.baidu.hugegraph.backend.id.SplicingIdGenerator;
+import com.baidu.hugegraph.backend.page.IdHolder;
+import com.baidu.hugegraph.backend.page.QueryList;
 import com.baidu.hugegraph.backend.query.Condition;
 import com.baidu.hugegraph.backend.query.ConditionQuery;
 import com.baidu.hugegraph.backend.query.ConditionQueryFlatten;
@@ -311,40 +313,24 @@ public Iterator<BackendEntry> query(Query query) {
             return super.query(query);
         }
 
-        ConditionQuery cq = (ConditionQuery) query;
-        if (cq.isFlattened()) {
+        QueryList queries = new QueryList(this.graph(), query,
+                                          q -> super.query(q));
+        for (ConditionQuery cq: ConditionQueryFlatten.flatten(
+                                (ConditionQuery) query)) {
+            Query q = this.optimizeQuery(cq);
             /*
              * NOTE: There are two possibilities for this query:
              * 1.sysprop-query, which would not be empty.
              * 2.index-query result(ids after optimization), which may be empty.
              */
-            Query q = this.optimizeQuery(cq);
-            // Return empty if there is no result after index-query
-            if (q.empty()) {
-                return Collections.emptyIterator();
-            }
-            return super.query(q);
-        }
-
-        // Flatten and optimize the query
-        List<Query> queries = new ArrayList<>();
-        IdQuery idQuery = new IdQuery(query.resultType(), query);
-        for (ConditionQuery fcq: ConditionQueryFlatten.flatten(cq)) {
-            Query q = this.optimizeQuery(fcq);
-            if (q.getClass() == IdQuery.class && !q.ids().isEmpty()) {
-                idQuery.query(q.ids());
+            if (q == null) {
+                queries.add(this.indexQuery(cq));
             } else if (!q.empty()) {
                 queries.add(q);
             }
         }
-        if (!idQuery.empty()) {
-            queries.add(idQuery);
-        }
-        ExtendableIterator<BackendEntry> rs = new ExtendableIterator<>();
-        for (Query q : queries) {
-            rs.extend(super.query(q));
-        }
-        return rs;
+
+        return !queries.empty() ? queries.fetch() : Collections.emptyIterator();
     }
 
     @Watched(prefix = "graph")
@@ -947,7 +933,7 @@ public static void verifyEdgesConditionQuery(ConditionQuery query) {
         }
     }
 
-    protected Query optimizeQuery(ConditionQuery query) {
+    private Query optimizeQuery(ConditionQuery query) {
         Id label = (Id) query.condition(HugeKeys.LABEL);
 
         // Optimize vertex query
@@ -1032,6 +1018,10 @@ protected Query optimizeQuery(ConditionQuery query) {
             }
         }
 
+        return null;
+    }
+
+    private List<IdHolder> indexQuery(ConditionQuery query) {
         /*
          * Optimize by index-query
          * It will return a list of id (maybe empty) if success,
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/config/CoreOptions.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/config/CoreOptions.java
index 7965f92c5f..b2cece77e8 100644
--- a/hugegraph-core/src/main/java/com/baidu/hugegraph/config/CoreOptions.java
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/config/CoreOptions.java
@@ -207,6 +207,14 @@ public static synchronized CoreOptions instance() {
                     (60 * 10)
             );
 
+    public static final ConfigOption<Integer> QUERY_PAGE_SIZE =
+            new ConfigOption<>(
+                    "query.page_size",
+                    "The size of each page when query using paging.",
+                    rangeInt(0, 10000),
+                    500
+            );
+
     public static final ConfigOption<Long> SNOWFLAKE_WORKER_ID =
             new ConfigOption<>(
                     "snowflake.worker_id",
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/version/CoreVersion.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/version/CoreVersion.java
index 1135e52449..6548edd6f0 100644
--- a/hugegraph-core/src/main/java/com/baidu/hugegraph/version/CoreVersion.java
+++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/version/CoreVersion.java
@@ -33,7 +33,7 @@ public class CoreVersion {
 
     // The second parameter of Version.of() is for IDE running without JAR
     public static final Version VERSION = Version.of(CoreVersion.class,
-                                                     "0.9.1");
+                                                     "0.9.2");
 
     public static final String GREMLIN_VERSION = "3.2.5";
 
diff --git a/hugegraph-dist/pom.xml b/hugegraph-dist/pom.xml
index f555eda2ad..7f697cdba8 100644
--- a/hugegraph-dist/pom.xml
+++ b/hugegraph-dist/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hugegraph</artifactId>
         <groupId>com.baidu.hugegraph</groupId>
-        <version>0.9.1</version>
+        <version>0.9.2</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <name>hugegraph-dist</name>
diff --git a/hugegraph-example/pom.xml b/hugegraph-example/pom.xml
index dece4997a1..a82ebdcab2 100644
--- a/hugegraph-example/pom.xml
+++ b/hugegraph-example/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hugegraph</artifactId>
         <groupId>com.baidu.hugegraph</groupId>
-        <version>0.9.1</version>
+        <version>0.9.2</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
diff --git a/hugegraph-hbase/pom.xml b/hugegraph-hbase/pom.xml
index f4dfc2a1bc..b75b3942e5 100644
--- a/hugegraph-hbase/pom.xml
+++ b/hugegraph-hbase/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hugegraph</artifactId>
         <groupId>com.baidu.hugegraph</groupId>
-        <version>0.9.1</version>
+        <version>0.9.2</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
diff --git a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseSerializer.java b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseSerializer.java
index 7e0d902922..71029d00ff 100644
--- a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseSerializer.java
+++ b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseSerializer.java
@@ -24,6 +24,6 @@
 public class HbaseSerializer extends BinarySerializer {
 
     public HbaseSerializer() {
-        super(false);
+        super(false, true);
     }
 }
diff --git a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseSessions.java b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseSessions.java
index 8ee05f3a7e..1d03b79851 100644
--- a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseSessions.java
+++ b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseSessions.java
@@ -64,6 +64,7 @@
 import org.apache.hadoop.hbase.util.VersionInfo;
 
 import com.baidu.hugegraph.backend.BackendException;
+import com.baidu.hugegraph.backend.serializer.BinarySerializer;
 import com.baidu.hugegraph.backend.store.BackendEntry.BackendColumn;
 import com.baidu.hugegraph.backend.store.BackendEntry.BackendIterator;
 import com.baidu.hugegraph.backend.store.BackendSession;
@@ -321,6 +322,16 @@ public void put(String table, byte[] family, byte[] rowkey,
             this.batch(table, put);
         }
 
+        /**
+         * Add a row record to a table(can be used when adding an index)
+         */
+        public void put(String table, byte[] family,
+                        byte[] rowkey, byte[] qualifier, byte[] value) {
+            Put put = new Put(rowkey);
+            put.addColumn(family, qualifier, value);
+            this.batch(table, put);
+        }
+
         /**
          * Delete a record by rowkey and qualifier from a table
          */
diff --git a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseStoreProvider.java b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseStoreProvider.java
index 09763ee6f9..d7f46c0c2c 100644
--- a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseStoreProvider.java
+++ b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseStoreProvider.java
@@ -52,7 +52,8 @@ public String version() {
          * [1.0] HugeGraph-1328: supports backend table version checking
          * [1.1] HugeGraph-1322: add support for full-text search
          * [1.2] #296: support range sortKey feature
+         * [1.3] #287: support pagination when doing index query
          */
-        return "1.2";
+        return "1.3";
     }
 }
diff --git a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseTable.java b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseTable.java
index 43b1e5e1f6..d8201b82ef 100644
--- a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseTable.java
+++ b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseTable.java
@@ -23,7 +23,6 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellScanner;
@@ -52,6 +51,7 @@
 import com.baidu.hugegraph.type.HugeType;
 import com.baidu.hugegraph.util.Bytes;
 import com.baidu.hugegraph.util.E;
+import com.baidu.hugegraph.util.InsertionOrderUtil;
 import com.baidu.hugegraph.util.Log;
 import com.google.common.collect.ImmutableList;
 
@@ -178,8 +178,10 @@ protected RowIterator queryById(Session session, Id id) {
     }
 
     protected RowIterator queryByIds(Session session, Set<Id> ids) {
-        Set<byte[]> rowkeys = ids.stream().map(Id::asBytes)
-                                 .collect(Collectors.toSet());
+        Set<byte[]> rowkeys = InsertionOrderUtil.newSet();
+        for (Id id : ids) {
+            rowkeys.add(id.asBytes());
+        }
         return session.get(this.table(), null, rowkeys);
     }
 
@@ -212,8 +214,8 @@ protected RowIterator queryByRange(Session session, Shard shard) {
         return session.scan(this.table(), start, end);
     }
 
-    private BackendEntryIterator newEntryIterator(RowIterator rows,
-                                                  Query query) {
+    protected BackendEntryIterator newEntryIterator(RowIterator rows,
+                                                    Query query) {
         return new BinaryEntryIterator<>(rows, query, (entry, row) -> {
             E.checkState(!row.isEmpty(), "Can't parse empty HBase result");
             byte[] id = row.getRow();
diff --git a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseTables.java b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseTables.java
index c07550dff1..4bfe758276 100644
--- a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseTables.java
+++ b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseTables.java
@@ -34,8 +34,12 @@
 import com.baidu.hugegraph.backend.query.Condition.Relation;
 import com.baidu.hugegraph.backend.query.ConditionQuery;
 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.serializer.BinarySerializer;
 import com.baidu.hugegraph.backend.store.BackendEntry;
 import com.baidu.hugegraph.backend.store.BackendEntry.BackendColumn;
+import com.baidu.hugegraph.backend.store.BackendEntryIterator;
 import com.baidu.hugegraph.backend.store.hbase.HbaseSessions.RowIterator;
 import com.baidu.hugegraph.backend.store.hbase.HbaseSessions.Session;
 import com.baidu.hugegraph.type.HugeType;
@@ -176,10 +180,19 @@ public IndexTable(String table) {
             super(table);
         }
 
+        @Override
+        public void insert(Session session, BackendEntry entry) {
+            assert entry.columns().size() == 1;
+            BackendColumn col = entry.columns().iterator().next();
+            session.put(this.table(), CF, col.name,
+                        BinarySerializer.EMPTY_BYTES, col.value);
+        }
+
         @Override
         public void eliminate(Session session, BackendEntry entry) {
             assert entry.columns().size() == 1;
-            super.delete(session, entry);
+            BackendColumn col = entry.columns().iterator().next();
+            session.delete(this.table(), CF, col.name);
         }
 
         @Override
@@ -208,19 +221,15 @@ public void delete(Session session, BackendEntry entry) {
         }
 
         @Override
-        protected void parseRowColumns(Result row, BackendEntry entry,
-                                       Query query) throws IOException {
-            if (query.limit() == Query.NO_LIMIT) {
-                super.parseRowColumns(row, entry, query);
-                return;
-            }
-            long total = query.total();
-            CellScanner cellScanner = row.cellScanner();
-            while (cellScanner.advance() && total-- > 0) {
-                Cell cell = cellScanner.current();
-                entry.columns(BackendColumn.of(CellUtil.cloneQualifier(cell),
-                                               CellUtil.cloneValue(cell)));
-            }
+        protected BackendEntryIterator newEntryIterator(RowIterator rows,
+                                                        Query query) {
+            return new BinaryEntryIterator<>(rows, query, (entry, row) -> {
+                assert row.size() == 1;
+                BackendColumn col = BackendColumn.of(row.getRow(), row.value());
+                entry = new BinaryBackendEntry(query.resultType(), col.name);
+                entry.columns(col);
+                return entry;
+            });
         }
     }
 
@@ -242,7 +251,7 @@ public SearchIndex(String store) {
         }
     }
 
-    public static class RangeIndex extends HbaseTable {
+    public static class RangeIndex extends IndexTable {
 
         public static final String TABLE = "ri";
 
diff --git a/hugegraph-mysql/pom.xml b/hugegraph-mysql/pom.xml
index 86e66f4795..b4c9cc5d7f 100644
--- a/hugegraph-mysql/pom.xml
+++ b/hugegraph-mysql/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hugegraph</artifactId>
         <groupId>com.baidu.hugegraph</groupId>
-        <version>0.9.1</version>
+        <version>0.9.2</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
diff --git a/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlTable.java b/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlTable.java
index 1edfb47fad..716e71d3a2 100644
--- a/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlTable.java
+++ b/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlTable.java
@@ -497,6 +497,9 @@ protected void wrapPage(StringBuilder select, Query query) {
             boolean startWithWhere = query.conditions().isEmpty();
             WhereBuilder where = new WhereBuilder(startWithWhere);
             where.gte(formatKeys(idColumnNames), values);
+            if (!startWithWhere) {
+                select.append(" AND");
+            }
             select.append(where.build());
         }
 
diff --git a/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlTables.java b/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlTables.java
index d26be4266b..c929bd4c4a 100644
--- a/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlTables.java
+++ b/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlTables.java
@@ -358,27 +358,6 @@ public Index(String table) {
             super(table);
         }
 
-        @Override
-        protected BackendEntry mergeEntries(BackendEntry e1, BackendEntry e2) {
-            MysqlBackendEntry current = (MysqlBackendEntry) e1;
-            MysqlBackendEntry next = (MysqlBackendEntry) e2;
-
-            E.checkState(current == null || current.type().isIndex(),
-                         "The current entry must be null or INDEX");
-            E.checkState(next != null && next.type().isIndex(),
-                         "The next entry must be INDEX");
-
-            if (current != null) {
-                String currentId = this.entryId(current);
-                String nextId = this.entryId(next);
-                if (currentId.equals(nextId)) {
-                    current.subRow(next.row());
-                    return current;
-                }
-            }
-            return next;
-        }
-
         protected abstract String entryId(MysqlBackendEntry entry);
     }
 
diff --git a/hugegraph-palo/pom.xml b/hugegraph-palo/pom.xml
index 7b6adde386..019fee14a5 100644
--- a/hugegraph-palo/pom.xml
+++ b/hugegraph-palo/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hugegraph</artifactId>
         <groupId>com.baidu.hugegraph</groupId>
-        <version>0.9.1</version>
+        <version>0.9.2</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
diff --git a/hugegraph-rocksdb/pom.xml b/hugegraph-rocksdb/pom.xml
index e56b9e7e62..9427f9076f 100644
--- a/hugegraph-rocksdb/pom.xml
+++ b/hugegraph-rocksdb/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hugegraph</artifactId>
         <groupId>com.baidu.hugegraph</groupId>
-        <version>0.9.1</version>
+        <version>0.9.2</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
diff --git a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBTable.java b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBTable.java
index d580e06a8d..10c8394fe8 100644
--- a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBTable.java
+++ b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBTable.java
@@ -207,8 +207,7 @@ protected BackendColumnIterator queryByRange(Session session, Shard shard) {
     protected static BackendEntryIterator newEntryIterator(
                                           BackendColumnIterator cols,
                                           Query query) {
-        return new BinaryEntryIterator<BackendColumn>(cols, query,
-                                                      (entry, col) -> {
+        return new BinaryEntryIterator<>(cols, query, (entry, col) -> {
             if (entry == null || !entry.belongToMe(col)) {
                 HugeType type = query.resultType();
                 // NOTE: only support BinaryBackendEntry currently
diff --git a/hugegraph-scylladb/pom.xml b/hugegraph-scylladb/pom.xml
index 229ad0a0c6..8d71f0bd52 100644
--- a/hugegraph-scylladb/pom.xml
+++ b/hugegraph-scylladb/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hugegraph</artifactId>
         <groupId>com.baidu.hugegraph</groupId>
-        <version>0.9.1</version>
+        <version>0.9.2</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
diff --git a/hugegraph-test/pom.xml b/hugegraph-test/pom.xml
index 1963d32d52..8d7addb7f9 100644
--- a/hugegraph-test/pom.xml
+++ b/hugegraph-test/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hugegraph</artifactId>
         <groupId>com.baidu.hugegraph</groupId>
-        <version>0.9.1</version>
+        <version>0.9.2</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
diff --git a/hugegraph-test/src/main/java/com/baidu/hugegraph/core/VertexCoreTest.java b/hugegraph-test/src/main/java/com/baidu/hugegraph/core/VertexCoreTest.java
index 0f8130fa9a..f25a6a7a14 100644
--- a/hugegraph-test/src/main/java/com/baidu/hugegraph/core/VertexCoreTest.java
+++ b/hugegraph-test/src/main/java/com/baidu/hugegraph/core/VertexCoreTest.java
@@ -1188,8 +1188,8 @@ public void testQueryByIntPropUsingLteWithMultiResults() {
         init5Persons();
 
         List<Vertex> vertexes = graph.traversal().V()
-                                .hasLabel("person").has("age", P.lte(20))
-                                .toList();
+                                     .hasLabel("person").has("age", P.lte(20))
+                                     .toList();
 
         Assert.assertEquals(4, vertexes.size());
     }
@@ -2998,7 +2998,7 @@ public void testScanVertex() {
 
         long splitSize = 1 * 1024 * 1024;
         Object splits = graph.graphTransaction()
-                        .metadata(HugeType.VERTEX, "splits", splitSize);
+                             .metadata(HugeType.VERTEX, "splits", splitSize);
         for (Shard split : (List<Shard>) splits) {
             ConditionQuery q = new ConditionQuery(HugeType.VERTEX);
             q.scan(split.start(), split.end());
@@ -3332,6 +3332,491 @@ public void testQueryByPageWithOffset() {
         });
     }
 
+    @Test
+    public void testQueryByLabelInPageWithLimitLtePageSize() {
+        Assume.assumeTrue("Not support paging",
+                          storeFeatures().supportsQueryByPage());
+
+        HugeGraph graph = graph();
+        GraphTraversalSource g = graph.traversal();
+        initPageTestData();
+
+        Assert.assertThrows(IllegalArgumentException.class, () -> {
+            g.V().hasLabel("unknown").has("~page", "").limit(1).toList();
+        });
+
+        Assert.assertThrows(IllegalStateException.class, () -> {
+            g.V().hasLabel("programmer").has("~page", "").limit(0).toList();
+        });
+
+        List<Vertex> vertices = g.V().hasLabel("programmer")
+                                 .has("~page", "").limit(1)
+                                 .toList();
+        Assert.assertEquals(1, vertices.size());
+        vertices.forEach(v -> {
+            Assert.assertEquals("programmer", v.label());
+        });
+
+        vertices = g.V().hasLabel("software")
+                    .has("~page", "").limit(5)
+                    .toList();
+        Assert.assertEquals(5, vertices.size());
+        vertices.forEach(v -> {
+            Assert.assertEquals("software", v.label());
+        });
+
+        vertices = g.V().hasLabel("programmer")
+                    .has("~page", "").limit(10)
+                    .toList();
+        Assert.assertEquals(10, vertices.size());
+        vertices.forEach(v -> {
+            Assert.assertEquals("programmer", v.label());
+        });
+    }
+
+    @Test
+    public void testQueryByLabelInPageWithLimitGtPageSize() {
+        Assume.assumeTrue("Not support paging",
+                          storeFeatures().supportsQueryByPage());
+
+        HugeGraph graph = graph();
+        GraphTraversalSource g = graph.traversal();
+        initPageTestData();
+
+        // Limit > page-size(10), test internal paging
+        List<Vertex> vertices = g.V().hasLabel("programmer")
+                                 .has("~page", "").limit(11)
+                                 .toList();
+        Assert.assertEquals(11, vertices.size());
+        vertices.forEach(v -> {
+            Assert.assertEquals("programmer", v.label());
+        });
+
+        vertices = g.V().hasLabel("software")
+                    .has("~page", "").limit(15)
+                    .toList();
+        Assert.assertEquals(15, vertices.size());
+        vertices.forEach(v -> {
+            Assert.assertEquals("software", v.label());
+        });
+
+        vertices = g.V().hasLabel("programmer")
+                    .has("~page", "").limit(20)
+                    .toList();
+        // Programmer only has 18
+        Assert.assertEquals(18, vertices.size());
+        vertices.forEach(v -> {
+            Assert.assertEquals("programmer", v.label());
+        });
+    }
+
+    @Test
+    public void testQueryBySingleLabelInPage() {
+        Assume.assumeTrue("Not support paging",
+                          storeFeatures().supportsQueryByPage());
+
+        HugeGraph graph = graph();
+        GraphTraversalSource g = graph.traversal();
+        initPageTestData();
+
+        GraphTraversal<Vertex, Vertex> itor = g.V().hasLabel("programmer")
+                                               .has("~page", "").limit(1);
+        Assert.assertEquals(1, IteratorUtils.count(itor));
+
+        String page = TraversalUtil.page(itor);
+        List<Vertex> vertices;
+
+        vertices = g.V().hasLabel("programmer")
+                    .has("~page", page).limit(1).toList();
+        Assert.assertEquals(1, vertices.size());
+        Vertex vertex1 = vertices.get(0);
+
+        vertices = g.V().hasLabel("programmer")
+                    .has("~page", page).limit(9).toList();
+        Assert.assertEquals(9, vertices.size());
+        Vertex vertex2 = vertices.get(0);
+        Assert.assertEquals(vertex1.id(), vertex2.id());
+        Assert.assertEquals(vertex1.label(), vertex2.label());
+        Assert.assertEquals(IteratorUtils.asList(vertex1.properties()),
+                            IteratorUtils.asList(vertex2.properties()));
+
+        vertices = g.V().hasLabel("programmer")
+                    .has("~page", page).limit(17).toList();
+        Assert.assertEquals(17, vertices.size());
+        Vertex vertex3 = vertices.get(16);
+
+        vertices = g.V().hasLabel("programmer")
+                    .has("~page", page).limit(18).toList();
+        Assert.assertEquals(17, vertices.size());
+        Vertex vertex4 = vertices.get(16);
+        Assert.assertEquals(vertex3.id(), vertex4.id());
+        Assert.assertEquals(vertex3.label(), vertex4.label());
+        Assert.assertEquals(IteratorUtils.asList(vertex3.properties()),
+                            IteratorUtils.asList(vertex4.properties()));
+    }
+
+    @Test
+    public void testQueryByMultiLabelInPage() {
+        Assume.assumeTrue("Not support paging",
+                          storeFeatures().supportsQueryByPage());
+
+        HugeGraph graph = graph();
+        GraphTraversalSource g = graph.traversal();
+        initPageTestData();
+
+        Assert.assertThrows(IllegalArgumentException.class, () -> {
+            g.V().hasLabel("unknown", "programmer")
+             .has("~page", "").limit(1)
+             .toList();
+        });
+
+        GraphTraversal<Vertex, Vertex> itor;
+
+        itor = g.V().hasLabel("programmer", "software").has("~page", "")
+                .limit(1);
+        Assert.assertEquals(1, IteratorUtils.count(itor));
+
+        String page = TraversalUtil.page(itor);
+        List<Vertex> vertices;
+
+        vertices = g.V().hasLabel("programmer", "software")
+                    .has("~page", page).limit(1).toList();
+        Assert.assertEquals(1, vertices.size());
+        Vertex vertex1 = vertices.get(0);
+
+        vertices = g.V().hasLabel("programmer", "software")
+                    .has("~page", page).limit(9).toList();
+        Assert.assertEquals(9, vertices.size());
+        Vertex vertex2 = vertices.get(0);
+        Assert.assertEquals(vertex1.id(), vertex2.id());
+        Assert.assertEquals(vertex1.label(), vertex2.label());
+        Assert.assertEquals(IteratorUtils.asList(vertex1.properties()),
+                            IteratorUtils.asList(vertex2.properties()));
+
+        vertices = g.V().hasLabel("programmer", "software")
+                    .has("~page", page).limit(18).toList();
+        Assert.assertEquals(18, vertices.size());
+        Vertex vertex3 = vertices.get(17);
+
+        vertices = g.V().hasLabel("programmer", "software")
+                    .has("~page", page).limit(33).toList();
+        Assert.assertEquals(33, vertices.size());
+        Vertex vertex4 = vertices.get(17);
+        Assert.assertEquals(vertex3.id(), vertex4.id());
+        Assert.assertEquals(vertex3.label(), vertex4.label());
+        Assert.assertEquals(IteratorUtils.asList(vertex3.properties()),
+                            IteratorUtils.asList(vertex4.properties()));
+    }
+
+    @Test
+    public void testQueryByPropertyInPageWithLimitLtePageSize() {
+        Assume.assumeTrue("Not support paging",
+                          storeFeatures().supportsQueryByPage());
+
+        HugeGraph graph = graph();
+        GraphTraversalSource g = graph.traversal();
+        initPageTestData();
+
+        Assert.assertThrows(IllegalStateException.class, () -> {
+            g.V().has("name", "marko").has("~page", "").limit(0).toList();
+        });
+
+        // Secondary
+        List<Vertex> vertices = g.V().has("name", "marko")
+                                 .has("~page", "").limit(1)
+                                 .toList();
+        Assert.assertEquals(1, vertices.size());
+        vertices.forEach(v -> {
+            Assert.assertEquals("marko", v.value("name"));
+        });
+
+        // Range
+        vertices = g.V().has("price", P.between(200, 400))
+                    .has("~page", "").limit(5)
+                    .toList();
+        Assert.assertEquals(5, vertices.size());
+        vertices.forEach(v -> {
+            Assert.assertTrue((int) v.value("price") >= 200);
+            Assert.assertTrue((int) v.value("price") < 400);
+        });
+
+        // Search
+        vertices = g.V().has("city", Text.contains("Beijing"))
+                    .has("~page", "").limit(10)
+                    .toList();
+        Assert.assertEquals(10, vertices.size());
+        vertices.forEach(v -> {
+            Assert.assertTrue(((String) v.value("city")).contains("Beijing"));
+        });
+    }
+
+    @Test
+    public void testQueryByPropertyInPageWithLimitGtPageSize() {
+        Assume.assumeTrue("Not support paging",
+                          storeFeatures().supportsQueryByPage());
+
+        HugeGraph graph = graph();
+        GraphTraversalSource g = graph.traversal();
+        initPageTestData();
+
+        // Limit > page-size(10), test internal paging
+        // Secondary
+        List<Vertex> vertices = g.V().has("name", "marko")
+                                 .has("~page", "").limit(11)
+                                 .toList();
+        Assert.assertEquals(11, vertices.size());
+        vertices.forEach(v -> {
+            Assert.assertEquals("marko", v.value("name"));
+        });
+
+        // Range
+        vertices = g.V().has("price", P.between(100, 400))
+                    .has("~page", "").limit(15)
+                    .toList();
+        Assert.assertEquals(12, vertices.size());
+        vertices.forEach(v -> {
+            Assert.assertTrue((int) v.value("price") >= 100);
+            Assert.assertTrue((int) v.value("price") < 400);
+        });
+
+        // Search
+        vertices = g.V().has("city", Text.contains("Beijing"))
+                    .has("~page", "").limit(20)
+                    .toList();
+        Assert.assertEquals(12, vertices.size());
+        vertices.forEach(v -> {
+            Assert.assertTrue(((String) v.value("city")).contains("Beijing"));
+        });
+    }
+
+    @Test
+    public void testQueryBySingleSecondaryPropertyInPage() {
+        Assume.assumeTrue("Not support paging",
+                          storeFeatures().supportsQueryByPage());
+
+        HugeGraph graph = graph();
+        GraphTraversalSource g = graph.traversal();
+        initPageTestData();
+
+        GraphTraversal<Vertex, Vertex> itor = g.V().has("name", "marko")
+                                               .has("~page", "").limit(1);
+        Assert.assertEquals(1, IteratorUtils.count(itor));
+
+        String page = TraversalUtil.page(itor);
+        List<Vertex> vertices;
+
+        vertices = g.V().has("name", "marko")
+                    .has("~page", page).limit(1).toList();
+        Assert.assertEquals(1, vertices.size());
+        Vertex vertex1 = vertices.get(0);
+
+        vertices = g.V().has("name", "marko")
+                    .has("~page", page).limit(9).toList();
+        Assert.assertEquals(9, vertices.size());
+        Assert.assertTrue(vertices.contains(vertex1));
+
+        vertices = g.V().has("name", "marko")
+                    .has("~page", page).limit(18).toList();
+        Assert.assertEquals(18, vertices.size());
+        Vertex vertex3 = vertices.get(17);
+
+        vertices = g.V().has("name", "marko")
+                    .has("~page", page).limit(40).toList();
+        Assert.assertEquals(33, vertices.size());
+        Assert.assertTrue(vertices.contains(vertex3));
+    }
+
+    @Test
+    public void testQueryBySingleRangePropertyInPage() {
+        Assume.assumeTrue("Not support paging",
+                          storeFeatures().supportsQueryByPage());
+
+        HugeGraph graph = graph();
+        GraphTraversalSource g = graph.traversal();
+        initPageTestData();
+
+        GraphTraversal<Vertex, Vertex> itor = g.V().has("price", P.gte(100))
+                                               .has("~page", "").limit(1);
+        Assert.assertEquals(1, IteratorUtils.count(itor));
+
+        String page = TraversalUtil.page(itor);
+        List<Vertex> vertices;
+
+        vertices = g.V().has("price", P.gte(100))
+                    .has("~page", page).limit(1).toList();
+        Assert.assertEquals(1, vertices.size());
+        Vertex vertex1 = vertices.get(0);
+
+        vertices = g.V().has("price", P.gte(100))
+                    .has("~page", page).limit(9).toList();
+        Assert.assertEquals(9, vertices.size());
+        Assert.assertTrue(vertices.contains(vertex1));
+
+        vertices = g.V().has("price", P.gte(100))
+                    .has("~page", page).limit(11).toList();
+        Assert.assertEquals(11, vertices.size());
+        Vertex vertex3 = vertices.get(10);
+
+        vertices = g.V().has("price", P.gte(100))
+                    .has("~page", page).limit(20).toList();
+        Assert.assertEquals(15, vertices.size());
+        Assert.assertTrue(vertices.contains(vertex3));
+    }
+
+    @Test
+    public void testQueryBySingleSearchPropertyInPage() {
+        Assume.assumeTrue("Not support paging",
+                          storeFeatures().supportsQueryByPage());
+
+        HugeGraph graph = graph();
+        GraphTraversalSource g = graph.traversal();
+        initPageTestData();
+
+        GraphTraversal<Vertex, Vertex> itor;
+
+        itor = g.V().has("city", Text.contains("Beijing Shanghai"))
+                .has("~page", "").limit(1);
+        Assert.assertEquals(1, IteratorUtils.count(itor));
+
+        String page = TraversalUtil.page(itor);
+        List<Vertex> vertices;
+
+        vertices = g.V().has("city", Text.contains("Beijing Shanghai"))
+                    .has("~page", page).limit(1).toList();
+        Assert.assertEquals(1, vertices.size());
+        Vertex vertex1 = vertices.get(0);
+
+        vertices = g.V().has("city", Text.contains("Beijing Shanghai"))
+                    .has("~page", page).limit(9).toList();
+        Assert.assertEquals(9, vertices.size());
+        Assert.assertTrue(vertices.contains(vertex1));
+
+        vertices = g.V().has("city", Text.contains("Beijing Shanghai"))
+                    .has("~page", page).limit(11).toList();
+        Assert.assertEquals(11, vertices.size());
+        Vertex vertex3 = vertices.get(10);
+
+        vertices = g.V().has("city", Text.contains("Beijing Shanghai"))
+                    .has("~page", page).limit(20).toList();
+        Assert.assertEquals(17, vertices.size());
+        Assert.assertTrue(vertices.contains(vertex3));
+    }
+
+    @Test
+    public void testQueryByCompositePropertyInPage() {
+        Assume.assumeTrue("Not support paging",
+                          storeFeatures().supportsQueryByPage());
+
+        HugeGraph graph = graph();
+        GraphTraversalSource g = graph.traversal();
+        initPageTestData();
+
+        GraphTraversal<Vertex, Vertex> itor;
+
+        itor = g.V().has("name", "marko").has("age", 30)
+                .has("~page", "").limit(1);
+        Assert.assertEquals(1, IteratorUtils.count(itor));
+
+        String page = TraversalUtil.page(itor);
+        List<Vertex> vertices;
+
+        vertices = g.V().has("name", "marko").has("age", 30)
+                    .has("~page", page).limit(1).toList();
+        Assert.assertEquals(1, vertices.size());
+        Vertex vertex1 = vertices.get(0);
+
+        vertices = g.V().has("name", "marko").has("age", 30)
+                    .has("~page", page).limit(9).toList();
+        Assert.assertEquals(9, vertices.size());
+        Assert.assertTrue(vertices.contains(vertex1));
+
+        vertices = g.V().has("name", "marko").has("age", 30)
+                    .has("~page", page).limit(11).toList();
+        Assert.assertEquals(11, vertices.size());
+        Vertex vertex3 = vertices.get(10);
+
+        vertices = g.V().has("name", "marko").has("age", 30)
+                    .has("~page", page).limit(40).toList();
+        Assert.assertEquals(17, vertices.size());
+        Assert.assertTrue(vertices.contains(vertex3));
+    }
+
+    @Test
+    public void testQueryByJointPropertyInPage() {
+        Assume.assumeTrue("Not support paging",
+                          storeFeatures().supportsQueryByPage());
+
+        HugeGraph graph = graph();
+        GraphTraversalSource g = graph.traversal();
+        initPageTestData();
+
+        Assert.assertThrows(HugeException.class, () -> {
+            g.V().has("name", "marko").has("city", Text.contains("Beijing"))
+             .has("~page", "").limit(10).toList();
+        });
+
+        Assert.assertThrows(HugeException.class, () -> {
+            g.V().has("age", 30).has("city", Text.contains("Beijing"))
+             .has("~page", "").limit(10).toList();
+        });
+
+        Assert.assertThrows(HugeException.class, () -> {
+            g.V().has("name", "marko").has("lang", "java")
+             .has("~page", "").limit(10).toList();
+        });
+
+        Assert.assertThrows(HugeException.class, () -> {
+            g.V().has("lang", "java").has("price", 200)
+             .has("~page", "").limit(10).toList();
+        });
+    }
+
+    @Test
+    public void testQueryByUnionIndexInPageWithSomeIndexNoData() {
+        Assume.assumeTrue("Not support paging",
+                          storeFeatures().supportsQueryByPage());
+
+        HugeGraph graph = graph();
+        SchemaManager schema = graph.schema();
+        GraphTraversalSource g = graph.traversal();
+        initPageTestData();
+
+        // Author has index by name but no data
+        schema.indexLabel("authorByName")
+              .onV("author")
+              .by("name")
+              .secondary()
+              .ifNotExist()
+              .create();
+
+        GraphTraversal<Vertex, Vertex> itor = g.V().has("name", "marko")
+                                               .has("~page", "").limit(1);
+        Assert.assertEquals(1, IteratorUtils.count(itor));
+
+        String page = TraversalUtil.page(itor);
+        List<Vertex> vertices;
+
+        vertices = g.V().has("name", "marko")
+                    .has("~page", page).limit(1).toList();
+        Assert.assertEquals(1, vertices.size());
+        Vertex vertex1 = vertices.get(0);
+
+        vertices = g.V().has("name", "marko")
+                    .has("~page", page).limit(9).toList();
+        Assert.assertEquals(9, vertices.size());
+        Assert.assertTrue(vertices.contains(vertex1));
+
+        vertices = g.V().has("name", "marko")
+                    .has("~page", page).limit(18).toList();
+        Assert.assertEquals(18, vertices.size());
+        Vertex vertex3 = vertices.get(17);
+
+        vertices = g.V().has("name", "marko")
+                    .has("~page", page).limit(40).toList();
+        Assert.assertEquals(33, vertices.size());
+        Assert.assertTrue(vertices.contains(vertex3));
+    }
+
     @Test
     public void testQueryBySecondaryIndexWithLimitAndOffset() {
         initPersonIndex(true);
@@ -3610,6 +4095,85 @@ private void init5Computers() {
         graph.tx().commit();
     }
 
+    private void initPageTestData() {
+        SchemaManager schema = graph().schema();
+        schema.propertyKey("lang").asText().ifNotExist().create();
+
+        schema.vertexLabel("programmer")
+              .properties("name", "age", "city")
+              .useCustomizeStringId()
+              .nullableKeys("age")
+              .ifNotExist()
+              .create();
+
+        schema.vertexLabel("software")
+              .properties("name", "lang", "price")
+              .useCustomizeStringId()
+              .nullableKeys("price")
+              .ifNotExist()
+              .create();
+
+        schema.indexLabel("programmerByNameAndAge")
+              .onV("programmer")
+              .by("name", "age")
+              .secondary()
+              .ifNotExist()
+              .create();
+
+        schema.indexLabel("programmerByCity")
+              .onV("programmer")
+              .search()
+              .by("city")
+              .ifNotExist()
+              .create();
+
+        schema.indexLabel("softwareByName")
+              .onV("software")
+              .secondary()
+              .by("name")
+              .ifNotExist()
+              .create();
+
+        schema.indexLabel("softwareByLang")
+              .onV("software")
+              .secondary()
+              .by("lang")
+              .ifNotExist()
+              .create();
+
+        schema.indexLabel("softwareByPrice")
+              .onV("software")
+              .by("price")
+              .range()
+              .ifNotExist()
+              .create();
+
+        String[] cities = {"Beijing Haidian", "Beijing Chaoyang", "Shanghai"};
+        for (int i = 1; i <= 18; i++) {
+            String id = "p_marko" + i;
+            /*
+             * The city of each programmer is:
+             * [1, 6]: Beijing Haidian, [7, 12]: Beijing Chaoyang,
+             * [13, 18]: Shanghai
+             */
+            String city = cities[(i - 1) / 6];
+            graph().addVertex(T.label, "programmer", T.id, id, "name", "marko",
+                              "age", 30, "city", city);
+        }
+
+        for (int i = 1; i <= 16; i++) {
+            String id = "s_marko" + i;
+            /*
+             * The price of each software is:
+             * [1, 4]: 100, [5, 8]: 200, [9, 12]: 300, [13, 16]: 400
+             */
+            int price = ((i - 1) / 4 + 1) * 100;
+            graph().addVertex(T.label, "software", T.id, id, "name", "marko",
+                              "lang", "java", "price", price);
+        }
+        graph().tx().commit();
+    }
+
     private Vertex vertex(String label, String pkName, Object pkValue) {
         List<Vertex> vertexes = graph().traversal().V()
                                 .hasLabel(label).has(pkName, pkValue)
diff --git a/hugegraph-test/src/main/java/com/baidu/hugegraph/core/VertexLabelCoreTest.java b/hugegraph-test/src/main/java/com/baidu/hugegraph/core/VertexLabelCoreTest.java
index bd527ddf0a..e276dd4ea5 100644
--- a/hugegraph-test/src/main/java/com/baidu/hugegraph/core/VertexLabelCoreTest.java
+++ b/hugegraph-test/src/main/java/com/baidu/hugegraph/core/VertexLabelCoreTest.java
@@ -642,7 +642,7 @@ public void testRemoveVertexLabelWithVertex() {
         graph().tx().commit();
 
         List<Vertex> vertex = graph().traversal().V().hasLabel("person")
-                              .toList();
+                                     .toList();
         Assert.assertNotNull(vertex);
         Assert.assertEquals(3, vertex.size());
 
@@ -679,7 +679,7 @@ public void testRemoveVertexLabelWithVertexAndRangeIndex() {
         graph().tx().commit();
 
         List<Vertex> vertex = graph().traversal().V().hasLabel("person")
-                              .has("age", P.inside(4, 10)).toList();
+                                     .has("age", P.inside(4, 10)).toList();
         Assert.assertNotNull(vertex);
         Assert.assertEquals(2, vertex.size());
 
diff --git a/hugegraph-test/src/main/resources/hugegraph.properties b/hugegraph-test/src/main/resources/hugegraph.properties
index 19fd6f94da..479584a48c 100644
--- a/hugegraph-test/src/main/resources/hugegraph.properties
+++ b/hugegraph-test/src/main/resources/hugegraph.properties
@@ -11,6 +11,8 @@ edge.tx_capacity=10000
 vertex.cache_expire=300
 edge.cache_expire=300
 
+query.page_size=10
+
 # cassandra backend config
 cassandra.host=127.0.0.1
 cassandra.port=9042
diff --git a/pom.xml b/pom.xml
index 4376904cbb..796be2e61d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -4,7 +4,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.baidu.hugegraph</groupId>
     <artifactId>hugegraph</artifactId>
-    <version>0.9.1</version>
+    <version>0.9.2</version>
     <packaging>pom</packaging>
     <prerequisites>
         <maven>3.3.9</maven>