From 2e5b9c80f445dbb03e43dad11dfecc9296113916 Mon Sep 17 00:00:00 2001
From: Martin Sumner <martin.sumner@adaptip.co.uk>
Date: Wed, 27 Sep 2017 12:15:18 +0100
Subject: [PATCH] Add max LMD to Riak metadata

This is an interim stage towwards enhancing the proxy object so that it contains more helper information (other than size).

The aim is to be able to run more efficient fold_heads queries that might filter on LMD range (so as not to have to co-ordinate the running of comparative queries).  For example if producing a tictactree to compare between two different offsets, a max LMD could be passed in so that changes beyond the time the first query was requested can be ignored.
---
 src/leveled_codec.erl  | 25 ++++++++++++++++++++++---
 src/leveled_tictac.erl |  4 ++--
 2 files changed, 24 insertions(+), 5 deletions(-)

diff --git a/src/leveled_codec.erl b/src/leveled_codec.erl
index 6be90d7..e3ca2f4 100644
--- a/src/leveled_codec.erl
+++ b/src/leveled_codec.erl
@@ -74,6 +74,11 @@
 -define(ALL_BUCKETS, <<"$all">>).
 
 -type recent_aae() :: #recent_aae{}.
+-type riak_metadata() :: {binary()|delete, % Sibling Metadata
+                            binary()|null, % Vclock Metadata
+                            integer()|null, % Hash of vclock
+                            {integer(), integer(), integer()}|null, % LMOD TS 
+                            integer()}. % Size in bytes of real object
 
 -spec magic_hash(any()) -> integer().
 %% @doc 
@@ -583,7 +588,7 @@ get_keyandobjhash(LK, Value) ->
 get_objhash(Tag, ObjMetaData) ->
     case Tag of
         ?RIAK_TAG ->
-            {_RMD, _VC, Hash, _Size} = ObjMetaData,
+            {_RMD, _VC, Hash, _LMD, _Size} = ObjMetaData,
             Hash;
         ?STD_TAG ->
             {Hash, _Size} = ObjMetaData,
@@ -595,20 +600,34 @@ build_metadata_object(PrimaryKey, MD) ->
     {Tag, _Bucket, _Key, null} = PrimaryKey,
     case Tag of
         ?RIAK_TAG ->
-            {SibData, Vclock, _Hash, _Size} = MD,
+            {SibData, Vclock, _Hash, _LMD, _Size} = MD,
             riak_metadata_to_binary(Vclock, SibData);
         ?STD_TAG ->
             MD
     end.
 
 
+-spec riak_extract_metadata(binary()|delete, non_neg_integer()) -> 
+                            {riak_metadata(), list()}.
+%% @doc
+%% Riak extract metadata should extract a metadata object which is a 
+%% five-tuple of:
+%% - Binary of sibling Metadata
+%% - Binary of vector clock metadata
+%% - Non-exportable hash of the vector clock metadata
+%% - The largest last modified date of the object
+%% - Size of the object
+%%
+%% The metadata object should be returned with the full list of last 
+%% modified dates (which will be used for recent anti-entropy index creation)
 riak_extract_metadata(delete, Size) ->
-    {{delete, null, null, Size}, []};
+    {{delete, null, null, null, Size}, []};
 riak_extract_metadata(ObjBin, Size) ->
     {VclockBin, SibBin, LastMods} = riak_metadata_from_binary(ObjBin),
     {{SibBin, 
             VclockBin, 
             erlang:phash2(lists:sort(binary_to_term(VclockBin))), 
+            lists:max(LastMods),
             Size},
         LastMods}.
 
diff --git a/src/leveled_tictac.erl b/src/leveled_tictac.erl
index d25132c..0ba615b 100644
--- a/src/leveled_tictac.erl
+++ b/src/leveled_tictac.erl
@@ -296,7 +296,7 @@ get_segment(Hash, TreeSize) ->
     get_segment(Hash, element(3, get_size(TreeSize))).
 
 
--spec tictac_hash(any(), any(), boolean()) -> integer().
+-spec tictac_hash(any(), any(), boolean()) -> {integer(), integer()}.
 %% @doc
 %% Hash the key and term, to either something repetable in Erlang, or using 
 %% the DJ Bernstein hash if it is the tree needs to be compared with one 
@@ -477,7 +477,7 @@ merge_test_withsize(Size) ->
 
 exportable_test() ->
     {Int1, Int2} = tictac_hash(<<"key">>, <<"value">>, true),
-    ?assertMatch({true, true}, {is_integer(Int1), is_integer(Int2)}).
+    ?assertMatch({true, true}, {Int1 >= 0, Int2 >=0}).
 
 -endif.