Implement pagination API

6 files changed, 600 insertions, 45 deletions

diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index 4776ac10d..5cfbd1d5f 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -812,14 +812,20 @@ db_req(#httpd{path_parts=[_, DocId | FileNameParts]}=Req, Db) ->
     db_attachment_req(Req, Db, DocId, FileNameParts).
 
 multi_all_docs_view(Req, Db, OP, Queries) ->
-    Args0 = couch_mrview_http:parse_params(Req, undefined),
+    Args = couch_views_http:parse_params(Req, undefined),
+    case couch_views_util:is_paginated(Args) of
+        false ->
+            stream_multi_all_docs_view(Req, Db, OP, Args, Queries);
+        true ->
+            paginate_multi_all_docs_view(Req, Db, OP, Args, Queries)
+    end.
+
+
+stream_multi_all_docs_view(Req, Db, OP, Args0, Queries) ->
     Args1 = Args0#mrargs{view_type=map},
-    ArgQueries = lists:map(fun({Query}) ->
-        QueryArg1 = couch_mrview_http:parse_params(Query, undefined,
-            Args1, [decoded]),
-        QueryArgs2 = couch_views_util:validate_args(QueryArg1),
-        set_namespace(OP, QueryArgs2)
-    end, Queries),
+    ArgQueries = chttpd_view:parse_queries(Req, Args1, Queries, fun(QArgs) ->
+        set_namespace(OP, QArgs)
+    end),
     Max = chttpd:chunked_response_buffer_size(),
     First = "{\"results\":[",
     {ok, Resp0} = chttpd:start_delayed_json_response(Req, 200, [], First),
@@ -842,8 +848,34 @@ multi_all_docs_view(Req, Db, OP, Queries) ->
     chttpd:end_delayed_json_response(Resp1).
 
 
+paginate_multi_all_docs_view(Req, Db, OP, Args0, Queries) ->
+    Args1 = Args0#mrargs{view_type=map},
+    ArgQueries = chttpd_view:parse_queries(Req, Args1, Queries, fun(QArgs) ->
+        set_namespace(OP, QArgs)
+    end),
+    KeyFun = fun({Props}) -> couch_util:get_value(id, Props) end,
+    #mrargs{page_size = PageSize} = Args0,
+    #httpd{path_parts = Parts} = Req,
+    UpdateSeq = fabric2_db:get_update_seq(Db),
+    EtagTerm = {Parts, UpdateSeq, Args0},
+    Response = couch_views_http:paginated(
+        Req, EtagTerm, PageSize, ArgQueries, KeyFun,
+        fun(Args) ->
+            all_docs_paginated_cb(Db, Args)
+        end),
+    chttpd:send_json(Req, Response).
+
+
 all_docs_view(Req, Db, Keys, OP) ->
-    Args0 = couch_mrview_http:parse_body_and_query(Req, Keys),
+    Args = couch_views_http:parse_body_and_query(Req, Keys),
+    case couch_views_util:is_paginated(Args) of
+        false ->
+            stream_all_docs_view(Req, Db, Args, OP);
+        true ->
+            paginate_all_docs_view(Req, Db, Args, OP)
+    end.
+
+stream_all_docs_view(Req, Db, Args0, OP) ->
     Args1 = Args0#mrargs{view_type=map},
     Args2 = couch_views_util:validate_args(Args1),
     Args3 = set_namespace(OP, Args2),
@@ -864,15 +896,46 @@ all_docs_view(Req, Db, Keys, OP) ->
     end.
 
 
+paginate_all_docs_view(Req, Db, Args0, OP) ->
+    Args1 = Args0#mrargs{view_type=map},
+    Args2 = chttpd_view:validate_args(Req, Args1),
+    Args3 = set_namespace(OP, Args2),
+    KeyFun = fun({Props}) -> couch_util:get_value(id, Props) end,
+    #httpd{path_parts = Parts} = Req,
+    UpdateSeq = fabric2_db:get_update_seq(Db),
+    EtagTerm = {Parts, UpdateSeq, Args3},
+    Response = couch_views_http:paginated(
+        Req, EtagTerm, Args3, KeyFun,
+        fun(Args) ->
+            all_docs_paginated_cb(Db, Args)
+        end),
+    chttpd:send_json(Req, Response).
+
+
+all_docs_paginated_cb(Db, Args) ->
+    #vacc{meta=MetaMap, buffer=Items} = case Args#mrargs.keys of
+        undefined ->
+            send_all_docs(Db, Args, #vacc{paginated=true});
+        Keys when is_list(Keys) ->
+            send_all_docs_keys(Db, Args, #vacc{paginated=true})
+    end,
+    {MetaMap, Items}.
+
+
 send_all_docs(Db, #mrargs{keys = undefined} = Args, VAcc0) ->
     Opts0 = fabric2_util:all_docs_view_opts(Args),
-    Opts = Opts0 ++ [{restart_tx, true}],
-    NS = couch_util:get_value(namespace, Opts),
+    NS = couch_util:get_value(namespace, Opts0),
     FoldFun = case NS of
         <<"_all_docs">> -> fold_docs;
         <<"_design">> -> fold_design_docs;
         <<"_local">> -> fold_local_docs
     end,
+    Opts = case couch_views_util:is_paginated(Args) of
+        false ->
+            Opts0 ++ [{restart_tx, true}];
+        true ->
+            Opts0
+    end,
     ViewCb = fun view_cb/2,
     Acc = {iter, Db, Args, VAcc0},
     {ok, {iter, _, _, VAcc1}} = fabric2_db:FoldFun(Db, ViewCb, Acc, Opts),
@@ -980,25 +1043,15 @@ view_cb({row, Row}, {iter, Db, Args, VAcc}) ->
             Row
     end,
     chttpd_stats:incr_rows(),
-    {Go, NewVAcc} = couch_mrview_http:view_cb({row, NewRow}, VAcc),
+    {Go, NewVAcc} = couch_views_http:view_cb({row, NewRow}, VAcc),
     {Go, {iter, Db, Args, NewVAcc}};
 
 view_cb(Msg, {iter, Db, Args, VAcc}) ->
-    {Go, NewVAcc} = couch_mrview_http:view_cb(Msg, VAcc),
+    {Go, NewVAcc} = couch_views_http:view_cb(Msg, VAcc),
     {Go, {iter, Db, Args, NewVAcc}};
 
-view_cb({row, Row} = Msg, Acc) ->
-    case lists:keymember(doc, 1, Row) of
-        true ->
-            chttpd_stats:incr_reads();
-        false ->
-            ok
-    end,
-    chttpd_stats:incr_rows(),
-    couch_mrview_http:view_cb(Msg, Acc);
-
 view_cb(Msg, Acc) ->
-    couch_mrview_http:view_cb(Msg, Acc).
+    couch_views_http:view_cb(Msg, Acc).
 
 db_doc_req(#httpd{method='DELETE'}=Req, Db, DocId) ->
     % check for the existence of the doc to handle the 404 case.
diff --git a/src/chttpd/src/chttpd_view.erl b/src/chttpd/src/chttpd_view.erl
index c9340fbe2..8e2a08e2b 100644
--- a/src/chttpd/src/chttpd_view.erl
+++ b/src/chttpd/src/chttpd_view.erl
@@ -14,18 +14,32 @@
 -include_lib("couch/include/couch_db.hrl").
 -include_lib("couch_mrview/include/couch_mrview.hrl").
 
--export([handle_view_req/3]).
+-export([
+    handle_view_req/3,
+    validate_args/2,
+    parse_queries/4,
+    view_cb/2
+]).
+
+-define(DEFAULT_ALL_DOCS_PAGE_SIZE, 2000).
+-define(DEFAULT_VIEWS_PAGE_SIZE, 2000).
 
 multi_query_view(Req, Db, DDoc, ViewName, Queries) ->
-    Args0 = couch_mrview_http:parse_params(Req, undefined),
+    Args = couch_views_http:parse_params(Req, undefined),
+    case couch_views_util:is_paginated(Args) of
+        false ->
+            stream_multi_query_view(Req, Db, DDoc, ViewName, Args, Queries);
+        true ->
+            paginate_multi_query_view(Req, Db, DDoc, ViewName, Args, Queries)
+    end.
+
+
+stream_multi_query_view(Req, Db, DDoc, ViewName, Args0, Queries) ->
     {ok, #mrst{views=Views}} = couch_mrview_util:ddoc_to_mrst(Db, DDoc),
     Args1 = couch_mrview_util:set_view_type(Args0, ViewName, Views),
-    ArgQueries = lists:map(fun({Query}) ->
-        QueryArg = couch_mrview_http:parse_params(Query, undefined,
-            Args1, [decoded]),
-        QueryArg1 = couch_mrview_util:set_view_type(QueryArg, ViewName, Views),
-        fabric_util:validate_args(Db, DDoc, QueryArg1)
-    end, Queries),
+    ArgQueries = parse_queries(Req, Args1, Queries, fun(QueryArg) ->
+        couch_mrview_util:set_view_type(QueryArg, ViewName, Views)
+    end),
     VAcc0 = #vacc{db=Db, req=Req, prepend="\r\n"},
     FirstChunk = "{\"results\":[",
     {ok, Resp0} = chttpd:start_delayed_json_response(VAcc0#vacc.req, 200, [], FirstChunk),
@@ -38,15 +52,46 @@ multi_query_view(Req, Db, DDoc, ViewName, Queries) ->
     {ok, Resp1} = chttpd:send_delayed_chunk(VAcc2#vacc.resp, "\r\n]}"),
     chttpd:end_delayed_json_response(Resp1).
 
+
+paginate_multi_query_view(Req, Db, DDoc, ViewName, Args0, Queries) ->
+    {ok, #mrst{views=Views}} = couch_mrview_util:ddoc_to_mrst(Db, DDoc),
+    ArgQueries = parse_queries(Req, Args0, Queries, fun(QueryArg) ->
+        couch_mrview_util:set_view_type(QueryArg, ViewName, Views)
+    end),
+    KeyFun = fun({Props}) -> couch_util:get_value(id, Props) end,
+    #mrargs{page_size = PageSize} = Args0,
+    #httpd{path_parts = Parts} = Req,
+    UpdateSeq = fabric2_db:get_update_seq(Db),
+    EtagTerm = {Parts, UpdateSeq, Args0},
+    Response = couch_views_http:paginated(
+        Req, EtagTerm, PageSize, ArgQueries, KeyFun,
+        fun(Args) ->
+           {ok, #vacc{meta=MetaMap, buffer=Items}} = couch_views:query(
+               Db, DDoc, ViewName, fun view_cb/2, #vacc{paginated=true}, Args),
+           {MetaMap, Items}
+        end),
+    chttpd:send_json(Req, Response).
+
+
 design_doc_post_view(Req, Props, Db, DDoc, ViewName, Keys) ->
     Args = couch_mrview_http:parse_body_and_query(Req, Props, Keys),
     fabric_query_view(Db, Req, DDoc, ViewName, Args).
 
 design_doc_view(Req, Db, DDoc, ViewName, Keys) ->
-    Args = couch_mrview_http:parse_params(Req, Keys),
+    Args = couch_views_http:parse_params(Req, Keys),
     fabric_query_view(Db, Req, DDoc, ViewName, Args).
 
+
 fabric_query_view(Db, Req, DDoc, ViewName, Args) ->
+    case couch_views_util:is_paginated(Args) of
+        false ->
+            stream_fabric_query_view(Db, Req, DDoc, ViewName, Args);
+        true ->
+            paginate_fabric_query_view(Db, Req, DDoc, ViewName, Args)
+    end.
+
+
+stream_fabric_query_view(Db, Req, DDoc, ViewName, Args) ->
     Max = chttpd:chunked_response_buffer_size(),
     Fun = fun view_cb/2,
     VAcc = #vacc{db=Db, req=Req, threshold=Max},
@@ -54,16 +99,31 @@ fabric_query_view(Db, Req, DDoc, ViewName, Args) ->
     {ok, Resp#vacc.resp}.
 
 
+paginate_fabric_query_view(Db, Req, DDoc, ViewName, Args0) ->
+    KeyFun = fun({Props}) -> couch_util:get_value(id, Props) end,
+    #httpd{path_parts = Parts} = Req,
+    UpdateSeq = fabric2_db:get_update_seq(Db),
+    ETagTerm = {Parts, UpdateSeq, Args0},
+    Response = couch_views_http:paginated(
+        Req, ETagTerm, Args0, KeyFun,
+        fun(Args) ->
+            VAcc0 = #vacc{paginated=true},
+            {ok, VAcc1} = couch_views:query(Db, DDoc, ViewName, fun view_cb/2, VAcc0, Args),
+            #vacc{meta=Meta, buffer=Items} = VAcc1,
+            {Meta, Items}
+        end),
+    chttpd:send_json(Req, Response).
+
 view_cb({row, Row} = Msg, Acc) ->
     case lists:keymember(doc, 1, Row) of
         true -> chttpd_stats:incr_reads();
         false -> ok
     end,
     chttpd_stats:incr_rows(),
-    couch_mrview_http:view_cb(Msg, Acc);
+    couch_views_http:view_cb(Msg, Acc);
 
 view_cb(Msg, Acc) ->
-    couch_mrview_http:view_cb(Msg, Acc).
+    couch_views_http:view_cb(Msg, Acc).
 
 
 handle_view_req(#httpd{method='POST',
@@ -111,6 +171,86 @@ assert_no_queries_param(_) ->
     }).
 
 
+validate_args(Req, #mrargs{page_size = PageSize} = Args) when is_integer(PageSize) ->
+    MaxPageSize = max_page_size(Req),
+    couch_views_util:validate_args(Args, [{page_size, MaxPageSize}]);
+
+validate_args(_Req, #mrargs{} = Args) ->
+    couch_views_util:validate_args(Args, []).
+
+
+max_page_size(#httpd{path_parts=[_Db, <<"_all_docs">>, <<"queries">>]}) ->
+    config:get_integer(
+        "request_limits", "_all_docs/queries", ?DEFAULT_ALL_DOCS_PAGE_SIZE);
+
+max_page_size(#httpd{path_parts=[_Db, <<"_all_docs">>]}) ->
+    config:get_integer(
+        "request_limits", "_all_docs", ?DEFAULT_ALL_DOCS_PAGE_SIZE);
+
+max_page_size(#httpd{path_parts=[_Db, <<"_local_docs">>, <<"queries">>]}) ->
+    config:get_integer(
+        "request_limits", "_all_docs/queries", ?DEFAULT_ALL_DOCS_PAGE_SIZE);
+
+max_page_size(#httpd{path_parts=[_Db, <<"_local_docs">>]}) ->
+    config:get_integer(
+        "request_limits", "_all_docs", ?DEFAULT_ALL_DOCS_PAGE_SIZE);
+
+max_page_size(#httpd{path_parts=[_Db, <<"_design_docs">>, <<"queries">>]}) ->
+    config:get_integer(
+        "request_limits", "_all_docs/queries", ?DEFAULT_ALL_DOCS_PAGE_SIZE);
+
+max_page_size(#httpd{path_parts=[_Db, <<"_design_docs">>]}) ->
+    config:get_integer(
+        "request_limits", "_all_docs", ?DEFAULT_ALL_DOCS_PAGE_SIZE);
+
+max_page_size(#httpd{path_parts=[
+        _Db, <<"_design">>, _DDocName, <<"_view">>, _View, <<"queries">>]}) ->
+    config:get_integer(
+        "request_limits", "_view/queries", ?DEFAULT_VIEWS_PAGE_SIZE);
+
+max_page_size(#httpd{path_parts=[
+        _Db, <<"_design">>, _DDocName, <<"_view">>, _View]}) ->
+    config:get_integer(
+        "request_limits", "_view", ?DEFAULT_VIEWS_PAGE_SIZE).
+
+
+parse_queries(Req, #mrargs{page_size = PageSize} = Args0, Queries, Fun)
+        when is_integer(PageSize) ->
+    MaxPageSize = max_page_size(Req),
+    if length(Queries) < PageSize -> ok; true ->
+        throw({
+            query_parse_error,
+            <<"Provided number of queries is more than given page_size">>
+        })
+    end,
+    couch_views_util:validate_args(Fun(Args0), [{page_size, MaxPageSize}]),
+    Args = Args0#mrargs{page_size = undefined},
+    lists:map(fun({Query}) ->
+        Args1 = couch_views_http:parse_params(Query, undefined, Args, [decoded]),
+        if not is_integer(Args1#mrargs.page_size) -> ok; true ->
+            throw({
+               query_parse_error,
+                <<"You cannot specify `page_size` inside the query">>
+            })
+        end,
+        Args2 = maybe_set_page_size(Args1, MaxPageSize),
+        couch_views_util:validate_args(Fun(Args2), [{page_size, MaxPageSize}])
+    end, Queries);
+
+parse_queries(_Req, #mrargs{} = Args, Queries, Fun) ->
+    lists:map(fun({Query}) ->
+        Args1 = couch_views_http:parse_params(Query, undefined, Args, [decoded]),
+        couch_views_util:validate_args(Fun(Args1))
+    end, Queries).
+
+
+maybe_set_page_size(#mrargs{page_size = undefined} = Args, MaxPageSize) ->
+    Args#mrargs{page_size = MaxPageSize};
+
+maybe_set_page_size(#mrargs{} = Args, _MaxPageSize) ->
+    Args.
+
+
 -ifdef(TEST).
 
 -include_lib("eunit/include/eunit.hrl").
diff --git a/src/couch_mrview/include/couch_mrview.hrl b/src/couch_mrview/include/couch_mrview.hrl
index bb0ab0b46..e0f80df81 100644
--- a/src/couch_mrview/include/couch_mrview.hrl
+++ b/src/couch_mrview/include/couch_mrview.hrl
@@ -81,7 +81,9 @@
     conflicts,
     callback,
     sorted = true,
-    extra = []
+    extra = [],
+    page_size = undefined,
+    bookmark=nil
 }).
 
 -record(vacc, {
@@ -95,7 +97,9 @@
     bufsize = 0,
     threshold = 1490,
     row_sent = false,
-    meta_sent = false
+    meta_sent = false,
+    paginated = false,
+    meta = #{}
 }).
 
 -record(lacc, {
diff --git a/src/couch_mrview/src/couch_mrview_http.erl b/src/couch_mrview/src/couch_mrview_http.erl
index 3cf8833d7..e1ba9d656 100644
--- a/src/couch_mrview/src/couch_mrview_http.erl
+++ b/src/couch_mrview/src/couch_mrview_http.erl
@@ -35,6 +35,8 @@
     parse_params/3,
     parse_params/4,
     view_cb/2,
+    row_to_obj/1,
+    row_to_obj/2,
     row_to_json/1,
     row_to_json/2,
     check_view_etag/3
@@ -413,11 +415,19 @@ prepend_val(#vacc{prepend=Prepend}) ->
 
 
 row_to_json(Row) ->
+    ?JSON_ENCODE(row_to_obj(Row)).
+
+
+row_to_json(Kind, Row) ->
+    ?JSON_ENCODE(row_to_obj(Kind, Row)).
+
+
+row_to_obj(Row) ->
     Id = couch_util:get_value(id, Row),
-    row_to_json(Id, Row).
+    row_to_obj(Id, Row).
 
 
-row_to_json(error, Row) ->
+row_to_obj(error, Row) ->
     % Special case for _all_docs request with KEYS to
     % match prior behavior.
     Key = couch_util:get_value(key, Row),
@@ -426,9 +436,8 @@ row_to_json(error, Row) ->
     ReasonProp = if Reason == undefined -> []; true ->
         [{reason, Reason}]
     end,
-    Obj = {[{key, Key}, {error, Val}] ++ ReasonProp},
-    ?JSON_ENCODE(Obj);
-row_to_json(Id0, Row) ->
+    {[{key, Key}, {error, Val}] ++ ReasonProp};
+row_to_obj(Id0, Row) ->
     Id = case Id0 of
         undefined -> [];
         Id0 -> [{id, Id0}]
@@ -439,8 +448,7 @@ row_to_json(Id0, Row) ->
         undefined -> [];
         Doc0 -> [{doc, Doc0}]
     end,
-    Obj = {Id ++ [{key, Key}, {value, Val}] ++ Doc},
-    ?JSON_ENCODE(Obj).
+    {Id ++ [{key, Key}, {value, Val}] ++ Doc}.
 
 
 parse_params(#httpd{}=Req, Keys) ->
@@ -523,6 +531,8 @@ parse_param(Key, Val, Args, IsDecoded) ->
             Args#mrargs{end_key_docid=couch_util:to_binary(Val)};
         "limit" ->
             Args#mrargs{limit=parse_pos_int(Val)};
+        "page_size" ->
+            Args#mrargs{page_size=parse_pos_int(Val)};
         "stale" when Val == "ok" orelse Val == <<"ok">> ->
             Args#mrargs{stable=true, update=false};
         "stale" when Val == "update_after" orelse Val == <<"update_after">> ->
diff --git a/src/couch_views/src/couch_views_http.erl b/src/couch_views/src/couch_views_http.erl
new file mode 100644
index 000000000..ae6725649
--- /dev/null
+++ b/src/couch_views/src/couch_views_http.erl
@@ -0,0 +1,292 @@
+% 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(couch_views_http).
+
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_mrview/include/couch_mrview.hrl").
+
+-export([
+    parse_body_and_query/2,
+    parse_body_and_query/3,
+    parse_params/2,
+    parse_params/4,
+    row_to_obj/1,
+    row_to_obj/2,
+    view_cb/2,
+    paginated/5,
+    paginated/6
+]).
+
+-define(BOOKMARK_VSN, 1).
+
+parse_body_and_query(#httpd{method='POST'} = Req, Keys) ->
+    Props = chttpd:json_body_obj(Req),
+    parse_body_and_query(Req, Props, Keys);
+
+parse_body_and_query(Req, Keys) ->
+    parse_params(chttpd:qs(Req), Keys, #mrargs{keys=Keys, group=undefined,
+        group_level=undefined}, [keep_group_level]).
+
+parse_body_and_query(Req, {Props}, Keys) ->
+    Args = #mrargs{keys=Keys, group=undefined, group_level=undefined},
+    BodyArgs = parse_params(Props, Keys, Args, [decoded]),
+    parse_params(chttpd:qs(Req), Keys, BodyArgs, [keep_group_level]).
+
+parse_params(#httpd{}=Req, Keys) ->
+    parse_params(chttpd:qs(Req), Keys);
+parse_params(Props, Keys) ->
+    Args = #mrargs{},
+    parse_params(Props, Keys, Args).
+
+
+parse_params(Props, Keys, Args) ->
+    parse_params(Props, Keys, Args, []).
+
+
+parse_params([{"bookmark", Bookmark}], _Keys, #mrargs{}, _Options) ->
+    bookmark_decode(Bookmark);
+
+parse_params(Props, Keys, #mrargs{}=Args, Options) ->
+    case couch_util:get_value("bookmark", Props, nil) of
+        nil ->
+            ok;
+        _ ->
+            throw({bad_request, "Cannot use `bookmark` with other options"})
+    end,
+    couch_mrview_http:parse_params(Props, Keys, Args, Options).
+
+
+row_to_obj(Row) ->
+    Id = couch_util:get_value(id, Row),
+    row_to_obj(Id, Row).
+
+
+row_to_obj(Id, Row) ->
+    couch_mrview_http:row_to_obj(Id, Row).
+
+
+view_cb(Msg, #vacc{paginated = false}=Acc) ->
+    couch_mrview_http:view_cb(Msg, Acc);
+view_cb(Msg, #vacc{paginated = true}=Acc) ->
+    paginated_cb(Msg, Acc).
+
+
+paginated_cb({row, Row}, #vacc{buffer=Buf}=Acc) ->
+    {ok, Acc#vacc{buffer = [row_to_obj(Row) | Buf]}};
+
+paginated_cb({error, Reason}, #vacc{}=_Acc) ->
+    throw({error, Reason});
+
+paginated_cb(complete, #vacc{buffer=Buf}=Acc) ->
+    {ok, Acc#vacc{buffer=lists:reverse(Buf)}};
+
+paginated_cb({meta, Meta}, #vacc{}=VAcc) ->
+    MetaMap = lists:foldl(fun(MetaData, Acc) ->
+        case MetaData of
+            {_Key, undefined} ->
+                Acc;
+            {total, _Value} ->
+                %% We set total_rows elsewere
+                Acc;
+            {Key, Value} ->
+                maps:put(list_to_binary(atom_to_list(Key)), Value, Acc)
+        end
+    end, #{}, Meta),
+    {ok, VAcc#vacc{meta=MetaMap}}.
+
+
+paginated(Req, EtagTerm, #mrargs{page_size = PageSize} = Args, KeyFun, Fun) ->
+    Etag = couch_httpd:make_etag(EtagTerm),
+    chttpd:etag_respond(Req, Etag, fun() ->
+        hd(do_paginated(PageSize, [set_limit(Args)], KeyFun, Fun))
+    end).
+
+
+paginated(Req, EtagTerm, PageSize, QueriesArgs, KeyFun, Fun) when is_list(QueriesArgs) ->
+    Etag = couch_httpd:make_etag(EtagTerm),
+    chttpd:etag_respond(Req, Etag, fun() ->
+        Results = do_paginated(PageSize, QueriesArgs, KeyFun, Fun),
+        #{results => Results}
+    end).
+
+
+do_paginated(PageSize, QueriesArgs, KeyFun, Fun) when is_list(QueriesArgs) ->
+    {_N, Results} = lists:foldl(fun(Args0, {Limit, Acc}) ->
+        case Limit > 0 of
+            true ->
+                Args = set_limit(Args0#mrargs{page_size = Limit}),
+                {Meta, Items} = Fun(Args),
+                Result = maybe_add_bookmark(
+                    PageSize, Args, Meta, Items, KeyFun),
+                #{total_rows := Total} = Result,
+                {Limit - Total, [Result | Acc]};
+            false ->
+                Bookmark = bookmark_encode(Args0),
+                Result = #{
+                    rows => [],
+                    next => Bookmark,
+                    total_rows => 0
+                },
+                {Limit, [Result | Acc]}
+        end
+    end, {PageSize, []}, QueriesArgs),
+    lists:reverse(Results).
+
+
+maybe_add_bookmark(PageSize, Args0, Response, Items, KeyFun) ->
+    #mrargs{page_size = Limit} = Args0,
+    Args = Args0#mrargs{page_size = PageSize},
+    case check_completion(Limit, Items) of
+        {Rows, nil} ->
+            maps:merge(Response, #{
+                rows => Rows,
+                total_rows => length(Rows)
+            });
+        {Rows, Next} ->
+            NextKey = KeyFun(Next),
+            if is_binary(NextKey) -> ok; true ->
+                throw("Provided KeyFun should return binary")
+            end,
+            Bookmark = bookmark_encode(Args#mrargs{start_key=NextKey}),
+            maps:merge(Response, #{
+                rows => Rows,
+                next => Bookmark,
+                total_rows => length(Rows)
+            })
+    end.
+
+
+set_limit(#mrargs{page_size = PageSize, limit = Limit} = Args)
+        when is_integer(PageSize) andalso Limit > PageSize ->
+    Args#mrargs{limit = PageSize + 1};
+
+set_limit(#mrargs{page_size = PageSize, limit = Limit} = Args)
+        when is_integer(PageSize)  ->
+    Args#mrargs{limit = Limit + 1}.
+
+
+check_completion(Limit, Items) when length(Items) > Limit ->
+    case lists:split(Limit, Items) of
+        {Head, [NextItem | _]} ->
+            {Head, NextItem};
+        {Head, []} ->
+            {Head, nil}
+    end;
+
+check_completion(_Limit, Items) ->
+    {Items, nil}.
+
+
+bookmark_encode(Args0) ->
+    Defaults = #mrargs{},
+    {RevTerms, Mask, _} = lists:foldl(fun(Value, {Acc, Mask, Idx}) ->
+        case element(Idx, Defaults) of
+            Value ->
+                {Acc, Mask, Idx + 1};
+            _Default when Idx == #mrargs.bookmark ->
+                {Acc, Mask, Idx + 1};
+            _Default ->
+                % Its `(Idx - 1)` because the initial `1`
+                % value already accounts for one bit.
+                {[Value | Acc], (1 bsl (Idx - 1)) bor Mask, Idx + 1}
+        end
+    end, {[], 0, 1}, tuple_to_list(Args0)),
+    Terms = lists:reverse(RevTerms),
+    TermBin = term_to_binary(Terms, [compressed, {minor_version, 2}]),
+    MaskBin = binary:encode_unsigned(Mask),
+    RawBookmark = <<?BOOKMARK_VSN, MaskBin/binary, TermBin/binary>>,
+    couch_util:encodeBase64Url(RawBookmark).
+
+
+bookmark_decode(Bookmark) ->
+    try
+        RawBin = couch_util:decodeBase64Url(Bookmark),
+        <<?BOOKMARK_VSN, MaskBin:4/binary, TermBin/binary>> = RawBin,
+        Mask = binary:decode_unsigned(MaskBin),
+        Index = mask_to_index(Mask, 1, []),
+        Terms = binary_to_term(TermBin, [safe]),
+        lists:foldl(fun({Idx, Value}, Acc) ->
+            setelement(Idx, Acc, Value)
+        end, #mrargs{}, lists:zip(Index, Terms))
+    catch _:_ ->
+        throw({bad_request, <<"Invalid bookmark">>})
+    end.
+
+
+mask_to_index(0, _Pos, Acc) ->
+    lists:reverse(Acc);
+mask_to_index(Mask, Pos, Acc) when is_integer(Mask), Mask > 0 ->
+    NewAcc = case Mask band 1 of
+        0 -> Acc;
+        1 -> [Pos | Acc]
+    end,
+    mask_to_index(Mask bsr 1, Pos + 1, NewAcc).
+
+
+-ifdef(TEST).
+
+-include_lib("eunit/include/eunit.hrl").
+
+bookmark_encode_decode_test() ->
+    ?assertEqual(
+        #mrargs{page_size = 5},
+        bookmark_decode(bookmark_encode(#mrargs{page_size = 5}))
+    ),
+
+    Randomized = lists:foldl(fun(Idx, Acc) ->
+        if Idx == #mrargs.bookmark -> Acc; true ->
+            setelement(Idx, Acc, couch_uuids:random())
+        end
+    end, #mrargs{}, lists:seq(1, record_info(size, mrargs))),
+
+    ?assertEqual(
+        Randomized,
+        bookmark_decode(bookmark_encode(Randomized))
+    ).
+
+
+check_completion_test() ->
+    ?assertEqual(
+        {[], nil},
+        check_completion(1, [])
+    ),
+    ?assertEqual(
+        {[1], nil},
+        check_completion(1, [1])
+    ),
+    ?assertEqual(
+        {[1], 2},
+        check_completion(1, [1, 2])
+    ),
+    ?assertEqual(
+        {[1], 2},
+        check_completion(1, [1, 2, 3])
+    ),
+    ?assertEqual(
+        {[1, 2], nil},
+        check_completion(3, [1, 2])
+    ),
+    ?assertEqual(
+        {[1, 2, 3], nil},
+        check_completion(3, [1, 2, 3])
+    ),
+    ?assertEqual(
+        {[1, 2, 3], 4},
+        check_completion(3, [1, 2, 3, 4])
+    ),
+    ?assertEqual(
+        {[1, 2, 3], 4},
+        check_completion(3, [1, 2, 3, 4, 5])
+    ),
+    ok.
+-endif.
\ No newline at end of file
diff --git a/src/couch_views/src/couch_views_util.erl b/src/couch_views/src/couch_views_util.erl
index 395660c02..154e9e270 100644
--- a/src/couch_views/src/couch_views_util.erl
+++ b/src/couch_views/src/couch_views_util.erl
@@ -15,7 +15,9 @@
 
 -export([
     ddoc_to_mrst/2,
-    validate_args/1
+    validate_args/1,
+    validate_args/2,
+    is_paginated/1
 ]).
 
 
@@ -79,10 +81,14 @@ ddoc_to_mrst(DbName, #doc{id=Id, body={Fields}}) ->
     {ok, IdxState#mrst{sig=couch_hash:md5_hash(term_to_binary(SigInfo))}}.
 
 
+validate_args(Args) ->
+    validate_args(Args, []).
+
+
 % This is mostly a copy of couch_mrview_util:validate_args/1 but it doesn't
 % update start / end keys and also throws a not_implemented error for reduce
 %
-validate_args(#mrargs{} = Args) ->
+validate_args(#mrargs{} = Args, Opts) ->
     GroupLevel = determine_group_level(Args),
     Reduce = Args#mrargs.reduce,
 
@@ -193,6 +199,24 @@ validate_args(#mrargs{} = Args) ->
         _ -> mrverror(<<"Invalid value for `sorted`.">>)
     end,
 
+    MaxPageSize = couch_util:get_value(page_size, Opts, 0),
+    case {Args#mrargs.page_size, MaxPageSize} of
+        {_, 0} -> ok;
+        {Value, _} -> validate_limit(<<"page_size">>, Value, 1, MaxPageSize)
+    end,
+
+    case {Args#mrargs.skip, MaxPageSize} of
+        {_, 0} -> ok;
+        {Skip, _} -> validate_limit(<<"skip">>, Skip, 0, MaxPageSize)
+    end,
+
+    case {is_list(Args#mrargs.keys), is_integer(Args#mrargs.page_size)} of
+        {true, true} ->
+            mrverror(<<"`page_size` is incompatible with `keys`">>);
+        _ ->
+            ok
+    end,
+
     case {Reduce, Args#mrargs.view_type} of
         {false, _} -> ok;
         {_, red} -> throw(not_implemented);
@@ -201,6 +225,31 @@ validate_args(#mrargs{} = Args) ->
 
     Args#mrargs{group_level=GroupLevel}.
 
+validate_limit(Name, Value, _Min, _Max) when not is_integer(Value) ->
+    mrverror(<<"`", Name/binary, "` should be an integer">>);
+
+validate_limit(Name, Value, Min, Max) when Value > Max ->
+    range_error_msg(Name, Min, Max);
+
+validate_limit(Name, Value, Min, Max) when Value < Min ->
+    range_error_msg(Name, Min, Max);
+
+validate_limit(_Name, _Value, _Min, _Max) ->
+    ok.
+
+range_error_msg(Name, Min, Max) ->
+    MinBin = list_to_binary(integer_to_list(Min)),
+    MaxBin = list_to_binary(integer_to_list(Max)),
+    mrverror(<<
+        "`",
+        Name/binary,
+        "` should be an integer in range [",
+        MinBin/binary,
+        " .. ",
+        MaxBin/binary,
+        "]"
+    >>).
+
 
 determine_group_level(#mrargs{group=undefined, group_level=undefined}) ->
     0;
@@ -220,3 +269,10 @@ determine_group_level(#mrargs{group_level=GroupLevel}) ->
 
 mrverror(Mesg) ->
     throw({query_parse_error, Mesg}).
+
+
+is_paginated(#mrargs{page_size = PageSize}) when is_integer(PageSize) ->
+    true;
+
+is_paginated(_) ->
+    false.