yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(ft-asm): follow proper `segment` fragment type

This commit is contained in:
Andrew Mayorov 2023-02-02 21:30:30 +03:00 committed by Ilya Averyanov
parent 1fedae8a16
commit 14b2a1013b
1 changed files with 40 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

80
apps/emqx_ft/src/emqx_ft_assembly.erl
View File

@ -47,7 +47,7 @@ append(Asm, Node, Fragments) when is_list(Fragments) ->
lists:foldl(fun(F, AsmIn) -> append(AsmIn, Node, F) end, Asm, Fragments); lists:foldl(fun(F, AsmIn) -> append(AsmIn, Node, F) end, Asm, Fragments);
append(Asm, Node, Fragment = #{fragment := {filemeta, _}}) -> append(Asm, Node, Fragment = #{fragment := {filemeta, _}}) ->
append_filemeta(Asm, Node, Fragment); append_filemeta(Asm, Node, Fragment);
append(Asm, Node, Segment = #{fragment := {segmentinfo, _}}) -> append(Asm, Node, Segment = #{fragment := {segment, _}}) ->
append_segmentinfo(Asm, Node, Segment). append_segmentinfo(Asm, Node, Segment).
update(Asm) -> update(Asm) ->
@ -105,7 +105,7 @@ append_filemeta(Asm, Node, Fragment = #{fragment := {filemeta, Meta}}) ->
meta = orddict:store(Meta, {Node, Fragment}, Asm#asm.meta) meta = orddict:store(Meta, {Node, Fragment}, Asm#asm.meta)
}. }.
append_segmentinfo(Asm, Node, Fragment = #{fragment := {segmentinfo, Info}}) -> append_segmentinfo(Asm, Node, Fragment = #{fragment := {segment, Info}}) ->
Offset = maps:get(offset, Info), Offset = maps:get(offset, Info),
Size = maps:get(size, Info), Size = maps:get(size, Info),
End = Offset + Size, End = Offset + Size,
@ -176,7 +176,7 @@ locality(Node) when Node =:= node() ->
locality(_RemoteNode) -> locality(_RemoteNode) ->
1. 1.
segsize(#{fragment := {segmentinfo, Info}}) -> segsize(#{fragment := {segment, Info}}) ->
maps:get(size, Info). maps:get(size, Info).
-ifdef(TEST). -ifdef(TEST).
@ -195,8 +195,8 @@ incomplete_test() ->
status( status(
update( update(
append(new(), node(), [ append(new(), node(), [
segmentinfo(p1, 0, 42), segment(p1, 0, 42),
segmentinfo(p1, 42, 100) segment(p1, 42, 100)
]) ])
) )
) )
@ -204,14 +204,14 @@ incomplete_test() ->
consistent_test() -> consistent_test() ->
Asm1 = append(new(), n1, [filemeta(m1, "blarg")]), Asm1 = append(new(), n1, [filemeta(m1, "blarg")]),
Asm2 = append(Asm1, n2, [segmentinfo(s2, 0, 42)]), Asm2 = append(Asm1, n2, [segment(s2, 0, 42)]),
Asm3 = append(Asm2, n3, [filemeta(m3, "blarg")]), Asm3 = append(Asm2, n3, [filemeta(m3, "blarg")]),
?assertMatch({complete, _}, status(meta, Asm3)). ?assertMatch({complete, _}, status(meta, Asm3)).
inconsistent_test() -> inconsistent_test() ->
Asm1 = append(new(), node(), [segmentinfo(s1, 0, 42)]), Asm1 = append(new(), node(), [segment(s1, 0, 42)]),
Asm2 = append(Asm1, n1, [filemeta(m1, "blarg")]), Asm2 = append(Asm1, n1, [filemeta(m1, "blarg")]),
Asm3 = append(Asm2, n2, [segmentinfo(s2, 0, 42), filemeta(m1, "blorg")]), Asm3 = append(Asm2, n2, [segment(s2, 0, 42), filemeta(m1, "blorg")]),
Asm4 = append(Asm3, n3, [filemeta(m3, "blarg")]), Asm4 = append(Asm3, n3, [filemeta(m3, "blarg")]),
?assertMatch( ?assertMatch(
{error, {error,
@ -226,10 +226,10 @@ inconsistent_test() ->
simple_coverage_test() -> simple_coverage_test() ->
Node = node(), Node = node(),
Segs = [ Segs = [
{node42, segmentinfo(n1, 20, 30)}, {node42, segment(n1, 20, 30)},
{Node, segmentinfo(n2, 0, 10)}, {Node, segment(n2, 0, 10)},
{Node, segmentinfo(n3, 50, 50)}, {Node, segment(n3, 50, 50)},
{Node, segmentinfo(n4, 10, 10)} {Node, segment(n4, 10, 10)}
], ],
Asm = append_many(new(), Segs), Asm = append_many(new(), Segs),
?assertMatch( ?assertMatch(
@ -247,14 +247,14 @@ simple_coverage_test() ->
redundant_coverage_test() -> redundant_coverage_test() ->
Node = node(), Node = node(),
Segs = [ Segs = [
{Node, segmentinfo(n1, 0, 20)}, {Node, segment(n1, 0, 20)},
{node1, segmentinfo(n2, 0, 10)}, {node1, segment(n2, 0, 10)},
{Node, segmentinfo(n3, 20, 40)}, {Node, segment(n3, 20, 40)},
{node2, segmentinfo(n4, 10, 10)}, {node2, segment(n4, 10, 10)},
{node2, segmentinfo(n5, 50, 20)}, {node2, segment(n5, 50, 20)},
{node3, segmentinfo(n6, 20, 20)}, {node3, segment(n6, 20, 20)},
{Node, segmentinfo(n7, 50, 10)}, {Node, segment(n7, 50, 10)},
{node1, segmentinfo(n8, 40, 10)} {node1, segment(n8, 40, 10)}
], ],
Asm = append_many(new(), Segs), Asm = append_many(new(), Segs),
?assertMatch( ?assertMatch(
@ -272,12 +272,12 @@ redundant_coverage_test() ->
redundant_coverage_prefer_local_test() -> redundant_coverage_prefer_local_test() ->
Node = node(), Node = node(),
Segs = [ Segs = [
{node1, segmentinfo(n1, 0, 20)}, {node1, segment(n1, 0, 20)},
{Node, segmentinfo(n2, 0, 10)}, {Node, segment(n2, 0, 10)},
{Node, segmentinfo(n3, 10, 10)}, {Node, segment(n3, 10, 10)},
{node2, segmentinfo(n4, 20, 20)}, {node2, segment(n4, 20, 20)},
{Node, segmentinfo(n5, 30, 10)}, {Node, segment(n5, 30, 10)},
{Node, segmentinfo(n6, 20, 10)} {Node, segment(n6, 20, 10)}
], ],
Asm = append_many(new(), Segs), Asm = append_many(new(), Segs),
?assertMatch( ?assertMatch(
@ -295,11 +295,11 @@ redundant_coverage_prefer_local_test() ->
missing_coverage_test() -> missing_coverage_test() ->
Node = node(), Node = node(),
Segs = [ Segs = [
{Node, segmentinfo(n1, 0, 10)}, {Node, segment(n1, 0, 10)},
{node1, segmentinfo(n3, 10, 20)}, {node1, segment(n3, 10, 20)},
{Node, segmentinfo(n2, 0, 20)}, {Node, segment(n2, 0, 20)},
{node2, segmentinfo(n4, 50, 50)}, {node2, segment(n4, 50, 50)},
{Node, segmentinfo(n5, 40, 60)} {Node, segment(n5, 40, 60)}
], ],
Asm = append_many(new(), Segs), Asm = append_many(new(), Segs),
?assertEqual( ?assertEqual(
@ -311,8 +311,8 @@ missing_coverage_test() ->
missing_end_coverage_test() -> missing_end_coverage_test() ->
Node = node(), Node = node(),
Segs = [ Segs = [
{Node, segmentinfo(n1, 0, 15)}, {Node, segment(n1, 0, 15)},
{node1, segmentinfo(n3, 10, 10)} {node1, segment(n3, 10, 10)}
], ],
Asm = append_many(new(), Segs), Asm = append_many(new(), Segs),
?assertEqual( ?assertEqual(
@ -322,11 +322,11 @@ missing_end_coverage_test() ->
missing_coverage_with_redudancy_test() -> missing_coverage_with_redudancy_test() ->
Segs = [ Segs = [
{node(), segmentinfo(n1, 0, 10)}, {node(), segment(n1, 0, 10)},
{node(), segmentinfo(n2, 0, 20)}, {node(), segment(n2, 0, 20)},
{node42, segmentinfo(n3, 10, 20)}, {node42, segment(n3, 10, 20)},
{node43, segmentinfo(n4, 10, 50)}, {node43, segment(n4, 10, 50)},
{node(), segmentinfo(n5, 40, 60)} {node(), segment(n5, 40, 60)}
], ],
Asm = append_many(new(), Segs), Asm = append_many(new(), Segs),
?assertEqual( ?assertEqual(
@ -351,11 +351,11 @@ filemeta(Path, Name) ->
}} }}
}. }.
segmentinfo(Path, Offset, Size) -> segment(Path, Offset, Size) ->
#{ #{
path => Path, path => Path,
fragment => fragment =>
{segmentinfo, #{ {segment, #{
offset => Offset, offset => Offset,
size => Size size => Size
}} }}