Merge pull request #10 from martinsumner/mas_riakmetadata

Full Riak Metadata

src/leveled_codec.erl

@@ -363,8 +363,8 @@ build_metadata_object(PrimaryKey, MD) ->
     {Tag, _Bucket, _Key, null} = PrimaryKey,
     case Tag of
         ?RIAK_TAG ->
-            {SibCount, Vclock, _Hash, _Size} = MD,
+            {SibData, Vclock, _Hash, _Size} = MD,
-            riak_metadata_to_binary(Vclock, SibCount);
+            riak_metadata_to_binary(Vclock, SibData);
         ?STD_TAG ->
             MD
     end.
@@ -373,23 +373,59 @@ build_metadata_object(PrimaryKey, MD) ->
 riak_extract_metadata(delete, Size) ->
     {delete, null, null, Size};
 riak_extract_metadata(ObjBin, Size) ->
-    {Vclock, SibCount} = riak_metadata_from_binary(ObjBin),
+    {Vclock, SibData} = riak_metadata_from_binary(ObjBin),
-    {SibCount, Vclock, erlang:phash2(ObjBin), Size}.
+    {SibData, Vclock, erlang:phash2(ObjBin), Size}.
 
 %% <<?MAGIC:8/integer, ?V1_VERS:8/integer, VclockLen:32/integer,
 %%%     VclockBin/binary, SibCount:32/integer, SibsBin/binary>>.
 
-riak_metadata_to_binary(Vclock, SibCount) ->
+riak_metadata_to_binary(Vclock, SibData) ->
     VclockBin = term_to_binary(Vclock),
     VclockLen = byte_size(VclockBin),
+    % <<?MAGIC:8/integer, ?V1_VERS:8/integer, VclockLen:32/integer,
+    %         VclockBin:VclockLen/binary, SibData:32/integer>>.
+    SibCount = length(SibData),
+    SibsBin = slimbin_contents(SibData),
     <<?MAGIC:8/integer, ?V1_VERS:8/integer, VclockLen:32/integer,
-            VclockBin:VclockLen/binary, SibCount:32/integer>>.
+            VclockBin:VclockLen/binary, SibCount:32/integer, SibsBin/binary>>.
     
 riak_metadata_from_binary(V1Binary) ->
     <<?MAGIC:8/integer, ?V1_VERS:8/integer, VclockLen:32/integer,
             Rest/binary>> = V1Binary,
-    <<VclockBin:VclockLen/binary, SibCount:32/integer, _Rest/binary>> = Rest,
+    % Just grab the Sibling count and not the full metadata
-    {binary_to_term(VclockBin), SibCount}.
+    % <<VclockBin:VclockLen/binary, SibCount:32/integer, _Rest/binary>> = Rest,
+    % {binary_to_term(VclockBin), SibCount}.
+    <<VclockBin:VclockLen/binary, SibCount:32/integer, SibsBin/binary>> = Rest,
+    SibMetaBinList =
+        case SibCount of
+            0 ->
+                [];
+            SC when is_integer(SC) ->
+                get_metadata_from_siblings(SibsBin, SibCount, [])
+        end,
+    {binary_to_term(VclockBin), SibMetaBinList}.
+
+% Fixes the value length for each sibling to be zero, and so includes no value
+slimbin_content(MetaBin) ->
+    MetaLen = byte_size(MetaBin),
+    <<0:32/integer,  MetaLen:32/integer, MetaBin:MetaLen/binary>>.
+
+slimbin_contents(SibMetaBinList) ->
+    F = fun(MetaBin, Acc) ->
+                <<Acc/binary, (slimbin_content(MetaBin))/binary>>
+        end,
+    lists:foldl(F, <<>>, SibMetaBinList).
+    
+get_metadata_from_siblings(<<>>, 0, SibMetaBinList) ->
+    SibMetaBinList;
+get_metadata_from_siblings(<<ValLen:32/integer, Rest0/binary>>,
+                            SibCount,
+                            SibMetaBinList) ->
+    <<_ValBin:ValLen/binary, MetaLen:32/integer, Rest1/binary>> = Rest0,
+    <<MetaBin:MetaLen/binary, Rest2/binary>> = Rest1,
+    get_metadata_from_siblings(Rest2,
+                                SibCount - 1,
+                                [MetaBin|SibMetaBinList]).
 
 
 

src/leveled_sft.erl

@@ -179,7 +179,7 @@
 -define(DWORD_SIZE, 8).
 -define(CURRENT_VERSION, {0,1}).
 -define(SLOT_COUNT, 256).
--define(SLOT_GROUPWRITE_COUNT, 64).
+-define(SLOT_GROUPWRITE_COUNT, 16).
 -define(BLOCK_SIZE, 32).
 -define(BLOCK_COUNT, 4).
 -define(FOOTERPOS_HEADERPOS, 2).