Add Bookmark support for mango json queries (#740)

This adds a bookmark that is sent with each query and can be used to
continue a query from a specific key. This will allow users to paginate
mango queries.

6 files changed, 369 insertions, 9 deletions

diff --git a/src/mango/src/mango_cursor.hrl b/src/mango/src/mango_cursor.hrl
index 58782e5f8..956466c89 100644
--- a/src/mango/src/mango_cursor.hrl
+++ b/src/mango/src/mango_cursor.hrl
@@ -20,5 +20,8 @@
     skip = 0,
     fields = undefined,
     user_fun,
-    user_acc
+    user_acc,
+    bookmark,
+    bookmark_docid,
+    bookmark_key
 }).
diff --git a/src/mango/src/mango_cursor_special.erl b/src/mango/src/mango_cursor_special.erl
index 8404bc04b..78cac7f5d 100644
--- a/src/mango/src/mango_cursor_special.erl
+++ b/src/mango/src/mango_cursor_special.erl
@@ -38,6 +38,7 @@ create(Db, Indexes, Selector, Opts) ->
     Limit = couch_util:get_value(limit, Opts, mango_opts:default_limit()),
     Skip = couch_util:get_value(skip, Opts, 0),
     Fields = couch_util:get_value(fields, Opts, all_fields),
+    Bookmark = couch_util:get_value(bookmark, Opts), 
 
     {ok, #cursor{
         db = Db,
@@ -47,7 +48,8 @@ create(Db, Indexes, Selector, Opts) ->
         opts = Opts,
         limit = Limit,
         skip = Skip,
-        fields = Fields
+        fields = Fields,
+        bookmark = Bookmark
     }}.
 
 
diff --git a/src/mango/src/mango_cursor_view.erl b/src/mango/src/mango_cursor_view.erl
index eb07bf898..5f3c7e9d2 100644
--- a/src/mango/src/mango_cursor_view.erl
+++ b/src/mango/src/mango_cursor_view.erl
@@ -39,6 +39,7 @@ create(Db, Indexes, Selector, Opts) ->
     Limit = couch_util:get_value(limit, Opts, mango_opts:default_limit()),
     Skip = couch_util:get_value(skip, Opts, 0),
     Fields = couch_util:get_value(fields, Opts, all_fields),
+    Bookmark = couch_util:get_value(bookmark, Opts), 
 
     {ok, #cursor{
         db = Db,
@@ -48,7 +49,8 @@ create(Db, Indexes, Selector, Opts) ->
         opts = Opts,
         limit = Limit,
         skip = Skip,
-        fields = Fields
+        fields = Fields,
+        bookmark = Bookmark
     }}.
 
 
