From f4e2e274e09161b4b3d4407044c59d9940f92003 Mon Sep 17 00:00:00 2001
From: martinsumner <martin.sumner@adaptip.co.uk>
Date: Wed, 14 Dec 2016 10:27:11 +0000
Subject: [PATCH] Reintroduce riak metadata extraction

The full riak metadata had been stripped from the Ledger update for
performance reasons.  However, the full metadata is required in order to
save a GET before a PUT.  Therefore we want to do isolated testing on
this change to establish the relative cost value in that cost saving.
---
 src/leveled_codec.erl | 54 +++++++++++++++++++++++++++++++++++--------
 1 file changed, 45 insertions(+), 9 deletions(-)

diff --git a/src/leveled_codec.erl b/src/leveled_codec.erl
index f08e2e9..132506c 100644
--- a/src/leveled_codec.erl
+++ b/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,
-            riak_metadata_to_binary(Vclock, SibCount);
+            {SibData, Vclock, _Hash, _Size} = MD,
+            riak_metadata_to_binary(Vclock, SibData);
         ?STD_TAG ->
             MD
     end.
@@ -373,24 +373,60 @@ 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),
-    {SibCount, Vclock, erlang:phash2(ObjBin), Size}.
+    {Vclock, SibData} = riak_metadata_from_binary(ObjBin),
+    {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,
-    {binary_to_term(VclockBin), SibCount}.
-   
+    % Just grab the Sibling count and not the full metadata
+    % <<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]).
+