This is an automated email from the ASF dual-hosted git repository.
nickva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/couchdb.git
The following commit(s) were added to refs/heads/main by this push:
new 25b1b9a22 Improve document grouping in fabric_doc_update
25b1b9a22 is described below
commit 25b1b9a22d9bc6a98c1f011cd20de3f3bfeec40b
Author: Nick Vatamaniuc <[email protected]>
AuthorDate: Wed Jun 10 17:29:07 2026 -0400
Improve document grouping in fabric_doc_update
The idea of doc/shard grouping and rotation was that different databases
would
start their workers on different copies, spread those uniformly and
consistently by database and range.
However, the order of entries were actually dependent on `dict:to_list/1`
internals. So, if `dict` was changed to return sorted items, like for
instance
maps do, we'd end up suddenly breaking the supposedly randomized spread (I
found this out by actually trying to use a map there instead of a dict).
To fix this, make the ordering explicit. Group the entries first, then order
each range's group by the unified `mem3:owners/3` membership order, so the
first copy of each range lands on the owner node, consistent with "Unify
membership hashes", and the result no longer depends on dict internals at
all.
While at it make it so we don't have to duplicate the main rotate function
in
tests and use the actual function we intend to test.
---
src/fabric/src/fabric.erl | 2 +
src/fabric/src/fabric_doc_update.erl | 143 ++++++++++++++++++++++++++---------
src/mem3/src/mem3.erl | 34 ++++++++-
3 files changed, 142 insertions(+), 37 deletions(-)
diff --git a/src/fabric/src/fabric.erl b/src/fabric/src/fabric.erl
index 1b8434b03..965f59c33 100644
--- a/src/fabric/src/fabric.erl
+++ b/src/fabric/src/fabric.erl
@@ -833,7 +833,9 @@ setup() ->
meta = []
},
ok = application:ensure_started(config),
+ meck:expect(mem3, shards, fun(_) -> [] end),
meck:expect(mem3, shards, fun(_, _) -> [] end),
+ meck:expect(mem3, shards, fun(_, _, _) -> [] end),
meck:expect(mem3, quorum, fun(_) -> 1 end),
meck:expect(rexi, cast, fun(_, _) -> ok end),
meck:expect(rexi_utils, recv, fun(_, _, _, _, _, _) -> {ok, {error, [{Doc,
conflict}]}} end),
diff --git a/src/fabric/src/fabric_doc_update.erl
b/src/fabric/src/fabric_doc_update.erl
index f0481b1fc..203f9c89d 100644
--- a/src/fabric/src/fabric_doc_update.erl
+++ b/src/fabric/src/fabric_doc_update.erl
@@ -339,32 +339,39 @@ good_reply(_, _) ->
-spec group_docs_by_shard(binary(), [#doc{}]) -> [{#shard{}, [#doc{}]}].
group_docs_by_shard(DbName, Docs) ->
+ Shards = mem3:shards(DbName),
+ Grouped = group_docs([{Doc, mem3:shards(DbName, Id, Shards)} || #doc{id =
Id} = Doc <- Docs]),
+ rotate_ranges(DbName, Grouped).
+
+group_docs(DocsShards) ->
dict:to_list(
lists:foldl(
- fun(#doc{id = Id} = Doc, D0) ->
+ fun({Doc, Shards}, D0) ->
lists:foldl(
fun(Shard, D1) ->
dict:append(Shard, Doc, D1)
end,
D0,
- begin
- case mem3:shards(DbName, Id) of
- [] ->
- [];
- [#shard{range = Range} | _] = Shards ->
- owner_order(DbName, Range, Shards)
- end
- end
+ Shards
)
end,
dict:new(),
- Docs
+ DocsShards
)
).
-owner_order(DbName, Range, Shards) ->
- Owners = mem3:owners(DbName, Range, [N || #shard{node = N} <- Shards]),
- [S || N <- Owners, #shard{node = N1} = S <- Shards, N1 =:= N].
+% Deterministically order each range's workers by ownership. When serialized,
+% workers start with the first entry of each range.
+rotate_ranges(DbName, Grouped) ->
+ FoldF = fun({#shard{range = R}, _} = E, D) -> orddict:append(R, E, D) end,
+ ByRange = orddict:to_list(lists:foldl(FoldF, orddict:new(), Grouped)),
+ lists:append([owner_order(DbName, R, G) || {R, G} <- ByRange]).
+
+% Order {Shards, Docs} kvs by the membership in mem3:owners/3 It's
+% important that we use the same ownership order as the replicator and peruser.
+owner_order(DbName, Range, Entries) ->
+ Owners = mem3:owners(DbName, Range, [N || {#shard{node = N}, _} <-
Entries]),
+ [E || N <- Owners, {#shard{node = N1}, _} = E <- Entries, N1 =:= N].
append_update_replies([], [], DocReplyDict) ->
DocReplyDict;
@@ -513,7 +520,11 @@ doc_update_test_() ->
fun parallel_in_flight_after_conflict/0,
fun serial_filters_conflicts_at_cast/0,
fun sws_false_mode_conflict_not_final/0,
- fun sws_false_mode_ok_can_outvote_conflict/0
+ fun sws_false_mode_ok_can_outvote_conflict/0,
+ fun group_docs_content_and_order/0,
+ fun rotate_ranges_rotates_each_range/0,
+ fun rotate_ranges_varies_by_db/0,
+ fun rotate_ranges_preserves_entries/0
]
}.
@@ -1149,33 +1160,93 @@ sws_false_mode_ok_can_outvote_conflict() ->
).
owner_order_test() ->
- S1 = #shard{name = <<"r1">>, node = n1, range = [0, 10]},
- S2 = #shard{name = <<"r1">>, node = n2, range = [0, 10]},
- S3 = #shard{name = <<"r1">>, node = n3, range = [0, 10]},
+ S1 = mk_shard(<<"r1">>, n1, [0, 10]),
+ S2 = mk_shard(<<"r1">>, n2, [0, 10]),
+ S3 = mk_shard(<<"r1">>, n3, [0, 10]),
% The rotation amounts, erlang:phash2({Db, Range}) rem 3, are 0 for
% {<<"dba">>, [0, 10]} and 2 for {<<"dbe">>, [0, 10]}
- ?assertEqual([S1, S2, S3], owner_order(<<"dba">>, [0, 10], [S2, S3, S1])),
- ?assertEqual([S3, S1, S2], owner_order(<<"dbe">>, [0, 10], [S2, S3, S1])),
- % the first copy is on the owner node
+ Entries = [{S2, d2}, {S3, d3}, {S1, d1}],
+ ?assertEqual([{S1, d1}, {S2, d2}, {S3, d3}], rotate_ranges(<<"dba">>,
Entries)),
+ ?assertEqual([{S3, d3}, {S1, d1}, {S2, d2}], rotate_ranges(<<"dbe">>,
Entries)),
+ % the first copy of each range is on the owner node
?assertEqual(n1, hd(mem3:owners(<<"dba">>, [0, 10], [n1, n2, n3]))),
?assertEqual(n3, hd(mem3:owners(<<"dbe">>, [0, 10], [n1, n2, n3]))).
-% needed for testing to avoid having to start the mem3 application
-group_docs_by_shard_hack(_DbName, Shards, Docs) ->
- dict:to_list(
- lists:foldl(
- fun(#doc{id = _Id} = Doc, D0) ->
- lists:foldl(
- fun(Shard, D1) ->
- dict:append(Shard, Doc, D1)
- end,
- D0,
- Shards
- )
- end,
- dict:new(),
- Docs
- )
+group_docs_content_and_order() ->
+ S1 = mk_shard(<<"r1">>, n1, [0, 10]),
+ S2 = mk_shard(<<"r1">>, n2, [0, 10]),
+ DocA = #doc{id = <<"a">>},
+ DocB = #doc{id = <<"b">>},
+ DocC = #doc{id = <<"c">>},
+ Grouped = group_docs([
+ {DocA, [S1, S2]},
+ {DocB, [S1]},
+ {DocC, [S2, S1]}
+ ]),
+ ?assertEqual(2, length(Grouped)),
+ % Every shard gets exactly its docs in original batch order
+ ?assertEqual([DocA, DocB, DocC], couch_util:get_value(S1, Grouped)),
+ ?assertEqual([DocA, DocC], couch_util:get_value(S2, Grouped)).
+
+rotate_ranges_rotates_each_range() ->
+ {[S11, S12, S13, S21, S22, S23], Entries} = rotate_fixture(),
+ % each range's group is sorted, rotated by the {DbName, Range} membership
+ % hash (both amounts are 2 for <<"dbe">>), and its first entry is the copy
+ % start_workers/1 will start first
+ ?assertEqual(
+ [
+ {S13, docs13},
+ {S11, docs11},
+ {S12, docs12},
+ {S23, docs23},
+ {S21, docs21},
+ {S22, docs22}
+ ],
+ rotate_ranges(<<"dbe">>, Entries)
).
+rotate_ranges_varies_by_db() ->
+ {[S11, _, S13, S21, S22, S23], Entries} = rotate_fixture(),
+ HeadsFor = fun(DbName) ->
+ Rotated = rotate_ranges(DbName, Entries),
+ Ranges = lists:usort([R || {#shard{range = R}, _} <- Rotated]),
+ [hd([S || {#shard{range = R1} = S, _} <- Rotated, R1 =:= R]) || R <-
Ranges]
+ end,
+ % different db names => different first copies (phash2({Db, Range}) rem 3
+ % amounts: dba {0, 0}, dbb {0, 1}, dbe {2, 2})
+ ?assertEqual([S11, S21], HeadsFor(<<"dba">>)),
+ ?assertEqual([S11, S22], HeadsFor(<<"dbb">>)),
+ ?assertEqual([S13, S23], HeadsFor(<<"dbe">>)),
+ ?assertNotEqual(HeadsFor(<<"dba">>), HeadsFor(<<"dbe">>)).
+
+rotate_ranges_preserves_entries() ->
+ {_, Entries} = rotate_fixture(),
+ ?assertEqual(lists:sort(Entries), lists:sort(rotate_ranges(<<"dba">>,
Entries))),
+ ?assertEqual(lists:sort(Entries), lists:sort(rotate_ranges(<<"dbb">>,
Entries))).
+
+rotate_fixture() ->
+ S11 = mk_shard(<<"r1">>, n1, [0, 10]),
+ S12 = mk_shard(<<"r1">>, n2, [0, 10]),
+ S13 = mk_shard(<<"r1">>, n3, [0, 10]),
+ S21 = mk_shard(<<"r2">>, n1, [11, 20]),
+ S22 = mk_shard(<<"r2">>, n2, [11, 20]),
+ S23 = mk_shard(<<"r2">>, n3, [11, 20]),
+ % In a jumbled order to see how rotate_ranges will sort it (or not)
+ Entries = [
+ {S22, docs22},
+ {S11, docs11},
+ {S13, docs13},
+ {S21, docs21},
+ {S12, docs12},
+ {S23, docs23}
+ ],
+ {[S11, S12, S13, S21, S22, S23], Entries}.
+
+mk_shard(Name, Node, Range) ->
+ #shard{name = Name, node = Node, range = Range}.
+
+% needed for testing to avoid having to start the mem3 application
+group_docs_by_shard_hack(DbName, Shards, Docs) ->
+ rotate_ranges(DbName, group_docs([{Doc, Shards} || Doc <- Docs])).
+
-endif.
diff --git a/src/mem3/src/mem3.erl b/src/mem3/src/mem3.erl
index b16240665..2b20d546a 100644
--- a/src/mem3/src/mem3.erl
+++ b/src/mem3/src/mem3.erl
@@ -19,7 +19,7 @@
nodes/0,
node_info/2,
props/1,
- shards/1, shards/2,
+ shards/1, shards/2, shards/3,
choose_shards/2,
n/1, n/2,
dbname/1,
@@ -170,6 +170,11 @@ shards_int(DbName, DocId, Options) when is_list(DocId) ->
shards_int(DbName, DocId, Options) ->
mem3_shards:for_docid(DbName, DocId, Options).
+-spec shards(DbName :: binary(), DocId :: binary(), Shards :: [#shard{}]) ->
[#shard{}].
+shards(DbName, DocId, Shards) when is_binary(DbName), is_binary(DocId) ->
+ HashKey = mem3_hash:calculate(DbName, DocId),
+ [S || #shard{range = [B, E]} = S <- Shards, B =< HashKey, HashKey =< E].
+
-spec ushards(DbName :: iodata()) -> [#shard{}].
ushards(DbName) ->
Nodes = [node() | erlang:nodes()],
@@ -637,6 +642,33 @@ live_owner_test() ->
% no live copy nodes
?assertError(badarg, live_owner(Owners, [])).
+shards_3_test_() ->
+ {
+ setup,
+ fun() ->
+ meck:new(mem3_hash, [passthrough]),
+ ok = meck:expect(mem3_hash, calculate, fun(_Db, DocId) ->
+ case DocId of
+ <<"low">> -> 5;
+ <<"hi">> -> 15;
+ <<"edge">> -> 10
+ end
+ end)
+ end,
+ fun(_) -> meck:unload() end,
+ fun() ->
+ S1a = #shard{node = n1, range = [0, 10]},
+ S1b = #shard{node = n2, range = [0, 10]},
+ S2a = #shard{node = n1, range = [11, 20]},
+ S2b = #shard{node = n2, range = [11, 20]},
+ Shards = [S1a, S2a, S1b, S2b],
+ ?assertEqual([S1a, S1b], shards(<<"db">>, <<"low">>, Shards)),
+ ?assertEqual([S2a, S2b], shards(<<"db">>, <<"hi">>, Shards)),
+ ?assertEqual([S1a, S1b], shards(<<"db">>, <<"edge">>, Shards)),
+ ?assertEqual([], shards(<<"db">>, <<"low">>, []))
+ end
+ }.
+
rotate_rand_degenerate_test() ->
?assertEqual([1], rotate_rand([1])).