@@ -85,8 +87,9 @@ execute(#cursor{db = Db, index = Idx} = Cursor0, UserFun, UserAcc) ->
                 end_key = mango_idx:end_key(Idx, Cursor#cursor.ranges),
                 include_docs = true
             },
-            #cursor{opts = Opts} = Cursor,
-            Args = apply_opts(Opts, BaseArgs),
+            #cursor{opts = Opts, bookmark = Bookmark} = Cursor,
+            Args0 = apply_opts(Opts, BaseArgs),
+            Args = mango_json_bookmark:update_args(Bookmark, Args0),
             UserCtx = couch_util:get_value(user_ctx, Opts, #user_ctx{}),
             DbOpts = [{user_ctx, UserCtx}],
             Result = case mango_idx:def(Idx) of
@@ -102,7 +105,10 @@ execute(#cursor{db = Db, index = Idx} = Cursor0, UserFun, UserAcc) ->
             end,
             case Result of
                 {ok, LastCursor} ->
-                    {ok, LastCursor#cursor.user_acc};
+                    NewBookmark = mango_json_bookmark:create(LastCursor),
+                    Arg = {add_key, bookmark, NewBookmark},
+                    {_Go, FinalUserAcc} = UserFun(Arg, LastCursor#cursor.user_acc),
+                    {ok, FinalUserAcc};
                 {error, Reason} ->
                     {error, Reason}
             end
@@ -180,8 +186,9 @@ handle_message({row, Props}, Cursor) ->
         {ok, Doc} ->
             case mango_selector:match(Cursor#cursor.selector, Doc) of
                 true ->
-                    FinalDoc = mango_fields:extract(Doc, Cursor#cursor.fields),
-                    handle_doc(Cursor, FinalDoc);
+                    Cursor1 = update_bookmark_keys(Cursor, Props),
+                    FinalDoc = mango_fields:extract(Doc, Cursor1#cursor.fields),
+                    handle_doc(Cursor1, FinalDoc);
                 false ->
                     {ok, Cursor}
             end;
@@ -286,6 +293,9 @@ apply_opts([{update, false} | Rest], Args) ->
         update = false
     },
     apply_opts(Rest, NewArgs);
+% apply_opts([{bookmark, Bookmark} | Rest], Args) when Bookmark =/= nil ->
+%     NewArgs = mango_json_bookmark:update_args(Bookmark, Args),
+%     apply_opts(Rest, NewArgs);
 apply_opts([{_, _} | Rest], Args) ->
     % Ignore unknown options
     apply_opts(Rest, Args).
@@ -310,3 +320,14 @@ is_design_doc(RowProps) ->
         <<"_design/", _/binary>> -> true;
         _ -> false
     end.
+
+
+update_bookmark_keys(#cursor{limit = Limit} = Cursor, Props) when Limit > 0 ->
+    Id = couch_util:get_value(id, Props), 
+    Key = couch_util:get_value(key, Props), 
+    Cursor#cursor {
+        bookmark_docid = Id,
+        bookmark_key = Key
+    };
+update_bookmark_keys(Cursor, _Props) ->
+    Cursor.
diff --git a/src/mango/src/mango_error.erl b/src/mango/src/mango_error.erl
index 7d77b5e9a..361016524 100644
--- a/src/mango/src/mango_error.erl
+++ b/src/mango/src/mango_error.erl
@@ -46,11 +46,18 @@ info(mango_cursor, {no_usable_index, selector_unsupported}) ->
         <<"There is no index available for this selector.">>
     };
 
+info(mango_json_bookmark, {invalid_bookmark, BadBookmark}) ->
+    {
+        400,
+        <<"invalid_bookmark">>,
+        fmt("Invalid bookmark value: ~s", [?JSON_ENCODE(BadBookmark)])
+    };
+
 info(mango_cursor_text, {invalid_bookmark, BadBookmark}) ->
     {
         400,
         <<"invalid_bookmark">>,
-        fmt("Invalid boomkark value: ~s", [?JSON_ENCODE(BadBookmark)])
+        fmt("Invalid bookmark value: ~s", [?JSON_ENCODE(BadBookmark)])
     };
 info(mango_cursor_text, multiple_text_indexes) ->
     {
diff --git a/src/mango/src/mango_json_bookmark.erl b/src/mango/src/mango_json_bookmark.erl
new file mode 100644
index 000000000..97f81cfb8
--- /dev/null
+++ b/src/mango/src/mango_json_bookmark.erl
@@ -0,0 +1,71 @@
+% Licensed 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.
+
+
+-module(mango_json_bookmark).
+
+-export([
+    update_args/2,
+    create/1
+]).
+
+
+-include_lib("couch_mrview/include/couch_mrview.hrl").
+-include("mango_cursor.hrl").
+-include("mango.hrl").
+
+update_args(EncodedBookmark,  #mrargs{skip = Skip} = Args) ->
+    Bookmark = unpack(EncodedBookmark),
+    case is_list(Bookmark) of
+        true -> 
+            {startkey, Startkey} = lists:keyfind(startkey, 1, Bookmark),
+            {startkey_docid, StartkeyDocId} = lists:keyfind(startkey_docid, 1, Bookmark),
+            Args#mrargs{
+                start_key = Startkey,
+                start_key_docid = StartkeyDocId,
+                skip = 1 + Skip
+            };
+        false ->
+            Args
+    end.
+    
+
+create(#cursor{bookmark_docid = BookmarkDocId, bookmark_key = BookmarkKey}) when BookmarkKey =/= undefined ->
+    QueryArgs = [
+        {startkey_docid, BookmarkDocId},
+        {startkey, BookmarkKey}
+    ],
+    Bin = term_to_binary(QueryArgs, [compressed, {minor_version,1}]),
+    couch_util:encodeBase64Url(Bin);
+create(#cursor{bookmark = Bookmark}) ->
+    Bookmark.
+
+
+unpack(nil) ->
+    nil;
+unpack(Packed) ->
+    try
+        Bookmark = binary_to_term(couch_util:decodeBase64Url(Packed)),
+        verify(Bookmark)
+    catch _:_ ->
+        ?MANGO_ERROR({invalid_bookmark, Packed})
+    end.
+
+verify(Bookmark) when is_list(Bookmark) ->
+    case lists:keymember(startkey, 1, Bookmark) andalso lists:keymember(startkey_docid, 1, Bookmark) of
+        true -> Bookmark;
+        _ -> throw(invalid_bookmark)
+    end;
+verify(_Bookmark) ->
+    throw(invalid_bookmark).
+
+   
\ No newline at end of file
diff --git a/src/mango/test/14-json-pagination.py b/src/mango/test/14-json-pagination.py
new file mode 100644
index 000000000..ddac15662
--- /dev/null
+++ b/src/mango/test/14-json-pagination.py
@@ -0,0 +1,256 @@
+# Licensed 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.
+
+import mango
+import copy
+
+DOCS = [
+    {
+        "_id": "100",
+        "name": "Jimi",
+        "location": "AUS",
+        "user_id": 1,
+        "same": "value"
+    },
+    {
+        "_id": "200",
+        "name": "Eddie",
+        "location": "BRA",
+        "user_id": 2,
+        "same": "value"
+    },
+    {
+        "_id": "300",
+        "name": "Harry",
+        "location": "CAN",
+        "user_id":3,
+        "same": "value"
+    },
+    {
+        "_id": "400",
+        "name": "Eddie",
+        "location": "DEN",
+        "user_id":4,
+        "same": "value"
+    },
+    {
+        "_id": "500",
+        "name": "Jones",
+        "location": "ETH",
+        "user_id":5,
+        "same": "value"
+    },
+    {
+        "_id": "600",
+        "name": "Winnifried",
+        "location": "FRA",
+        "user_id":6,
+        "same": "value"
+    },
+    {
+        "_id": "700",
+        "name": "Marilyn",
+        "location": "GHA",
+        "user_id":7,
+        "same": "value"
+    },
+    {
+        "_id": "800",
+        "name": "Sandra",
+        "location": "ZAR",
+        "user_id":8,
+        "same": "value"
+    },
+]
+
+class PaginateJsonDocs(mango.DbPerClass):
+    def setUp(self):
+        self.db.recreate()
+        self.db.save_docs(copy.deepcopy(DOCS))
+
+    def test_all_docs_paginate_to_end(self):
+        selector = {"_id": {"$gt": 0}}  
+        # Page 1
+        resp = self.db.find(selector, fields=["_id"], limit=5, return_raw=True)
+        bookmark = resp['bookmark']
+        docs = resp['docs']
+        assert docs[0]['_id'] == '100'
+        assert len(docs) == 5
+
+        # Page 2
+        resp = self.db.find(selector, fields=["_id"], bookmark= bookmark, limit=5, return_raw=True)
+        bookmark = resp['bookmark']
+        docs = resp['docs']
+        assert docs[0]['_id'] == '600'
+        assert len(docs) == 3
+
+        # Page 3 
+        resp = self.db.find(selector, bookmark= bookmark, limit=5, return_raw=True)
+        bookmark = resp['bookmark']
+        docs = resp['docs']
+        assert len(docs) == 0
+
+    def test_return_previous_bookmark_for_empty(self):
+        selector = {"_id": {"$gt": 0}}  
+        # Page 1
+        resp = self.db.find(selector, fields=["_id"], return_raw=True)
+        bookmark1 = resp['bookmark']
+        docs = resp['docs']
+        assert len(docs) == 8
+
+        resp = self.db.find(selector, fields=["_id"], return_raw=True, bookmark=bookmark1)
+        bookmark2 = resp['bookmark']
+        docs = resp['docs']
+        assert len(docs) == 0
+
+        resp = self.db.find(selector, fields=["_id"], return_raw=True, bookmark=bookmark2)
+        bookmark3 = resp['bookmark']
+        docs = resp['docs']
+        assert bookmark3 == bookmark2
+        assert len(docs) == 0
+
+    def test_all_docs_with_skip(self):
+        selector = {"_id": {"$gt": 0}}  
+        # Page 1
+        resp = self.db.find(selector, fields=["_id"], skip=2, limit=5, return_raw=True)
+        bookmark = resp['bookmark']
+        docs = resp['docs']
+        assert docs[0]['_id'] == '300'
+        assert len(docs) == 5
+
+        # Page 2
+        resp = self.db.find(selector, fields=["_id"], bookmark= bookmark, limit=5, return_raw=True)
+        bookmark = resp['bookmark']
+        docs = resp['docs']
+        assert docs[0]['_id'] == '800'
+        assert len(docs) == 1
+        resp = self.db.find(selector, bookmark= bookmark, limit=5, return_raw=True)
+        bookmark = resp['bookmark']
+        docs = resp['docs']
+        assert len(docs) == 0
+
+    def test_all_docs_reverse(self):
+        selector = {"_id": {"$gt": 0}} 
+        resp = self.db.find(selector, fields=["_id"], sort=[{"_id": "desc"}], limit=5, return_raw=True)
+        docs = resp['docs']
+        bookmark1 = resp["bookmark"]
+        assert len(docs) == 5
+        assert docs[0]['_id'] == '800'
+
+        resp = self.db.find(selector, fields=["_id"], sort=[{"_id": "desc"}], limit=5, return_raw=True, bookmark=bookmark1)
+        docs = resp['docs']
+        bookmark2 = resp["bookmark"]
+        assert len(docs) == 3
+        assert docs[0]['_id'] == '300'
+
+        resp = self.db.find(selector, fields=["_id"], sort=[{"_id": "desc"}], limit=5, return_raw=True, bookmark=bookmark2)
+        docs = resp['docs']
+        assert len(docs) == 0
+
+    def test_bad_bookmark(self):
+        try:
+            self.db.find({"_id": {"$gt": 0}}, bookmark="bad-bookmark")
+        except Exception, e:
+            resp = e.response.json()
+            assert resp["error"] == "invalid_bookmark"
+            assert resp["reason"] == "Invalid bookmark value: \"bad-bookmark\""
+            assert e.response.status_code == 400
+        else:
+            raise AssertionError("Should have thrown error for bad bookmark")
+    
+    def test_throws_error_on_text_bookmark(self):
+        bookmark = 'g2wAAAABaANkABFub2RlMUBjb3VjaGRiLm5ldGwAAAACYQBiP____2poAkY_8AAAAAAAAGEHag'
+        try:
+            self.db.find({"_id": {"$gt": 0}}, bookmark=bookmark)
+        except Exception, e:
+            resp = e.response.json()
+            assert resp["error"] == "invalid_bookmark"
+            assert e.response.status_code == 400
+        else:
+            raise AssertionError("Should have thrown error for bad bookmark")
+    
+    def test_index_pagination(self):
+        self.db.create_index(["location"])
+        selector = {"location": {"$gt": "A"}} 
+        resp = self.db.find(selector, fields=["_id"], limit=5, return_raw=True)
+        docs = resp['docs']
+        bookmark1 = resp["bookmark"]
+        assert len(docs) == 5
+        assert docs[0]['_id'] == '100'
+
+        resp = self.db.find(selector, fields=["_id"], limit=5, return_raw=True, bookmark=bookmark1)
+        docs = resp['docs']
+        bookmark2 = resp["bookmark"]
+        assert len(docs) == 3
+        assert docs[0]['_id'] == '600'
+
+        resp = self.db.find(selector, fields=["_id"], limit=5, return_raw=True, bookmark=bookmark2)
+        docs = resp['docs']
+        assert len(docs) == 0
+
+    def test_index_pagination_two_keys(self):
+        self.db.create_index(["location", "user_id"])
+        selector = {"location": {"$gt": "A"}, "user_id": {"$gte": 1}} 
+        resp = self.db.find(selector, fields=["_id"], limit=5, return_raw=True)
+        docs = resp['docs']
+        bookmark1 = resp["bookmark"]
+        assert len(docs) == 5
+        assert docs[0]['_id'] == '100'
+
+        resp = self.db.find(selector, fields=["_id"], limit=5, return_raw=True, bookmark=bookmark1)
+        docs = resp['docs']
+        bookmark2 = resp["bookmark"]
+        assert len(docs) == 3
+        assert docs[0]['_id'] == '600'
+
+        resp = self.db.find(selector, fields=["_id"], limit=5, return_raw=True, bookmark=bookmark2)
+        docs = resp['docs']
+        assert len(docs) == 0
+
+    def test_index_pagination_reverse(self):
+        self.db.create_index(["location", "user_id"])
+        selector = {"location": {"$gt": "A"}, "user_id": {"$gte": 1}} 
+        sort = [{"location": "desc"}, {"user_id": "desc"}]
+        resp = self.db.find(selector, fields=["_id"], sort=sort, limit=5, return_raw=True)
+        docs = resp['docs']
+        bookmark1 = resp["bookmark"]
+        assert len(docs) == 5
+        assert docs[0]['_id'] == '800'
+
+        resp = self.db.find(selector, fields=["_id"], limit=5, sort=sort, return_raw=True, bookmark=bookmark1)
+        docs = resp['docs']
+        bookmark2 = resp["bookmark"]
+        assert len(docs) == 3
+        assert docs[0]['_id'] == '300'
+
+        resp = self.db.find(selector, fields=["_id"], limit=5, sort=sort, return_raw=True, bookmark=bookmark2)
+        docs = resp['docs']
+        assert len(docs) == 0
+
+    def test_index_pagination_same_emitted_key(self):
+        self.db.create_index(["same"])
+        selector = {"same": {"$gt": ""}} 
+        resp = self.db.find(selector, fields=["_id"], limit=5, return_raw=True)
+        docs = resp['docs']
+        bookmark1 = resp["bookmark"]
+        assert len(docs) == 5
+        assert docs[0]['_id'] == '100'
+
+        resp = self.db.find(selector, fields=["_id"], limit=5, return_raw=True, bookmark=bookmark1)
+        docs = resp['docs']
+        bookmark2 = resp["bookmark"]
+        assert len(docs) == 3
+        assert docs[0]['_id'] == '600'
+
+        resp = self.db.find(selector, fields=["_id"], limit=5, return_raw=True, bookmark=bookmark2)
+        docs = resp['docs']
+        assert len(docs) == 0