1% Licensed under the Apache License, Version 2.0 (the "License"); you may not 2% use this file except in compliance with the License. You may obtain a copy of 3% the License at 4% 5% http://www.apache.org/licenses/LICENSE-2.0 6% 7% Unless required by applicable law or agreed to in writing, software 8% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 9% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 10% License for the specific language governing permissions and limitations under 11% the License. 12 13-module(couch_partition). 14 15 16-export([ 17 extract/1, 18 from_docid/1, 19 is_member/2, 20 21 start_key/1, 22 end_key/1, 23 shard_key/1, 24 25 validate_dbname/2, 26 validate_docid/1, 27 validate_partition/1, 28 29 hash/1 30]). 31 32 33-include_lib("couch/include/couch_db.hrl"). 34 35 36extract(Value) when is_binary(Value) -> 37 case binary:split(Value, <<":">>) of 38 [Partition, Rest] -> 39 {Partition, Rest}; 40 _ -> 41 undefined 42 end; 43 44extract(_) -> 45 undefined. 46 47 48from_docid(DocId) -> 49 case extract(DocId) of 50 undefined -> 51 throw({illegal_docid, <<"Doc id must be of form partition:id">>}); 52 {Partition, _} -> 53 Partition 54 end. 55 56 57is_member(DocId, Partition) -> 58 case extract(DocId) of 59 {Partition, _} -> 60 true; 61 _ -> 62 false 63 end. 64 65 66start_key(Partition) -> 67 <<Partition/binary, ":">>. 68 69 70end_key(Partition) -> 71 <<Partition/binary, ";">>. 72 73 74shard_key(Partition) -> 75 <<Partition/binary, ":foo">>. 76 77 78validate_dbname(DbName, Options) when is_list(DbName) -> 79 validate_dbname(?l2b(DbName), Options); 80validate_dbname(DbName, Options) when is_binary(DbName) -> 81 Props = couch_util:get_value(props, Options, []), 82 IsPartitioned = couch_util:get_value(partitioned, Props, false), 83 84 if not IsPartitioned -> ok; true -> 85 86 DbsDbName = config:get("mem3", "shards_db", "_dbs"), 87 NodesDbName = config:get("mem3", "nodes_db", "_nodes"), 88 UsersDbSuffix = config:get("couchdb", "users_db_suffix", "_users"), 89 Suffix = couch_db:dbname_suffix(DbName), 90 91 SysDbNames = [ 92 iolist_to_binary(DbsDbName), 93 iolist_to_binary(NodesDbName) 94 | ?SYSTEM_DATABASES 95 ], 96 97 Suffices = [ 98 <<"_replicator">>, 99 <<"_users">>, 100 iolist_to_binary(UsersDbSuffix) 101 ], 102 103 IsSysDb = lists:member(DbName, SysDbNames) 104 orelse lists:member(Suffix, Suffices), 105 106 if not IsSysDb -> ok; true -> 107 throw({bad_request, <<"Cannot partition a system database">>}) 108 end 109 end. 110 111 112validate_docid(<<"_design/", _/binary>>) -> 113 ok; 114validate_docid(<<"_local/", _/binary>>) -> 115 ok; 116validate_docid(DocId) when is_binary(DocId) -> 117 % When this function is called we already know that 118 % DocId is already valid thus we only need to 119 % ensure that the partition exists and is not empty. 120 case extract(DocId) of 121 undefined -> 122 throw({illegal_docid, <<"Doc id must be of form partition:id">>}); 123 {Partition, PartitionedDocId} -> 124 validate_partition(Partition), 125 couch_doc:validate_docid(PartitionedDocId) 126 end. 127 128 129validate_partition(<<>>) -> 130 throw({illegal_partition, <<"Partition must not be empty">>}); 131validate_partition(Partition) when is_binary(Partition) -> 132 case Partition of 133 <<"_", _/binary>> -> 134 Msg1 = <<"Partition must not start with an underscore">>, 135 throw({illegal_partition, Msg1}); 136 _ -> 137 ok 138 end, 139 case couch_util:validate_utf8(Partition) of 140 true -> 141 ok; 142 false -> 143 Msg2 = <<"Partition must be valid UTF-8">>, 144 throw({illegal_partition, Msg2}) 145 end, 146 case extract(Partition) of 147 {_, _} -> 148 Msg3 = <<"Partition must not contain a colon">>, 149 throw({illegal_partition, Msg3}); 150 undefined -> 151 ok 152 end; 153validate_partition(_) -> 154 throw({illegal_partition, <<"Partition must be a string">>}). 155 156 157% Document ids that start with an underscore 158% (i.e., _local and _design) do not contain a 159% partition and thus do not use the partition 160% hashing. 161hash(<<"_", _/binary>> = DocId) -> 162 erlang:crc32(DocId); 163hash(DocId) when is_binary(DocId) -> 164 erlang:crc32(from_docid(DocId)). 165