Implement partitioned dbs

This change introduces the ability for users to place a group of
documents in a single shard range by specifying a "partition key" in the
document id. A partition key is denoted by everything preceding a colon
':' in the document id.

Every document id (except for design documents) in a partitioned
database is required to have a partition key.

Co-authored-by: Garren Smith <garren.smith@gmail.com>
Co-authored-by: Robert Newson <rnewson@apache.org>

5 files changed, 205 insertions, 25 deletions

diff --git a/src/couch/src/couch_db.erl b/src/couch/src/couch_db.erl
index 8ff73e4d2..2c6f41bf7 100644
--- a/src/couch/src/couch_db.erl
+++ b/src/couch/src/couch_db.erl
@@ -57,6 +57,7 @@
     is_system_db/1,
     is_clustered/1,
     is_system_db_name/1,
+    is_partitioned/1,
 
     set_revs_limit/2,
     set_purge_infos_limit/2,
@@ -85,6 +86,9 @@
     get_minimum_purge_seq/1,
     purge_client_exists/3,
 
+    validate_docid/2,
+    doc_from_json_obj_validate/2,
+
     update_doc/3,
     update_doc/4,
     update_docs/4,
@@ -217,6 +221,10 @@ is_clustered(#db{}) ->
 is_clustered(?OLD_DB_REC = Db) ->
     ?OLD_DB_MAIN_PID(Db) == undefined.
 
+is_partitioned(#db{options = Options}) ->
+    Props = couch_util:get_value(props, Options, []),
+    couch_util:get_value(partitioned, Props, false).
+
 ensure_full_commit(#db{main_pid=Pid, instance_start_time=StartTime}) ->
     ok = gen_server:call(Pid, full_commit, infinity),
     {ok, StartTime}.
@@ -798,6 +806,30 @@ name(#db{name=Name}) ->
 name(?OLD_DB_REC = Db) ->
     ?OLD_DB_NAME(Db).
 
+
+validate_docid(#db{} = Db, DocId) when is_binary(DocId) ->
+    couch_doc:validate_docid(DocId, name(Db)),
+    case is_partitioned(Db) of
+        true ->
+            couch_partition:validate_docid(DocId);
+        false ->
+            ok
+    end.
+
+
+doc_from_json_obj_validate(#db{} = Db, DocJson) ->
+    Doc = couch_doc:from_json_obj_validate(DocJson, name(Db)),
+    {Props} = DocJson,
+    case couch_util:get_value(<<"_id">>, Props) of
+        DocId when is_binary(DocId) ->
+            % Only validate the docid if it was provided
+            validate_docid(Db, DocId);
+        _ ->
+            ok
+    end,
+    Doc.
+
+
 update_doc(Db, Doc, Options) ->
     update_doc(Db, Doc, Options, interactive_edit).
 
diff --git a/src/couch/src/couch_httpd.erl b/src/couch/src/couch_httpd.erl
index 861fd58c4..3cdfc0ca3 100644
--- a/src/couch/src/couch_httpd.erl
+++ b/src/couch/src/couch_httpd.erl
@@ -878,6 +878,8 @@ error_info(md5_mismatch) ->
     {400, <<"content_md5_mismatch">>, <<"Possible message corruption.">>};
 error_info({illegal_docid, Reason}) ->
     {400, <<"illegal_docid">>, Reason};
+error_info({illegal_partition, Reason}) ->
+    {400, <<"illegal_partition">>, Reason};
 error_info(not_found) ->
     {404, <<"not_found">>, <<"missing">>};
 error_info({not_found, Reason}) ->
diff --git a/src/couch/src/couch_httpd_db.erl b/src/couch/src/couch_httpd_db.erl
index ced146e39..6cfae9610 100644
--- a/src/couch/src/couch_httpd_db.erl
+++ b/src/couch/src/couch_httpd_db.erl
@@ -266,8 +266,7 @@ db_req(#httpd{method='GET',path_parts=[_DbName]}=Req, Db) ->
 
 db_req(#httpd{method='POST',path_parts=[_DbName]}=Req, Db) ->
     couch_httpd:validate_ctype(Req, "application/json"),
-    DbName = couch_db:name(Db),
-    Doc = couch_doc:from_json_obj_validate(couch_httpd:json_body(Req), DbName),
+    Doc = couch_db:doc_from_json_obj_validate(Db, couch_httpd:json_body(Req)),
     validate_attachment_names(Doc),
     Doc2 = case Doc#doc.id of
         <<"">> ->
@@ -313,7 +312,6 @@ db_req(#httpd{method='POST',path_parts=[_,<<"_bulk_docs">>]}=Req, Db) ->
     couch_stats:increment_counter([couchdb, httpd, bulk_requests]),
     couch_httpd:validate_ctype(Req, "application/json"),
     {JsonProps} = couch_httpd:json_body_obj(Req),
-    DbName = couch_db:name(Db),
     case couch_util:get_value(<<"docs">>, JsonProps) of
     undefined ->
         send_error(Req, 400, <<"bad_request">>, <<"Missing JSON list of 'docs'">>);
@@ -331,7 +329,7 @@ db_req(#httpd{method='POST',path_parts=[_,<<"_bulk_docs">>]}=Req, Db) ->
         true ->
             Docs = lists:map(
                 fun({ObjProps} = JsonObj) ->
-                    Doc = couch_doc:from_json_obj_validate(JsonObj, DbName),
+                    Doc = couch_db:doc_from_json_obj_validate(Db, JsonObj),
                     validate_attachment_names(Doc),
                     Id = case Doc#doc.id of
                         <<>> -> couch_uuids:new();
@@ -365,7 +363,7 @@ db_req(#httpd{method='POST',path_parts=[_,<<"_bulk_docs">>]}=Req, Db) ->
             end;
         false ->
             Docs = lists:map(fun(JsonObj) ->
-                    Doc = couch_doc:from_json_obj_validate(JsonObj, DbName),
+                    Doc = couch_db:doc_from_json_obj_validate(Db, JsonObj),
                     validate_attachment_names(Doc),
                     Doc
                 end, DocsArray),
@@ -502,17 +500,15 @@ db_req(#httpd{path_parts=[_, DocId | FileNameParts]}=Req, Db) ->
 db_doc_req(#httpd{method='DELETE'}=Req, Db, DocId) ->
     % check for the existence of the doc to handle the 404 case.
     couch_doc_open(Db, DocId, nil, []),
-    DbName = couch_db:name(Db),
     case couch_httpd:qs_value(Req, "rev") of
     undefined ->
-        update_doc(Req, Db, DocId,
-                couch_doc_from_req(Req, DocId, {[{<<"_deleted">>,true}]},
-                    DbName));
+        JsonObj = {[{<<"_deleted">>,true}]},
+        Doc = couch_doc_from_req(Req, Db, DocId, JsonObj),
+        update_doc(Req, Db, DocId, Doc);
     Rev ->
-        update_doc(Req, Db, DocId,
-                couch_doc_from_req(Req, DocId,
-                    {[{<<"_rev">>, ?l2b(Rev)},{<<"_deleted">>,true}]},
-                    DbName))
+        JsonObj = {[{<<"_rev">>, ?l2b(Rev)},{<<"_deleted">>,true}]},
+        Doc = couch_doc_from_req(Req, Db, DocId, JsonObj),
+        update_doc(Req, Db, DocId, Doc)
     end;
 
 db_doc_req(#httpd{method = 'GET', mochi_req = MochiReq} = Req, Db, DocId) ->
@@ -565,8 +561,7 @@ db_doc_req(#httpd{method = 'GET', mochi_req = MochiReq} = Req, Db, DocId) ->
 
 db_doc_req(#httpd{method='POST'}=Req, Db, DocId) ->
     couch_httpd:validate_referer(Req),
-    DbName = couch_db:name(Db),
-    couch_doc:validate_docid(DocId, DbName),
+    couch_db:validate_docid(Db, DocId),
     couch_httpd:validate_ctype(Req, "multipart/form-data"),
     Form = couch_httpd:parse_form(Req),
     case couch_util:get_value("_doc", Form) of
@@ -574,7 +569,7 @@ db_doc_req(#httpd{method='POST'}=Req, Db, DocId) ->
         Rev = couch_doc:parse_rev(couch_util:get_value("_rev", Form)),
         {ok, [{ok, Doc}]} = couch_db:open_doc_revs(Db, DocId, [Rev], []);
     Json ->
-        Doc = couch_doc_from_req(Req, DocId, ?JSON_DECODE(Json), DbName)
+        Doc = couch_doc_from_req(Req, Db, DocId, ?JSON_DECODE(Json))
     end,
     UpdatedAtts = [
         couch_att:new([
@@ -600,15 +595,14 @@ db_doc_req(#httpd{method='POST'}=Req, Db, DocId) ->
     update_doc(Req, Db, DocId, NewDoc);
 
 db_doc_req(#httpd{method='PUT'}=Req, Db, DocId) ->
-    DbName = couch_db:name(Db),
-    couch_doc:validate_docid(DocId, DbName),
+    couch_db:validate_docid(Db, DocId),
 
     case couch_util:to_list(couch_httpd:header_value(Req, "Content-Type")) of
     ("multipart/related;" ++ _) = ContentType ->
         couch_httpd:check_max_request_length(Req),
         {ok, Doc0, WaitFun, Parser} = couch_doc:doc_from_multi_part_stream(
             ContentType, fun() -> receive_request_data(Req) end),
-        Doc = couch_doc_from_req(Req, DocId, Doc0, DbName),
+        Doc = couch_doc_from_req(Req, Db, DocId, Doc0),
         try
             Result = update_doc(Req, Db, DocId, Doc),
             WaitFun(),
@@ -620,7 +614,7 @@ db_doc_req(#httpd{method='PUT'}=Req, Db, DocId) ->
         end;
     _Else ->
         Body = couch_httpd:json_body(Req),
-        Doc = couch_doc_from_req(Req, DocId, Body, DbName),
+        Doc = couch_doc_from_req(Req, Db, DocId, Body),
         update_doc(Req, Db, DocId, Doc)
     end;
 
@@ -805,7 +799,7 @@ update_doc(Req, Db, DocId, #doc{deleted=Deleted}=Doc, Headers, UpdateType) ->
                 {rev, NewRevStr}]})
     end.
 
-couch_doc_from_req(Req, DocId, #doc{revs=Revs}=Doc, _) ->
+couch_doc_from_req(Req, _Db, DocId, #doc{revs=Revs}=Doc) ->
     validate_attachment_names(Doc),
     Rev = case couch_httpd:qs_value(Req, "rev") of
     undefined ->
@@ -832,9 +826,9 @@ couch_doc_from_req(Req, DocId, #doc{revs=Revs}=Doc, _) ->
         end
     end,
     Doc#doc{id=DocId, revs=Revs2};
-couch_doc_from_req(Req, DocId, Json, DbName) ->
-    couch_doc_from_req(Req, DocId,
-        couch_doc:from_json_obj_validate(Json, DbName), DbName).
+couch_doc_from_req(Req, Db, DocId, Json) ->
+    Doc = couch_db:doc_from_json_obj_validate(Db, Json),
+    couch_doc_from_req(Req, Db, DocId, Doc).
 
 % Useful for debugging
 % couch_doc_open(Db, DocId) ->
@@ -1042,7 +1036,7 @@ db_attachment_req(#httpd{method=Method,mochi_req=MochiReq}=Req, Db, DocId, FileN
                 % check for the existence of the doc to handle the 404 case.
                 couch_doc_open(Db, DocId, nil, [])
             end,
-            couch_doc:validate_docid(DocId, couch_db:name(Db)),
+            couch_db:validate_docid(Db, DocId),
             #doc{id=DocId};
         Rev ->
             case couch_db:open_doc_revs(Db, DocId, [Rev], []) of
diff --git a/src/couch/src/couch_partition.erl b/src/couch/src/couch_partition.erl
new file mode 100644
index 000000000..783921f0a
--- /dev/null
+++ b/src/couch/src/couch_partition.erl
@@ -0,0 +1,148 @@
+% 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_partition).
+
+
+-export([
+    extract/1,
+    from_docid/1,
+    is_member/2,
+
+    validate_dbname/2,
+    validate_docid/1,
+    validate_partition/1,
+
+    hash/1
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+
+
+extract(Value) when is_binary(Value) ->
+    case binary:split(Value, <<":">>) of
+        [Partition, Rest] ->
+            {Partition, Rest};
+        _ ->
+            undefined
+    end;
+
+extract(_) ->
+    undefined.
+
+
+from_docid(DocId) ->
+    case extract(DocId) of
+        undefined ->
+            throw({illegal_docid, <<"Doc id must be of form partition:id">>});
+        {Partition, _} ->
+            Partition
+    end.
+
+
+is_member(DocId, Partition) ->
+    case extract(DocId) of
+        {Partition, _} ->
+            true;
+        _ ->
+            false
+    end.
+
+
+validate_dbname(DbName, Options) when is_list(DbName) ->
+    validate_dbname(?l2b(DbName), Options);
+validate_dbname(DbName, Options) when is_binary(DbName) ->
+    Props = couch_util:get_value(props, Options, []),
+    IsPartitioned = couch_util:get_value(partitioned, Props, false),
+
+    if not IsPartitioned -> ok; true ->
+
+        DbsDbName = config:get("mem3", "shards_db", "_dbs"),
+        NodesDbName = config:get("mem3", "nodes_db", "_nodes"),
+        UsersDbSuffix = config:get("couchdb", "users_db_suffix", "_users"),
+        Suffix = couch_db:dbname_suffix(DbName),
+
+        SysDbNames = [
+                iolist_to_binary(DbsDbName),
+                iolist_to_binary(NodesDbName)
+                | ?SYSTEM_DATABASES
+            ],
+
+        Suffices = [
+                <<"_replicator">>,
+                <<"_users">>,
+                iolist_to_binary(UsersDbSuffix)
+            ],
+
+        IsSysDb = lists:member(DbName, SysDbNames)
+                orelse lists:member(Suffix, Suffices),
+
+        if not IsSysDb -> ok; true ->
+            throw({bad_request, <<"Cannot partition a system database">>})
+        end
+    end.
+
+
+validate_docid(<<"_design/", _/binary>>) ->
+    ok;
+validate_docid(<<"_local/", _/binary>>) ->
+    ok;
+validate_docid(DocId) when is_binary(DocId) ->
+    % When this function is called we already know that
+    % DocId is already valid thus we only need to
+    % ensure that the partition exists and is not empty.
+    case extract(DocId) of
+        undefined ->
+            throw({illegal_docid, <<"Doc id must be of form partition:id">>});
+        {Partition, PartitionedDocId} ->
+            validate_partition(Partition),
+            couch_doc:validate_docid(PartitionedDocId)
+    end.
+
+
+validate_partition(<<>>) ->
+    throw({illegal_partition, <<"Partition must not be empty">>});
+validate_partition(Partition) when is_binary(Partition) ->
+    case Partition of
+        <<"_", _/binary>> ->
+            Msg1 = <<"Partition must not start with an underscore">>,
+            throw({illegal_partition, Msg1});
+        _ ->
+            ok
+    end,
+    case couch_util:validate_utf8(Partition) of
+        true ->
+            ok;
+        false ->
+            Msg2 = <<"Partition must be valid UTF-8">>,
+            throw({illegal_partition, Msg2})
+    end,
+    case extract(Partition) of
+        {_, _} ->
+            Msg3 = <<"Partition must not contain a colon">>,
+            throw({illegal_partition, Msg3});
+        undefined ->
+            ok
+    end;
+validate_partition(_) ->
+    throw({illegal_partition, <<"Partition must be a string">>}).
+
+
+% Document ids that start with an underscore
+% (i.e., _local and _design) do not contain a
+% partition and thus do not use the partition
+% hashing.
+hash(<<"_", _/binary>> = DocId) ->
+    erlang:crc32(DocId);
+hash(DocId) when is_binary(DocId) ->
+    erlang:crc32(from_docid(DocId)).
diff --git a/src/couch/src/couch_server.erl b/src/couch/src/couch_server.erl
index df447d1c7..395ec31a9 100644
--- a/src/couch/src/couch_server.erl
+++ b/src/couch/src/couch_server.erl
@@ -116,6 +116,7 @@ close_lru() ->
 
 create(DbName, Options0) ->
     Options = maybe_add_sys_db_callbacks(DbName, Options0),
+    couch_partition:validate_dbname(DbName, Options),
     case gen_server:call(couch_server, {create, DbName, Options}, infinity) of
     {ok, Db0} ->
         Ctx = couch_util:get_value(user_ctx, Options, #user_ctx{}),
@@ -221,6 +222,9 @@ init([]) ->
     % Mark pluggable storage engines as a supported feature
     config:enable_feature('pluggable-storage-engines'),
 
+    % Mark partitioned databases as a supported feature
+    config:enable_feature(partitions),
+
     % read config and register for configuration changes
 
     % just stop if one of the config settings change. couch_server_sup