umgeher/leveled
0
Fork 0
Code Issues Pull requests Releases Activity Actions

Merge pull request #86 from martinsumner/mas-aae-riakimplnotes

Browse source 
Mas aae riakimplnotes
This commit is contained in:
Martin Sumner 2017-08-10 17:16:40 +01:00 • committed by GitHub
commit 5f3d44ffca
8 changed files with 474 additions and 247 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -2,3 +2,4 @@
/.eunit /.eunit
/_build /_build
*~ *~
.DS_Store

108
docs/ANTI_ENTROPY.md
View file

@ -126,6 +126,14 @@ The AAE store is re-usable for checking consistency between databases, but with
The AAE process in production system commonly raises false positives (prompts repairs that are unnecessary), sometimes for [known reasons](https://github.com/basho/riak_kv/issues/1189), sometimes for unknown reasons, especially following rebuilds which follow ring changes. The repair process has [a throttle](http://docs.basho.com/riak/kv/2.2.3/using/cluster-operations/active-anti-entropy/#throttling) to prevent this from impacting a production system, but this commonly needs to be re-tuned based on experience. The AAE process in production system commonly raises false positives (prompts repairs that are unnecessary), sometimes for [known reasons](https://github.com/basho/riak_kv/issues/1189), sometimes for unknown reasons, especially following rebuilds which follow ring changes. The repair process has [a throttle](http://docs.basho.com/riak/kv/2.2.3/using/cluster-operations/active-anti-entropy/#throttling) to prevent this from impacting a production system, but this commonly needs to be re-tuned based on experience.
There is also a significant cost with AAE. The following chart compares performance with a mixed load test including 2% 2i queries, majority GETs (80% approx), minority PUTs (20% approx) with a large potential key space. The load test was run on a 5-node i2.xlarge cluster, comparing Riak performance with either a leveldb or leveled backend with and without anti-entropy enabled.
![](pics/BaselineCompare_with2i_AAE.png)
The purple and blue lines sho Riak+leveled throughput with/without AAE enabled, the reg and orange lines show the same for Riak+leveldb.
The AAE configuration used in this test was aggressive in terms of AAE tree rebuild (every four hours), but the significant gap in throughput seen with and without AAE persists even during spells in the test when there is no rebuild activity taking place (e.g. 13,000 to 15,000 elapsed seconds). Maintaining an additional key-store strictly for entropy management places a non-trivial cost on throughput of a heavily loaded cluster.
### Proposed Leveled AAE ### Proposed Leveled AAE
The first stage in considering an alternative approach to anti-entropy, was to question the necessity of having a dedicated AAE database that needs to reflect all key changes in the actual vnode store. This separate store is currently necessary as the hashtree needs a store sorted by segment ID that make that store easier to scan for rebuilds of the tree, hence avoiding the three main costs with scanning over the primary database: The first stage in considering an alternative approach to anti-entropy, was to question the necessity of having a dedicated AAE database that needs to reflect all key changes in the actual vnode store. This separate store is currently necessary as the hashtree needs a store sorted by segment ID that make that store easier to scan for rebuilds of the tree, hence avoiding the three main costs with scanning over the primary database:
@ -279,3 +287,103 @@ Some notes on re-using this alternative anti-entropy mechanism within Riak:
* The initial intention is to implement the hashtree query functions based around the coverage_fsm behaviour, but with the option to stipulate externally the offset. So to test for differences between clusters, the user could concurrently query the two clusters for the same offset (or a random offset), whereas to find entropy within a cluster two concurrently run queries could be compared for different offsets. * The initial intention is to implement the hashtree query functions based around the coverage_fsm behaviour, but with the option to stipulate externally the offset. So to test for differences between clusters, the user could concurrently query the two clusters for the same offset (or a random offset), whereas to find entropy within a cluster two concurrently run queries could be compared for different offsets.
* A surprising feature of read repair is that it will read repair to fallback nodes, not just primary nodes. This means that in read-intensive workloads, write activity may dramatically increase during node failure (as a large proportion of reads will become write events) - increasing the chance of servers falling domino style. However, in some circumstances the extra duplication can also [increase the chance of data loss](https://github.com/russelldb/russelldb.github.io/blob/master/3.2.kv679-solution.md)! This also increases greatly the volume of unnecessary data to be handed-off when the primary returns. Without active anti-entropy, and in the absence of other safety checks like `notfound_ok` being set to false, or `pr` being set to at least 1 - there will be scenarios where this feature may be helpful. As part of improving active anti-entropy, it may be wise to re-visit the tuning of anti-entropy features that existed prior to AAE, in particular should it be possible to configure read-repair to act on primary nodes only. * A surprising feature of read repair is that it will read repair to fallback nodes, not just primary nodes. This means that in read-intensive workloads, write activity may dramatically increase during node failure (as a large proportion of reads will become write events) - increasing the chance of servers falling domino style. However, in some circumstances the extra duplication can also [increase the chance of data loss](https://github.com/russelldb/russelldb.github.io/blob/master/3.2.kv679-solution.md)! This also increases greatly the volume of unnecessary data to be handed-off when the primary returns. Without active anti-entropy, and in the absence of other safety checks like `notfound_ok` being set to false, or `pr` being set to at least 1 - there will be scenarios where this feature may be helpful. As part of improving active anti-entropy, it may be wise to re-visit the tuning of anti-entropy features that existed prior to AAE, in particular should it be possible to configure read-repair to act on primary nodes only.
## Some notes on the experience of Riak implementation
### Phase 1 - Initial Test of Folds with Core node_worker_pool
As an initial proving stage of implementation, the riak_core_node_worker_pool has been implemented in riak_kv and riak_core, and then the listkeys function has been change so that it can be switched between using the node_worker_pool (new behaviour) and running in parallel using the vnode_worker_pool (old behaviour).
This also required a change to the backends for leveldb and leveled (there was not a quick way of making this change in bitcask), such that they could be changed to use a new ``snap_prefold`` capability. With snap_prefold configured (and requested by the kv_vnode), the backend will return a folder over a database snapshot which has already been taken. So rather than returning ``{async, Folder}`` whereby calling ``Folder()`` would make the snapshot then run the fold, this can now return ``{queue, Folder}`` where the snapshot has already been taken so if the calling of ``Folder()`` is deferred due to queueing it will still be approximately consistent to the time the request was initiated.
The following branches are required to provide this feature:
```
git clone -b master https://github.com/martinsumner/leveled.git
git clone -b mas-nodeworkerpool https://github.com/martinsumner/riak_core.git
git clone -b mas-leveled-corenodeworker https://github.com/martinsumner/riak_kv.git
git clone -b mas-leveled-2.0.34 https://github.com/martinsumner/eleveldb.git
```
The purpose of these branches was to do an initial test of how bad the problem of running listkeys is, how does running listkeys compare between leveldb and leveled, and how does the impact change if the node_worker_pool rather than the vnode_worker_pool is used (i.e. with and without the ``snap_prefold`` capability enabled).
This was tested with the following configuration:
- 5-node cluster, i2.2xlarge
- 64-partition ring-size
- 100K GETs : 20K UPDATEs : 1 LIST_KEYS operations ratios
- 6KB fixed object size (smaller object size chosen when compared to previous tests to try and reduce the relative importance of write amplification in leveldb, and see the listkeys impact more clearly)
- 400M key-space (pareto distribution)
This initial test showed that there was a minimal impact on throughput with running these listkeys operations leveldb when having ``snap_prefold`` either disabled or enabled. All three leveldb tests (without listkeys, with listkeys and snap_prefold, with listkeys and without snap_prefold) achieved an overall throughput within the margin of error of cloud testing.
![](pics/volume_listkeys_compare_6KB.png)
For the 6-hour test the total throughput achieved was:
- leveldb with no listkeys - 272.4M
- leveldb with listkeys vnode_pool - 270.1M (- 0.85%)
- leveldb with listkeys node_pool - 278.0M (+ 2.04%)
- leveled with listkeys node_pool - 333.0M (+ 22.23%)
Of note is the escalating response times of the fold as the size of the database increased. The nature of the cluster would mean that with unconstrained hardware resource the node_worker_pool should take 5 x longer than the vnode_worker_pool - but within the test it was generally less than 3 times longer. However, this was 3 times longer than a large time which increased with database size in a non-linear fashion.
With leveled there was a strongly favourable comparison, with both improved response times, and less dramatic rises in those times as the database grew.
![](pics/listkeys_jobtime_compare.png)
Even when using the (constrained) node_worker_pool the leveled implementation runs list_keys jobs at a rate of 155K keys per second. By contrast leveldb runs at 45K keys per second with the vnode_worker_pool and just 16K keys per second when throttled by using the node_worker_pool.
These rates change with object size with leveldb, but not with leveled. so when testing with object sizes of 10KB, rates drop to 27K keys per second for the same test with leveldb using the vnode_worker_pool. However, rates are unchanged for leveled, as the operation in leveled is independent of object size (as object values do not need to be scanned).
The intention is though, not to use list_keys operations in AAE - but instead to dynamically produce TicTac Trees either by folding objects (or heads with leveled) or folding over specialist indexes. More relevant tests will follow in the next phase.
### Some Implementation Considerations
One thing that deserves debate with regards to multi-data-centre replication, is should the existing AAE process just be abandoned. There have been three considerations that have driven this implementation towards considering abandonment:
- The bad history of unexplained problems. Issues with AAE commonly been associated with [hashing of unsorted objects](https://github.com/basho/riak_kv/issues/1189), but issues continue to be seen even when running versions that have this resolved. It is feared that the knowledge of the existence of this issue has led to other AAE issues not being investigated, as it was more efficient to simply assume all AAE issues were ultimately caused by Issue 1189.
- The problems of upgrading the hashtree. A software change which alters the form of the existing hashtree is not going to be easy to implement, and includes a non-functional risk associated with handling two hashtrees in parallel for a transition period.
- A general sense of complexity with regards to existing AAE, especially around locking and rebuilding. Also a general fear of the complexity and long-term supportability of eleveldb: although the historical stability of the project indicates it wis well written, and each branch is documented to a high standard, it requires a significant context shift for those used to working with Erlang.
- The performance overheads of AAE stores (see testing results above). These will vary between setups, and depend heavily on the size of the key-space and the write-throughput - but managing another key store for the purpose of anti-entropy is non-trivial.
With a leveled backend, and the efficient support for dynamically building a TicTac tree without a separate AAE database - the AAE database appears now redundant and its disadvantages means an legacy-AAE-free approach seems optimal. However, there are problems with trying the same new approach with eleveldb and bitcask backends:
- the slow fold times highlighted in [phase 1 testing](ANTI_ENTROPY.md#phase-1---initial-test-of-folds-with-core-node_worker_pool) when using leveldb backends.
- the difficulty of managing bitcask snapshots, as they appear to require the database to be re-opened by another process - and so may require all keys to be held in memory twice during the life of the snapshot. Presumably bitcask folds will also be similarly slow (although they were not tested in Phase 1).
It should also be noted that:
- there is no advantage of removing eleveldb as a dependency, if it is still in use as a backend.
- there may be a history of false accusations of AAE problems, as the AAE throttle is applied in Riak even when AAE is not disabled - so the correlation of AAE throttling announcements in logs and short-term Riak performance issues may have created a false impression of this stability being AAE related. The riak_kv_entropy_manager applies the throttle whenever there are node connectivity issues, regardless of AAE activity.
- if a consistent hashing algorithm is forced (between AAE merkle trees and TicTac trees), it should be possible to scan a legacy AAE store and produce a mergeable TicTac tree. There may be a simple transition from Merkle trees as-is to TicTac trees to-be, which will allow for mixed ring-size multi-data-centre comparison (i.e. by running a coverage fold over the AAE trees).
So moving forward to phase 2, consideration is being given to having a ``native_aaetree`` capability, that would be supported by the Leveled backend. If this capability exists, then when handling a coverage fold for an AAE report, the riak_kv_vnode would request a ``{queue, Folder}`` directly from the backend to be sent to the core_node_worker_pool. If such a capability doesn't exist, the ``{queue, Folder}`` would instead be requested from the hashtree PID for that vnode, not the actual backend. This would mean any existing leveldb or bitcask users could continue to use AAE as-is, whilst still gaining the support for open-source MDC comparison independent of ring-sizes.
The primary issue that such an implementation would need to resolve is how to handle the situation when the AAE tree is not in a usable state (for example whilst awaiting completion of a build or a rebuild). Would it be possible to wait for the hashtree to be available? Could replication comparisons be suspended whilst hashtrees are not ready? Could coverage plans be re-computed to account for the fact that no lock can be obtained on the clocked hahstree?
#### AAE Hashtree locks
The AAE hashtree lock situation is complex, but can be summarised as:
- there are three types of relevant locks: a lock on the hashtree itself (acquired via riak_kv_index_hashtree:get_lock), a concurrency lock managed by the riak_kv_entropy_manager, another concurrency lock managed by riak_core_bg_manager.
- the hashtree lock when acquired is released only by the death of the acquiring process (so this lock should only be acquired by temporary processes such as a riak_kv_exchange_fsm).
- to acquire the hashtree lock, it needs to not be acquired, but also the ``built`` state of the hashtree store must be true.
- when rebuilding a hashtree (for example after expiry), the ``built`` state is changed from true, so that whilst the hashtree is being built a lock cannot be acquired for na exchange.
- it seems likely that a fold over the hashtree should be safe even when lock has been acquired for an exchange (as long as there is no overlap of flushing the in-memory queue), but not when the state of the hashtree is not built.
- however during a coverage fold for MDC AAE, a build may crash the fold so there may be a need to check for running folds before prompting a rebuild.
### Phase 2
tbc

BIN
docs/pics/.DS_Store vendored Normal file
View file

Binary file not shown.

BIN
docs/pics/BaselineCompare_with2i_AAE.png Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 630 KiB

BIN
docs/pics/listkeys_jobtime_compare.png Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 206 KiB

BIN
docs/pics/volume_listkeys_compare_6KB.png Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 659 KiB

242
src/leveled_bookie.erl
View file

@ -16,7 +16,7 @@
%% %%
%% %%
%% -------- Actors --------- %% -------- Actors ---------
%% %%
%% The store is fronted by a Bookie, who takes support from different actors: %% The store is fronted by a Bookie, who takes support from different actors:
%% - An Inker who persists new data into the journal, and returns items from %% - An Inker who persists new data into the journal, and returns items from
%% the journal based on sequence number %% the journal based on sequence number
@ -70,7 +70,7 @@
loadqueue_ledgercache/1, loadqueue_ledgercache/1,
push_ledgercache/2, push_ledgercache/2,
snapshot_store/5, snapshot_store/5,
fetch_value/2]). fetch_value/2]).
-include_lib("eunit/include/eunit.hrl"). -include_lib("eunit/include/eunit.hrl").
@ -102,6 +102,7 @@
put_timing :: tuple() | undefined, put_timing :: tuple() | undefined,
get_timing :: tuple() | undefined}). get_timing :: tuple() | undefined}).
-type book_state() :: #state{}.
%%%============================================================================ %%%============================================================================
%%% API %%% API
@ -113,7 +114,7 @@
%% provide this startup method. This will start a KV store from the previous %% provide this startup method. This will start a KV store from the previous
%% store at root path - or an empty one if there is no store at the path. %% store at root path - or an empty one if there is no store at the path.
%% %%
%% Fiddling with the LedgerCacheSize and JournalSize may improve performance, %% Fiddling with the LedgerCacheSize and JournalSize may improve performance,
%% but these are primarily exposed to support special situations (e.g. very %% but these are primarily exposed to support special situations (e.g. very
%% low memory installations), there should not be huge variance in outcomes %% low memory installations), there should not be huge variance in outcomes
%% from modifying these numbers. %% from modifying these numbers.
@ -158,7 +159,7 @@ book_start(RootPath, LedgerCacheSize, JournalSize, SyncStrategy) ->
%% up before deletion) %% up before deletion)
%% %%
%% TODO: %% TODO:
%% The reload_strategy is exposed as currently no firm decision has been made %% The reload_strategy is exposed as currently no firm decision has been made
%% about how recovery should work. For instance if we were to trust everything %% about how recovery should work. For instance if we were to trust everything
%% as permanent in the Ledger once it is persisted, then there would be no %% as permanent in the Ledger once it is persisted, then there would be no
%% need to retain a skinny history of key changes in the Journal after %% need to retain a skinny history of key changes in the Journal after
@ -200,7 +201,7 @@ book_tempput(Pid, Bucket, Key, Object, IndexSpecs, Tag, TTL)
%% - A Primary Key and a Value %% - A Primary Key and a Value
%% - IndexSpecs - a set of secondary key changes associated with the %% - IndexSpecs - a set of secondary key changes associated with the
%% transaction %% transaction
%% - A tag indictaing the type of object. Behaviour for metadata extraction, %% - A tag indictaing the type of object. Behaviour for metadata extraction,
%% and ledger compaction will vary by type. There are three currently %% and ledger compaction will vary by type. There are three currently
%% implemented types i (Index), o (Standard), o_rkv (Riak). Keys added with %% implemented types i (Index), o (Standard), o_rkv (Riak). Keys added with
%% Index tags are not fetchable (as they will not be hashed), but are %% Index tags are not fetchable (as they will not be hashed), but are
@ -314,9 +315,9 @@ book_head(Pid, Bucket, Key, Tag) ->
%% objects with a given tag %% objects with a given tag
%% {tictactree_idx, %% {tictactree_idx,
%% {Bucket, IdxField, StartValue, EndValue}, %% {Bucket, IdxField, StartValue, EndValue},
%% TreeSize, %% TreeSize,
%% PartitionFilter} %% PartitionFilter}
%% -> compile a hashtree for the items on the index. A partition filter is %% -> compile a hashtree for the items on the index. A partition filter is
%% required to avoid adding an index entry in this vnode as a fallback. %% required to avoid adding an index entry in this vnode as a fallback.
%% There is no de-duplicate of results, duplicate reuslts corrupt the tree. %% There is no de-duplicate of results, duplicate reuslts corrupt the tree.
%% {tictactree_obj, %% {tictactree_obj,
@ -384,7 +385,7 @@ init([Opts]) ->
undefined -> undefined ->
% Start from file not snapshot % Start from file not snapshot
{InkerOpts, PencillerOpts} = set_options(Opts), {InkerOpts, PencillerOpts} = set_options(Opts),
CacheJitter = ?CACHE_SIZE div (100 div ?CACHE_SIZE_JITTER), CacheJitter = ?CACHE_SIZE div (100 div ?CACHE_SIZE_JITTER),
CacheSize = get_opt(cache_size, Opts, ?CACHE_SIZE) CacheSize = get_opt(cache_size, Opts, ?CACHE_SIZE)
+ erlang:phash2(self()) rem CacheJitter, + erlang:phash2(self()) rem CacheJitter,
@ -398,9 +399,9 @@ init([Opts]) ->
limit_minutes = LimitMinutes, limit_minutes = LimitMinutes,
unit_minutes = UnitMinutes} unit_minutes = UnitMinutes}
end, end,
{Inker, Penciller} = startup(InkerOpts, PencillerOpts, RecentAAE), {Inker, Penciller} = startup(InkerOpts, PencillerOpts, RecentAAE),
NewETS = ets:new(mem, [ordered_set]), NewETS = ets:new(mem, [ordered_set]),
leveled_log:log("B0001", [Inker, Penciller]), leveled_log:log("B0001", [Inker, Penciller]),
{ok, #state{inker=Inker, {ok, #state{inker=Inker,
@ -467,12 +468,12 @@ handle_call({get, Bucket, Key, Tag}, _From, State) ->
State#state.penciller, State#state.penciller,
State#state.ledger_cache) of State#state.ledger_cache) of
not_present -> not_present ->
GT0 = leveled_log:get_timing(State#state.get_timing, GT0 = leveled_log:get_timing(State#state.get_timing,
SWh, SWh,
head_not_present), head_not_present),
{reply, not_found, State#state{get_timing=GT0}}; {reply, not_found, State#state{get_timing=GT0}};
Head -> Head ->
GT0 = leveled_log:get_timing(State#state.get_timing, GT0 = leveled_log:get_timing(State#state.get_timing,
SWh, SWh,
head_found), head_found),
SWg = os:timestamp(), SWg = os:timestamp(),
@ -518,8 +519,8 @@ handle_call({head, Bucket, Key, Tag}, _From, State) ->
end end
end; end;
handle_call({snapshot, _Requestor, SnapType, _Timeout}, _From, State) -> handle_call({snapshot, _Requestor, SnapType, _Timeout}, _From, State) ->
% TODO: clean-up passing of Requestor (which was previously just used in % TODO: clean-up passing of Requestor (which was previously just used in
% logs) and so can now be ignored, and timeout which is ignored - but % logs) and so can now be ignored, and timeout which is ignored - but
% probably shouldn't be. % probably shouldn't be.
Reply = snapshot_store(State, SnapType), Reply = snapshot_store(State, SnapType),
{reply, Reply, State}; {reply, Reply, State};
@ -595,13 +596,21 @@ handle_call({return_folder, FolderType}, _From, State) ->
TreeSize, TreeSize,
PartitionFilter), PartitionFilter),
State}; State};
{foldheads_allkeys, Tag, FoldHeadsFun} -> {foldheads_allkeys, Tag, FoldHeadsFun,
CheckPresence, SnapPreFold} ->
{reply, {reply,
foldheads_allkeys(State, Tag, FoldHeadsFun), foldheads_allkeys(State, Tag,
FoldHeadsFun,
CheckPresence,
SnapPreFold),
State}; State};
{foldheads_bybucket, Tag, Bucket, FoldHeadsFun} -> {foldheads_bybucket, Tag, Bucket, FoldHeadsFun,
CheckPresence, SnapPreFold} ->
{reply, {reply,
foldheads_bybucket(State, Tag, Bucket, FoldHeadsFun), foldheads_bybucket(State, Tag, Bucket,
FoldHeadsFun,
CheckPresence,
SnapPreFold),
State}; State};
{foldobjects_allkeys, Tag, FoldObjectsFun} -> {foldobjects_allkeys, Tag, FoldObjectsFun} ->
{reply, {reply,
@ -622,7 +631,7 @@ handle_call({return_folder, FolderType}, _From, State) ->
Field, FromTerm, ToTerm, Field, FromTerm, ToTerm,
FoldObjectsFun), FoldObjectsFun),
State} State}
end; end;
handle_call({compact_journal, Timeout}, _From, State) -> handle_call({compact_journal, Timeout}, _From, State) ->
ok = leveled_inker:ink_compactjournal(State#state.inker, ok = leveled_inker:ink_compactjournal(State#state.inker,
@ -703,9 +712,9 @@ snapshot_store(LedgerCache0, Penciller, Inker, SnapType, Query) ->
LedgerCache#ledger_cache.index, LedgerCache#ledger_cache.index,
LedgerCache#ledger_cache.min_sqn, LedgerCache#ledger_cache.min_sqn,
LedgerCache#ledger_cache.max_sqn}, LedgerCache#ledger_cache.max_sqn},
LongRunning = LongRunning =
case Query of case Query of
undefined -> undefined ->
true; true;
no_lookup -> no_lookup ->
true; true;
@ -728,7 +737,7 @@ snapshot_store(LedgerCache0, Penciller, Inker, SnapType, Query) ->
{ok, LedgerSnapshot, JournalSnapshot}; {ok, LedgerSnapshot, JournalSnapshot};
ledger -> ledger ->
{ok, LedgerSnapshot, null} {ok, LedgerSnapshot, null}
end. end.
snapshot_store(State, SnapType) -> snapshot_store(State, SnapType) ->
snapshot_store(State, SnapType, undefined). snapshot_store(State, SnapType, undefined).
@ -837,7 +846,7 @@ get_nextbucket(NextBucket, NextKey, Tag, LedgerSnapshot, BKList) ->
leveled_log:log("B0010",[NB]), leveled_log:log("B0010",[NB]),
[] []
end. end.
index_query(State, index_query(State,
Bucket, Bucket,
@ -955,7 +964,7 @@ tictactree(State, Tag, Bucket, Query, JournalCheck, TreeSize, Filter) ->
Tag), Tag),
PassHashFun} PassHashFun}
end, end,
AccFun = accumulate_tree(Filter, AccFun = accumulate_tree(Filter,
JournalCheck, JournalCheck,
JournalSnapshot, JournalSnapshot,
@ -965,7 +974,7 @@ tictactree(State, Tag, Bucket, Query, JournalCheck, TreeSize, Filter) ->
EndKey, EndKey,
AccFun, AccFun,
Tree), Tree),
% Close down snapshot when complete so as not to hold removed % Close down snapshot when complete so as not to hold removed
% files open % files open
ok = leveled_penciller:pcl_close(LedgerSnapshot), ok = leveled_penciller:pcl_close(LedgerSnapshot),
@ -982,22 +991,26 @@ tictactree(State, Tag, Bucket, Query, JournalCheck, TreeSize, Filter) ->
foldobjects_allkeys(State, Tag, FoldObjectsFun) -> foldobjects_allkeys(State, Tag, FoldObjectsFun) ->
StartKey = leveled_codec:to_ledgerkey(null, null, Tag), StartKey = leveled_codec:to_ledgerkey(null, null, Tag),
EndKey = leveled_codec:to_ledgerkey(null, null, Tag), EndKey = leveled_codec:to_ledgerkey(null, null, Tag),
foldobjects(State, Tag, StartKey, EndKey, FoldObjectsFun, false). foldobjects(State, Tag, StartKey, EndKey, FoldObjectsFun,
false, true).
foldheads_allkeys(State, Tag, FoldHeadsFun) -> foldheads_allkeys(State, Tag, FoldHeadsFun, CheckPresence, SnapPreFold) ->
StartKey = leveled_codec:to_ledgerkey(null, null, Tag), StartKey = leveled_codec:to_ledgerkey(null, null, Tag),
EndKey = leveled_codec:to_ledgerkey(null, null, Tag), EndKey = leveled_codec:to_ledgerkey(null, null, Tag),
foldobjects(State, Tag, StartKey, EndKey, FoldHeadsFun, true). foldobjects(State, Tag, StartKey, EndKey, FoldHeadsFun,
{true, CheckPresence}, SnapPreFold).
foldobjects_bybucket(State, Tag, Bucket, FoldObjectsFun) -> foldobjects_bybucket(State, Tag, Bucket, FoldObjectsFun) ->
StartKey = leveled_codec:to_ledgerkey(Bucket, null, Tag), StartKey = leveled_codec:to_ledgerkey(Bucket, null, Tag),
EndKey = leveled_codec:to_ledgerkey(Bucket, null, Tag), EndKey = leveled_codec:to_ledgerkey(Bucket, null, Tag),
foldobjects(State, Tag, StartKey, EndKey, FoldObjectsFun, false). foldobjects(State, Tag, StartKey, EndKey, FoldObjectsFun,
false, true).
foldheads_bybucket(State, Tag, Bucket, FoldHeadsFun) -> foldheads_bybucket(State, Tag, Bucket, FoldHeadsFun, CheckPresence, SnapPreFold) ->
StartKey = leveled_codec:to_ledgerkey(Bucket, null, Tag), StartKey = leveled_codec:to_ledgerkey(Bucket, null, Tag),
EndKey = leveled_codec:to_ledgerkey(Bucket, null, Tag), EndKey = leveled_codec:to_ledgerkey(Bucket, null, Tag),
foldobjects(State, Tag, StartKey, EndKey, FoldHeadsFun, true). foldobjects(State, Tag, StartKey, EndKey, FoldHeadsFun,
{true, CheckPresence}, SnapPreFold).
foldobjects_byindex(State, Tag, Bucket, foldobjects_byindex(State, Tag, Bucket,
Field, FromTerm, ToTerm, FoldObjectsFun) -> Field, FromTerm, ToTerm, FoldObjectsFun) ->
@ -1005,34 +1018,53 @@ foldobjects_byindex(State, Tag, Bucket,
leveled_codec:to_ledgerkey(Bucket, null, ?IDX_TAG, Field, FromTerm), leveled_codec:to_ledgerkey(Bucket, null, ?IDX_TAG, Field, FromTerm),
EndKey = EndKey =
leveled_codec:to_ledgerkey(Bucket, null, ?IDX_TAG, Field, ToTerm), leveled_codec:to_ledgerkey(Bucket, null, ?IDX_TAG, Field, ToTerm),
foldobjects(State, Tag, StartKey, EndKey, FoldObjectsFun, false). foldobjects(State, Tag, StartKey, EndKey, FoldObjectsFun,
false, true).
-spec foldobjects(book_state(), atom(), tuple(), tuple(), fun(),
foldobjects(_State, Tag, StartKey, EndKey, FoldObjectsFun, DeferredFetch) -> false|{true, boolean()}, boolean()) ->
{FoldFun, InitAcc} = case is_tuple(FoldObjectsFun) of {async, fun()}.
true -> %% @doc
FoldObjectsFun; %% The object folder should be passed DeferredFetch and SnapPreFold.
false -> %% DeferredFetch can either be false (which will return to the fold function
{FoldObjectsFun, []} %% the full object), or {true, CheckPresence} - in which case a proxy object
end, %% will be created that if understood by the fold function will allow the fold
% For fold_objects the snapshot has been moved inside of the Folder %% function to work on the head of the object, and defer fetching the body in
% function. %% case such a fetch is unecessary.
% %% SnapPreFold determines if the snapshot of the database is done prior to
% fold_objects and fold_heads are called by the riak_kv_sweeper in Riak, %% returning the Folder function (SnapPreFold=true) or when the Folder function
% and the sweeper prompts the fold before checking to see if the fold is %% is called as Folder() {SnapPreFold=false}
% ready to be run. This may lead to the fold being called on an old foldobjects(State, Tag, StartKey, EndKey, FoldObjectsFun,
% snapshot. DeferredFetch, SnapPreFold) ->
Self = self(), {FoldFun, InitAcc} =
Folder = case is_tuple(FoldObjectsFun) of
fun() -> true ->
{ok, % FoldObjectsFun is already a tuple with a Fold function and an
LedgerSnapshot, % initial accumulator
JournalSnapshot} = book_snapshotstore(Self, Self, 5400), FoldObjectsFun;
false ->
% no initial accumulatr passed, and so should be just a list
{FoldObjectsFun, []}
end,
SnapFun =
case SnapPreFold of
true ->
{ok, LS, JS} = snapshot_store(State, store, undefined),
fun() -> {ok, LS, JS} end;
false ->
Self = self(),
% Timeout will be ignored, as will Requestor % Timeout will be ignored, as will Requestor
% %
% This uses the external snapshot - as the snpshot will need % This uses the external snapshot - as the snapshot will need
% to have consistent state between Bookie and Penciller when % to have consistent state between Bookie and Penciller when
% it is made. % it is made.
fun() -> book_snapshotstore(Self, Self, 5400) end
end,
Folder =
fun() ->
{ok, LedgerSnapshot, JournalSnapshot} = SnapFun(),
AccFun = accumulate_objects(FoldFun, AccFun = accumulate_objects(FoldFun,
JournalSnapshot, JournalSnapshot,
Tag, Tag,
@ -1099,7 +1131,7 @@ readycache_forsnapshot(LedgerCache, Query) ->
min_sqn=LedgerCache#ledger_cache.min_sqn, min_sqn=LedgerCache#ledger_cache.min_sqn,
max_sqn=LedgerCache#ledger_cache.max_sqn}; max_sqn=LedgerCache#ledger_cache.max_sqn};
_ -> _ ->
Idx = Idx =
case Query of case Query of
no_lookup -> no_lookup ->
empty_index; empty_index;
@ -1141,21 +1173,21 @@ set_options(Opts) ->
JournalSizeJitter = MaxJournalSize0 div (100 div ?JOURNAL_SIZE_JITTER), JournalSizeJitter = MaxJournalSize0 div (100 div ?JOURNAL_SIZE_JITTER),
MaxJournalSize = MaxJournalSize0 - MaxJournalSize = MaxJournalSize0 -
erlang:phash2(self()) rem JournalSizeJitter, erlang:phash2(self()) rem JournalSizeJitter,
SyncStrat = get_opt(sync_strategy, Opts, sync), SyncStrat = get_opt(sync_strategy, Opts, sync),
WRP = get_opt(waste_retention_period, Opts), WRP = get_opt(waste_retention_period, Opts),
AltStrategy = get_opt(reload_strategy, Opts, []), AltStrategy = get_opt(reload_strategy, Opts, []),
ReloadStrategy = leveled_codec:inker_reload_strategy(AltStrategy), ReloadStrategy = leveled_codec:inker_reload_strategy(AltStrategy),
PCLL0CacheSize = get_opt(max_pencillercachesize, Opts), PCLL0CacheSize = get_opt(max_pencillercachesize, Opts),
RootPath = get_opt(root_path, Opts), RootPath = get_opt(root_path, Opts),
JournalFP = RootPath ++ "/" ++ ?JOURNAL_FP, JournalFP = RootPath ++ "/" ++ ?JOURNAL_FP,
LedgerFP = RootPath ++ "/" ++ ?LEDGER_FP, LedgerFP = RootPath ++ "/" ++ ?LEDGER_FP,
ok = filelib:ensure_dir(JournalFP), ok = filelib:ensure_dir(JournalFP),
ok = filelib:ensure_dir(LedgerFP), ok = filelib:ensure_dir(LedgerFP),
{#inker_options{root_path = JournalFP, {#inker_options{root_path = JournalFP,
reload_strategy = ReloadStrategy, reload_strategy = ReloadStrategy,
max_run_length = get_opt(max_run_length, Opts), max_run_length = get_opt(max_run_length, Opts),
@ -1181,7 +1213,7 @@ startup(InkerOpts, PencillerOpts, RecentAAE) ->
fetch_head(Key, Penciller, LedgerCache) -> fetch_head(Key, Penciller, LedgerCache) ->
SW = os:timestamp(), SW = os:timestamp(),
CacheResult = CacheResult =
case LedgerCache#ledger_cache.mem of case LedgerCache#ledger_cache.mem of
undefined -> undefined ->
[]; [];
@ -1239,7 +1271,7 @@ accumulate_tree(FilterFun, JournalCheck, InkerClone, HashFun) ->
get_hashaccumulator(JournalCheck, get_hashaccumulator(JournalCheck,
InkerClone, InkerClone,
AddKeyFun). AddKeyFun).
get_hashaccumulator(JournalCheck, InkerClone, AddKeyFun) -> get_hashaccumulator(JournalCheck, InkerClone, AddKeyFun) ->
Now = leveled_codec:integer_now(), Now = leveled_codec:integer_now(),
AccFun = AccFun =
@ -1256,7 +1288,7 @@ get_hashaccumulator(JournalCheck, InkerClone, AddKeyFun) ->
false -> false ->
Acc Acc
end; end;
_ -> _ ->
AddKeyFun(B, K, H, Acc) AddKeyFun(B, K, H, Acc)
end; end;
false -> false ->
@ -1292,30 +1324,25 @@ accumulate_objects(FoldObjectsFun, InkerClone, Tag, DeferredFetch) ->
end, end,
JK = {leveled_codec:to_ledgerkey(B, K, Tag), SQN}, JK = {leveled_codec:to_ledgerkey(B, K, Tag), SQN},
case DeferredFetch of case DeferredFetch of
true -> {true, true} ->
InJournal = InJournal =
leveled_inker:ink_keycheck(InkerClone, leveled_inker:ink_keycheck(InkerClone,
LK, LK,
SQN), SQN),
case InJournal of case InJournal of
probably -> probably ->
Size = leveled_codec:get_size(LK, V), ProxyObj = make_proxy_object(LK, JK,
MDBin = MD, V,
leveled_codec:build_metadata_object(LK, InkerClone),
MD), FoldObjectsFun(B, K,ProxyObj, Acc);
Value = {proxy_object,
MDBin,
Size,
{fun fetch_value/2,
InkerClone,
JK}},
FoldObjectsFun(B,
K,
term_to_binary(Value),
Acc);
missing -> missing ->
Acc Acc
end; end;
{true, false} ->
ProxyObj = make_proxy_object(LK, JK,
MD, V,
InkerClone),
FoldObjectsFun(B, K,ProxyObj, Acc);
false -> false ->
R = fetch_value(InkerClone, JK), R = fetch_value(InkerClone, JK),
case R of case R of
@ -1323,7 +1350,7 @@ accumulate_objects(FoldObjectsFun, InkerClone, Tag, DeferredFetch) ->
Acc; Acc;
Value -> Value ->
FoldObjectsFun(B, K, Value, Acc) FoldObjectsFun(B, K, Value, Acc)
end end
end; end;
false -> false ->
@ -1332,6 +1359,13 @@ accumulate_objects(FoldObjectsFun, InkerClone, Tag, DeferredFetch) ->
end, end,
AccFun. AccFun.
make_proxy_object(LK, JK, MD, V, InkerClone) ->
Size = leveled_codec:get_size(LK, V),
MDBin = leveled_codec:build_metadata_object(LK, MD),
term_to_binary({proxy_object,
MDBin,
Size,
{fun fetch_value/2, InkerClone, JK}}).
check_presence(Key, Value, InkerClone) -> check_presence(Key, Value, InkerClone) ->
{LedgerKey, SQN} = leveled_codec:strip_to_keyseqonly({Key, Value}), {LedgerKey, SQN} = leveled_codec:strip_to_keyseqonly({Key, Value}),
@ -1455,7 +1489,7 @@ maybepush_ledgercache(MaxCacheSize, Cache, Penciller) ->
end. end.
maybe_withjitter(CacheSize, MaxCacheSize) -> maybe_withjitter(CacheSize, MaxCacheSize) ->
if if
CacheSize > MaxCacheSize -> CacheSize > MaxCacheSize ->
R = leveled_rand:uniform(7 * MaxCacheSize), R = leveled_rand:uniform(7 * MaxCacheSize),
@ -1471,7 +1505,7 @@ maybe_withjitter(CacheSize, MaxCacheSize) ->
get_loadfun(RecentAAE) -> get_loadfun(RecentAAE) ->
PrepareFun = PrepareFun =
fun(Tag, PK, SQN, Obj, VS, IdxSpecs) -> fun(Tag, PK, SQN, Obj, VS, IdxSpecs) ->
preparefor_ledgercache(Tag, PK, SQN, Obj, VS, IdxSpecs, RecentAAE) preparefor_ledgercache(Tag, PK, SQN, Obj, VS, IdxSpecs, RecentAAE)
end, end,
@ -1480,11 +1514,11 @@ get_loadfun(RecentAAE) ->
{MinSQN, MaxSQN, OutputTree} = Acc0, {MinSQN, MaxSQN, OutputTree} = Acc0,
{SQN, InkTag, PK} = KeyInJournal, {SQN, InkTag, PK} = KeyInJournal,
% VBin may already be a term % VBin may already be a term
{VBin, VSize} = ExtractFun(ValueInJournal), {VBin, VSize} = ExtractFun(ValueInJournal),
{Obj, IdxSpecs} = leveled_codec:split_inkvalue(VBin), {Obj, IdxSpecs} = leveled_codec:split_inkvalue(VBin),
case SQN of case SQN of
SQN when SQN < MinSQN -> SQN when SQN < MinSQN ->
{loop, Acc0}; {loop, Acc0};
SQN when SQN < MaxSQN -> SQN when SQN < MaxSQN ->
Chngs = PrepareFun(InkTag, PK, SQN, Obj, VSize, IdxSpecs), Chngs = PrepareFun(InkTag, PK, SQN, Obj, VSize, IdxSpecs),
{loop, {loop,
@ -1549,7 +1583,7 @@ generate_multiple_objects(Count, KeyNumber, ObjL) ->
KeyNumber + 1, KeyNumber + 1,
ObjL ++ [{Key, Value, IndexSpec}]). ObjL ++ [{Key, Value, IndexSpec}]).
ttl_test() -> ttl_test() ->
RootPath = reset_filestructure(), RootPath = reset_filestructure(),
{ok, Bookie1} = book_start([{root_path, RootPath}]), {ok, Bookie1} = book_start([{root_path, RootPath}]),
@ -1569,7 +1603,7 @@ ttl_test() ->
{ok, _} = book_head(Bookie1, "Bucket", K, ?STD_TAG) {ok, _} = book_head(Bookie1, "Bucket", K, ?STD_TAG)
end, end,
ObjL1), ObjL1),
ObjL2 = generate_multiple_objects(100, 101), ObjL2 = generate_multiple_objects(100, 101),
Past = leveled_codec:integer_now() - 300, Past = leveled_codec:integer_now() - 300,
lists:foreach(fun({K, V, S}) -> ok = book_tempput(Bookie1, lists:foreach(fun({K, V, S}) -> ok = book_tempput(Bookie1,
@ -1585,7 +1619,7 @@ ttl_test() ->
not_found = book_head(Bookie1, "Bucket", K, ?STD_TAG) not_found = book_head(Bookie1, "Bucket", K, ?STD_TAG)
end, end,
ObjL2), ObjL2),
{async, BucketFolder} = book_returnfolder(Bookie1, {async, BucketFolder} = book_returnfolder(Bookie1,
{bucket_stats, "Bucket"}), {bucket_stats, "Bucket"}),
{_Size, Count} = BucketFolder(), {_Size, Count} = BucketFolder(),
@ -1600,7 +1634,7 @@ ttl_test() ->
{false, undefined}}), {false, undefined}}),
KeyList = IndexFolder(), KeyList = IndexFolder(),
?assertMatch(20, length(KeyList)), ?assertMatch(20, length(KeyList)),
{ok, Regex} = re:compile("f8"), {ok, Regex} = re:compile("f8"),
{async, {async,
IndexFolderTR} = book_returnfolder(Bookie1, IndexFolderTR} = book_returnfolder(Bookie1,
@ -1611,10 +1645,10 @@ ttl_test() ->
{true, Regex}}), {true, Regex}}),
TermKeyList = IndexFolderTR(), TermKeyList = IndexFolderTR(),
?assertMatch(10, length(TermKeyList)), ?assertMatch(10, length(TermKeyList)),
ok = book_close(Bookie1), ok = book_close(Bookie1),
{ok, Bookie2} = book_start([{root_path, RootPath}]), {ok, Bookie2} = book_start([{root_path, RootPath}]),
{async, {async,
IndexFolderTR2} = book_returnfolder(Bookie2, IndexFolderTR2} = book_returnfolder(Bookie2,
{index_query, {index_query,
@ -1624,7 +1658,7 @@ ttl_test() ->
{false, Regex}}), {false, Regex}}),
KeyList2 = IndexFolderTR2(), KeyList2 = IndexFolderTR2(),
?assertMatch(10, length(KeyList2)), ?assertMatch(10, length(KeyList2)),
lists:foreach(fun({K, _V, _S}) -> lists:foreach(fun({K, _V, _S}) ->
not_found = book_get(Bookie2, "Bucket", K, ?STD_TAG) not_found = book_get(Bookie2, "Bucket", K, ?STD_TAG)
end, end,
@ -1633,7 +1667,7 @@ ttl_test() ->
not_found = book_head(Bookie2, "Bucket", K, ?STD_TAG) not_found = book_head(Bookie2, "Bucket", K, ?STD_TAG)
end, end,
ObjL2), ObjL2),
ok = book_close(Bookie2), ok = book_close(Bookie2),
reset_filestructure(). reset_filestructure().
@ -1729,7 +1763,7 @@ foldobjects_vs_hashtree_test() ->
?STD_TAG, ?STD_TAG,
false}), false}),
KeyHashList1 = lists:usort(HTFolder1()), KeyHashList1 = lists:usort(HTFolder1()),
FoldObjectsFun = fun(B, K, V, Acc) -> FoldObjectsFun = fun(B, K, V, Acc) ->
[{B, K, erlang:phash2(term_to_binary(V))}|Acc] end, [{B, K, erlang:phash2(term_to_binary(V))}|Acc] end,
{async, HTFolder2} = {async, HTFolder2} =
@ -1737,7 +1771,7 @@ foldobjects_vs_hashtree_test() ->
{foldobjects_allkeys, ?STD_TAG, FoldObjectsFun}), {foldobjects_allkeys, ?STD_TAG, FoldObjectsFun}),
KeyHashList2 = HTFolder2(), KeyHashList2 = HTFolder2(),
?assertMatch(KeyHashList1, lists:usort(KeyHashList2)), ?assertMatch(KeyHashList1, lists:usort(KeyHashList2)),
FoldHeadsFun = FoldHeadsFun =
fun(B, K, ProxyV, Acc) -> fun(B, K, ProxyV, Acc) ->
{proxy_object, {proxy_object,
@ -1747,14 +1781,14 @@ foldobjects_vs_hashtree_test() ->
V = FetchFun(Clone, JK), V = FetchFun(Clone, JK),
[{B, K, erlang:phash2(term_to_binary(V))}|Acc] [{B, K, erlang:phash2(term_to_binary(V))}|Acc]
end, end,
{async, HTFolder3} = {async, HTFolder3} =
book_returnfolder(Bookie1, book_returnfolder(Bookie1,
{foldheads_allkeys, ?STD_TAG, FoldHeadsFun}), {foldheads_allkeys, ?STD_TAG, FoldHeadsFun}),
KeyHashList3 = HTFolder3(), KeyHashList3 = HTFolder3(),
?assertMatch(KeyHashList1, lists:usort(KeyHashList3)), ?assertMatch(KeyHashList1, lists:usort(KeyHashList3)),
FoldHeadsFun2 = FoldHeadsFun2 =
fun(B, K, ProxyV, Acc) -> fun(B, K, ProxyV, Acc) ->
{proxy_object, {proxy_object,
MD, MD,
@ -1763,13 +1797,13 @@ foldobjects_vs_hashtree_test() ->
{Hash, _Size} = MD, {Hash, _Size} = MD,
[{B, K, Hash}|Acc] [{B, K, Hash}|Acc]
end, end,
{async, HTFolder4} = {async, HTFolder4} =
book_returnfolder(Bookie1, book_returnfolder(Bookie1,
{foldheads_allkeys, ?STD_TAG, FoldHeadsFun2}), {foldheads_allkeys, ?STD_TAG, FoldHeadsFun2}),
KeyHashList4 = HTFolder4(), KeyHashList4 = HTFolder4(),
?assertMatch(KeyHashList1, lists:usort(KeyHashList4)), ?assertMatch(KeyHashList1, lists:usort(KeyHashList4)),
ok = book_close(Bookie1), ok = book_close(Bookie1),
reset_filestructure(). reset_filestructure().
@ -1793,7 +1827,7 @@ foldobjects_vs_foldheads_bybucket_test() ->
?STD_TAG, ?STD_TAG,
Future) end, Future) end,
ObjL2), ObjL2),
FoldObjectsFun = fun(B, K, V, Acc) -> FoldObjectsFun = fun(B, K, V, Acc) ->
[{B, K, erlang:phash2(term_to_binary(V))}|Acc] end, [{B, K, erlang:phash2(term_to_binary(V))}|Acc] end,
{async, HTFolder1A} = {async, HTFolder1A} =
@ -1812,7 +1846,7 @@ foldobjects_vs_foldheads_bybucket_test() ->
KeyHashList1B = HTFolder1B(), KeyHashList1B = HTFolder1B(),
?assertMatch(false, ?assertMatch(false,
lists:usort(KeyHashList1A) == lists:usort(KeyHashList1B)), lists:usort(KeyHashList1A) == lists:usort(KeyHashList1B)),
FoldHeadsFun = FoldHeadsFun =
fun(B, K, ProxyV, Acc) -> fun(B, K, ProxyV, Acc) ->
{proxy_object, {proxy_object,
@ -1822,7 +1856,7 @@ foldobjects_vs_foldheads_bybucket_test() ->
V = FetchFun(Clone, JK), V = FetchFun(Clone, JK),
[{B, K, erlang:phash2(term_to_binary(V))}|Acc] [{B, K, erlang:phash2(term_to_binary(V))}|Acc]
end, end,
{async, HTFolder2A} = {async, HTFolder2A} =
book_returnfolder(Bookie1, book_returnfolder(Bookie1,
{foldheads_bybucket, {foldheads_bybucket,
@ -1841,7 +1875,7 @@ foldobjects_vs_foldheads_bybucket_test() ->
lists:usort(KeyHashList1A) == lists:usort(KeyHashList2A)), lists:usort(KeyHashList1A) == lists:usort(KeyHashList2A)),
?assertMatch(true, ?assertMatch(true,
lists:usort(KeyHashList1B) == lists:usort(KeyHashList2B)), lists:usort(KeyHashList1B) == lists:usort(KeyHashList2B)),
ok = book_close(Bookie1), ok = book_close(Bookie1),
reset_filestructure(). reset_filestructure().
@ -1873,7 +1907,7 @@ scan_table_test() ->
<<"F1-bin">>, <<"F1-bin">>,
<<"AA2">>), <<"AA2">>),
Tab0 = ets:new(mem, [ordered_set]), Tab0 = ets:new(mem, [ordered_set]),
SK_A0 = leveled_codec:to_ledgerkey(<<"B1">>, SK_A0 = leveled_codec:to_ledgerkey(<<"B1">>,
null, null,
?IDX_TAG, ?IDX_TAG,

370
test/end_to_end/tictac_SUITE.erl
View file

@ -21,6 +21,8 @@ all() -> [
]. ].
-define(LMD_FORMAT, "~4..0w~2..0w~2..0w~2..0w~2..0w"). -define(LMD_FORMAT, "~4..0w~2..0w~2..0w~2..0w~2..0w").
-define(V1_VERS, 1).
-define(MAGIC, 53). % riak_kv -> riak_object
many_put_compare(_Config) -> many_put_compare(_Config) ->
TreeSize = small, TreeSize = small,
@ -31,7 +33,7 @@ many_put_compare(_Config) ->
RootPathB = testutil:reset_filestructure("testB"), RootPathB = testutil:reset_filestructure("testB"),
RootPathC = testutil:reset_filestructure("testC"), RootPathC = testutil:reset_filestructure("testC"),
RootPathD = testutil:reset_filestructure("testD"), RootPathD = testutil:reset_filestructure("testD"),
% Start the first database, load a test object, close it, start it again % Start the first database, load a test object, close it, start it again
StartOpts1 = [{root_path, RootPathA}, StartOpts1 = [{root_path, RootPathA},
{max_pencillercachesize, 16000}, {max_pencillercachesize, 16000},
@ -52,11 +54,11 @@ many_put_compare(_Config) ->
{sync_strategy, testutil:sync_strategy()}], {sync_strategy, testutil:sync_strategy()}],
{ok, Bookie2} = leveled_bookie:book_start(StartOpts2), {ok, Bookie2} = leveled_bookie:book_start(StartOpts2),
testutil:check_forobject(Bookie2, TestObject), testutil:check_forobject(Bookie2, TestObject),
% Generate 200K objects to be sued within the test, and load them into % Generate 200K objects to be sued within the test, and load them into
% the first store (outputting the generated objects as a list of lists) % the first store (outputting the generated objects as a list of lists)
% to be used elsewhere % to be used elsewhere
GenList = [2, 20002, 40002, 60002, 80002, GenList = [2, 20002, 40002, 60002, 80002,
100002, 120002, 140002, 160002, 180002], 100002, 120002, 140002, 160002, 180002],
CLs = testutil:load_objects(20000, CLs = testutil:load_objects(20000,
@ -65,20 +67,20 @@ many_put_compare(_Config) ->
TestObject, TestObject,
fun testutil:generate_smallobjects/2, fun testutil:generate_smallobjects/2,
20000), 20000),
% Start a new store, and load the same objects (except fot the original % Start a new store, and load the same objects (except fot the original
% test object) into this store % test object) into this store
StartOpts3 = [{root_path, RootPathB}, StartOpts3 = [{root_path, RootPathB},
{max_journalsize, 200000000}, {max_journalsize, 200000000},
{max_pencillercachesize, 16000}, {max_pencillercachesize, 16000},
{sync_strategy, testutil:sync_strategy()}], {sync_strategy, testutil:sync_strategy()}],
{ok, Bookie3} = leveled_bookie:book_start(StartOpts3), {ok, Bookie3} = leveled_bookie:book_start(StartOpts3),
lists:foreach(fun(ObjL) -> testutil:riakload(Bookie3, ObjL) end, CLs), lists:foreach(fun(ObjL) -> testutil:riakload(Bookie3, ObjL) end, CLs),
% Now run a tictac query against both stores to see th extent to which % Now run a tictac query against both stores to see the extent to which
% state between stores is consistent % state between stores is consistent
TicTacQ = {tictactree_obj, TicTacQ = {tictactree_obj,
{o_rkv, "Bucket", null, null, false}, {o_rkv, "Bucket", null, null, false},
TreeSize, TreeSize,
@ -99,15 +101,97 @@ many_put_compare(_Config) ->
[timer:now_diff(os:timestamp(), SWC0)]), [timer:now_diff(os:timestamp(), SWC0)]),
io:format("Tree comparison shows ~w different leaves~n", io:format("Tree comparison shows ~w different leaves~n",
[length(SegList0)]), [length(SegList0)]),
AltList = leveled_tictac:find_dirtyleaves(TreeA, AltList =
leveled_tictac:new_tree(0)), leveled_tictac:find_dirtyleaves(TreeA,
leveled_tictac:new_tree(0, TreeSize)),
io:format("Tree comparison shows ~w altered leaves~n", io:format("Tree comparison shows ~w altered leaves~n",
[length(AltList)]), [length(AltList)]),
true = length(SegList0) == 1, true = length(SegList0) == 1,
% only the test object should be different % only the test object should be different
true = length(AltList) > 10000, true = length(AltList) > 10000,
% check there are a significant number of differences from empty % check there are a significant number of differences from empty
% Now run the same query by putting the tree-building responsibility onto
% the fold_objects_fun
ApplyHash =
fun(HashFun) ->
fun(_Key, Value) ->
{proxy_object, HeadBin, _Size, _FetchFun} = binary_to_term(Value),
<<?MAGIC:8/integer, ?V1_VERS:8/integer, VclockLen:32/integer,
Rest/binary>> = HeadBin,
<<VclockBin:VclockLen/binary, _NotNeeded/binary>> = Rest,
HashFun(lists:sort(binary_to_term(VclockBin)))
end
end,
FoldObjectsFun =
fun(_Bucket, Key, Value, Acc) ->
leveled_tictac:add_kv(Acc, Key, Value, ApplyHash(fun erlang:phash2/1))
end,
FoldQ0 = {foldheads_bybucket,
o_rkv,
"Bucket",
{FoldObjectsFun, leveled_tictac:new_tree(0, TreeSize)},
false, true},
{async, TreeAObjFolder0} =
leveled_bookie:book_returnfolder(Bookie2, FoldQ0),
SWB0Obj = os:timestamp(),
TreeAObj0 = TreeAObjFolder0(),
io:format("Build tictac tree via object fold with no "++
"presence check and 200K objects in ~w~n",
[timer:now_diff(os:timestamp(), SWB0Obj)]),
true = length(leveled_tictac:find_dirtyleaves(TreeA, TreeAObj0)) == 0,
FoldQ1 = {foldheads_bybucket,
o_rkv,
"Bucket",
{FoldObjectsFun, leveled_tictac:new_tree(0, TreeSize)},
true, true},
{async, TreeAObjFolder1} =
leveled_bookie:book_returnfolder(Bookie2, FoldQ1),
SWB1Obj = os:timestamp(),
TreeAObj1 = TreeAObjFolder1(),
io:format("Build tictac tree via object fold with "++
"presence check and 200K objects in ~w~n",
[timer:now_diff(os:timestamp(), SWB1Obj)]),
true = length(leveled_tictac:find_dirtyleaves(TreeA, TreeAObj1)) == 0,
% AAE trees within riak are based on a sha of the vector clock. So to
% compare with an AAE tree we need to compare outputs when we're hashing
% a hash
AltHashFun =
fun(Term) ->
erlang:phash2(crypto:hash(sha, term_to_binary(Term)))
end,
AltFoldObjectsFun =
fun(_Bucket, Key, Value, Acc) ->
leveled_tictac:add_kv(Acc, Key, Value, ApplyHash(AltHashFun))
end,
AltFoldQ0 = {foldheads_bybucket,
o_rkv,
"Bucket",
{AltFoldObjectsFun, leveled_tictac:new_tree(0, TreeSize)},
false, true},
{async, TreeAAltObjFolder0} =
leveled_bookie:book_returnfolder(Bookie2, AltFoldQ0),
SWB2Obj = os:timestamp(),
TreeAAltObj = TreeAAltObjFolder0(),
io:format("Build tictac tree via object fold with no "++
"presence check and 200K objects and alt hash in ~w~n",
[timer:now_diff(os:timestamp(), SWB2Obj)]),
{async, TreeBAltObjFolder0} =
leveled_bookie:book_returnfolder(Bookie3, AltFoldQ0),
SWB3Obj = os:timestamp(),
TreeBAltObj = TreeBAltObjFolder0(),
io:format("Build tictac tree via object fold with no "++
"presence check and 200K objects and alt hash in ~w~n",
[timer:now_diff(os:timestamp(), SWB3Obj)]),
true =
length(leveled_tictac:find_dirtyleaves(TreeBAltObj, TreeAAltObj)) == 1,
%% Finding differing keys
FoldKeysFun = FoldKeysFun =
fun(SegListToFind) -> fun(SegListToFind) ->
fun(_B, K, Acc) -> fun(_B, K, Acc) ->
@ -129,14 +213,14 @@ many_put_compare(_Config) ->
[length(SegKeyList), [length(SegKeyList),
length(SegList0), length(SegList0),
timer:now_diff(os:timestamp(), SWSKL0)]), timer:now_diff(os:timestamp(), SWSKL0)]),
true = length(SegKeyList) >= 1, true = length(SegKeyList) >= 1,
true = length(SegKeyList) < 10, true = length(SegKeyList) < 10,
true = lists:member("Key1.1.4567.4321", SegKeyList), true = lists:member("Key1.1.4567.4321", SegKeyList),
% Now remove the object which represents the difference between these % Now remove the object which represents the difference between these
% stores and confirm that the tictac trees will now match % stores and confirm that the tictac trees will now match
testutil:book_riakdelete(Bookie2, B1, K1, []), testutil:book_riakdelete(Bookie2, B1, K1, []),
{async, TreeAFolder0} = leveled_bookie:book_returnfolder(Bookie2, TicTacQ), {async, TreeAFolder0} = leveled_bookie:book_returnfolder(Bookie2, TicTacQ),
SWA1 = os:timestamp(), SWA1 = os:timestamp(),
@ -149,7 +233,7 @@ many_put_compare(_Config) ->
[length(SegList1)]), [length(SegList1)]),
true = length(SegList1) == 0, true = length(SegList1) == 0,
% Removed test object so tictac trees should match % Removed test object so tictac trees should match
ok = testutil:book_riakput(Bookie3, TestObject, TestSpec), ok = testutil:book_riakput(Bookie3, TestObject, TestSpec),
{async, TreeBFolder0} = leveled_bookie:book_returnfolder(Bookie3, TicTacQ), {async, TreeBFolder0} = leveled_bookie:book_returnfolder(Bookie3, TicTacQ),
SWB1 = os:timestamp(), SWB1 = os:timestamp(),
@ -162,10 +246,10 @@ many_put_compare(_Config) ->
% Bookie 2 (compared to it being in Bookie2 not Bookie3) % Bookie 2 (compared to it being in Bookie2 not Bookie3)
ok = leveled_bookie:book_close(Bookie3), ok = leveled_bookie:book_close(Bookie3),
% Replace Bookie 3 with two stores Bookie 4 and Bookie 5 where the ojects % Replace Bookie 3 with two stores Bookie 4 and Bookie 5 where the ojects
% have been randomly split between the stores % have been randomly split between the stores
StartOpts4 = [{root_path, RootPathC}, StartOpts4 = [{root_path, RootPathC},
{max_journalsize, 200000000}, {max_journalsize, 200000000},
{max_pencillercachesize, 24000}, {max_pencillercachesize, 24000},
@ -176,7 +260,7 @@ many_put_compare(_Config) ->
{max_pencillercachesize, 24000}, {max_pencillercachesize, 24000},
{sync_strategy, testutil:sync_strategy()}], {sync_strategy, testutil:sync_strategy()}],
{ok, Bookie5} = leveled_bookie:book_start(StartOpts5), {ok, Bookie5} = leveled_bookie:book_start(StartOpts5),
SplitFun = SplitFun =
fun(Obj) -> fun(Obj) ->
case erlang:phash2(Obj) rem 2 of case erlang:phash2(Obj) rem 2 of
@ -192,11 +276,11 @@ many_put_compare(_Config) ->
testutil:riakload(Bookie5, ObjLB) testutil:riakload(Bookie5, ObjLB)
end, end,
CLs), CLs),
% query both the stores, then merge the trees - the result should be the % query both the stores, then merge the trees - the result should be the
% same as the result from the tree created aginst the store with both % same as the result from the tree created aginst the store with both
% partitions % partitions
{async, TreeC0Folder} = leveled_bookie:book_returnfolder(Bookie4, TicTacQ), {async, TreeC0Folder} = leveled_bookie:book_returnfolder(Bookie4, TicTacQ),
{async, TreeC1Folder} = leveled_bookie:book_returnfolder(Bookie5, TicTacQ), {async, TreeC1Folder} = leveled_bookie:book_returnfolder(Bookie5, TicTacQ),
SWD0 = os:timestamp(), SWD0 = os:timestamp(),
@ -207,18 +291,18 @@ many_put_compare(_Config) ->
TreeC1 = TreeC1Folder(), TreeC1 = TreeC1Folder(),
io:format("Build tictac tree with 100K objects in ~w~n", io:format("Build tictac tree with 100K objects in ~w~n",
[timer:now_diff(os:timestamp(), SWD1)]), [timer:now_diff(os:timestamp(), SWD1)]),
TreeC2 = leveled_tictac:merge_trees(TreeC0, TreeC1), TreeC2 = leveled_tictac:merge_trees(TreeC0, TreeC1),
SegList3 = leveled_tictac:find_dirtyleaves(TreeC2, TreeB), SegList3 = leveled_tictac:find_dirtyleaves(TreeC2, TreeB),
io:format("Tree comparison following delete shows ~w different leaves~n", io:format("Tree comparison following delete shows ~w different leaves~n",
[length(SegList3)]), [length(SegList3)]),
true = length(SegList3) == 0, true = length(SegList3) == 0,
ok = leveled_bookie:book_close(Bookie2), ok = leveled_bookie:book_close(Bookie2),
ok = leveled_bookie:book_close(Bookie4), ok = leveled_bookie:book_close(Bookie4),
ok = leveled_bookie:book_close(Bookie5). ok = leveled_bookie:book_close(Bookie5).
index_compare(_Config) -> index_compare(_Config) ->
TreeSize = small, TreeSize = small,
@ -226,7 +310,7 @@ index_compare(_Config) ->
JS = 50000000, JS = 50000000,
SS = testutil:sync_strategy(), SS = testutil:sync_strategy(),
SegmentCount = 256 * 256, SegmentCount = 256 * 256,
% Test requires multiple different databases, so want to mount them all % Test requires multiple different databases, so want to mount them all
% on individual file paths % on individual file paths
RootPathA = testutil:reset_filestructure("testA"), RootPathA = testutil:reset_filestructure("testA"),
@ -239,7 +323,7 @@ index_compare(_Config) ->
{ok, Book1B} = leveled_bookie:book_start(RootPathB, LS, JS, SS), {ok, Book1B} = leveled_bookie:book_start(RootPathB, LS, JS, SS),
{ok, Book1C} = leveled_bookie:book_start(RootPathC, LS, JS, SS), {ok, Book1C} = leveled_bookie:book_start(RootPathC, LS, JS, SS),
{ok, Book1D} = leveled_bookie:book_start(RootPathD, LS, JS, SS), {ok, Book1D} = leveled_bookie:book_start(RootPathD, LS, JS, SS),
% Generate nine lists of objects % Generate nine lists of objects
BucketBin = list_to_binary("Bucket"), BucketBin = list_to_binary("Bucket"),
GenMapFun = GenMapFun =
@ -248,13 +332,13 @@ index_compare(_Config) ->
Indexes = testutil:get_randomindexes_generator(8), Indexes = testutil:get_randomindexes_generator(8),
testutil:generate_objects(10000, binary_uuid, [], V, Indexes) testutil:generate_objects(10000, binary_uuid, [], V, Indexes)
end, end,
ObjLists = lists:map(GenMapFun, lists:seq(1, 9)), ObjLists = lists:map(GenMapFun, lists:seq(1, 9)),
% Load all nine lists into Book1A % Load all nine lists into Book1A
lists:foreach(fun(ObjL) -> testutil:riakload(Book1A, ObjL) end, lists:foreach(fun(ObjL) -> testutil:riakload(Book1A, ObjL) end,
ObjLists), ObjLists),
% Split nine lists across Book1B to Book1D, three object lists in each % Split nine lists across Book1B to Book1D, three object lists in each
lists:foreach(fun(ObjL) -> testutil:riakload(Book1B, ObjL) end, lists:foreach(fun(ObjL) -> testutil:riakload(Book1B, ObjL) end,
lists:sublist(ObjLists, 1, 3)), lists:sublist(ObjLists, 1, 3)),
@ -262,7 +346,7 @@ index_compare(_Config) ->
lists:sublist(ObjLists, 4, 3)), lists:sublist(ObjLists, 4, 3)),
lists:foreach(fun(ObjL) -> testutil:riakload(Book1D, ObjL) end, lists:foreach(fun(ObjL) -> testutil:riakload(Book1D, ObjL) end,
lists:sublist(ObjLists, 7, 3)), lists:sublist(ObjLists, 7, 3)),
GetTicTacTreeFun = GetTicTacTreeFun =
fun(X, Bookie) -> fun(X, Bookie) ->
SW = os:timestamp(), SW = os:timestamp(),
@ -279,18 +363,18 @@ index_compare(_Config) ->
[X, timer:now_diff(os:timestamp(), SW)]), [X, timer:now_diff(os:timestamp(), SW)]),
R R
end, end,
% Get a TicTac tree representing one of the indexes in Bucket A % Get a TicTac tree representing one of the indexes in Bucket A
TicTacTree1_Full = GetTicTacTreeFun(1, Book1A), TicTacTree1_Full = GetTicTacTreeFun(1, Book1A),
TicTacTree1_P1 = GetTicTacTreeFun(1, Book1B), TicTacTree1_P1 = GetTicTacTreeFun(1, Book1B),
TicTacTree1_P2 = GetTicTacTreeFun(1, Book1C), TicTacTree1_P2 = GetTicTacTreeFun(1, Book1C),
TicTacTree1_P3 = GetTicTacTreeFun(1, Book1D), TicTacTree1_P3 = GetTicTacTreeFun(1, Book1D),
% Merge the tree across the partitions % Merge the tree across the partitions
TicTacTree1_Joined = lists:foldl(fun leveled_tictac:merge_trees/2, TicTacTree1_Joined = lists:foldl(fun leveled_tictac:merge_trees/2,
TicTacTree1_P1, TicTacTree1_P1,
[TicTacTree1_P2, TicTacTree1_P3]), [TicTacTree1_P2, TicTacTree1_P3]),
% Go compare! Also check we're not comparing empty trees % Go compare! Also check we're not comparing empty trees
DL1_0 = leveled_tictac:find_dirtyleaves(TicTacTree1_Full, DL1_0 = leveled_tictac:find_dirtyleaves(TicTacTree1_Full,
TicTacTree1_Joined), TicTacTree1_Joined),
@ -303,7 +387,7 @@ index_compare(_Config) ->
ok = leveled_bookie:book_close(Book1B), ok = leveled_bookie:book_close(Book1B),
ok = leveled_bookie:book_close(Book1C), ok = leveled_bookie:book_close(Book1C),
ok = leveled_bookie:book_close(Book1D), ok = leveled_bookie:book_close(Book1D),
% Double chekc all is well still after a restart % Double chekc all is well still after a restart
% Book1A to get all objects % Book1A to get all objects
{ok, Book2A} = leveled_bookie:book_start(RootPathA, LS, JS, SS), {ok, Book2A} = leveled_bookie:book_start(RootPathA, LS, JS, SS),
@ -316,12 +400,12 @@ index_compare(_Config) ->
TicTacTree2_P1 = GetTicTacTreeFun(2, Book2B), TicTacTree2_P1 = GetTicTacTreeFun(2, Book2B),
TicTacTree2_P2 = GetTicTacTreeFun(2, Book2C), TicTacTree2_P2 = GetTicTacTreeFun(2, Book2C),
TicTacTree2_P3 = GetTicTacTreeFun(2, Book2D), TicTacTree2_P3 = GetTicTacTreeFun(2, Book2D),
% Merge the tree across the partitions % Merge the tree across the partitions
TicTacTree2_Joined = lists:foldl(fun leveled_tictac:merge_trees/2, TicTacTree2_Joined = lists:foldl(fun leveled_tictac:merge_trees/2,
TicTacTree2_P1, TicTacTree2_P1,
[TicTacTree2_P2, TicTacTree2_P3]), [TicTacTree2_P2, TicTacTree2_P3]),
% Go compare! Also check we're not comparing empty trees % Go compare! Also check we're not comparing empty trees
DL2_0 = leveled_tictac:find_dirtyleaves(TicTacTree2_Full, DL2_0 = leveled_tictac:find_dirtyleaves(TicTacTree2_Full,
TicTacTree2_Joined), TicTacTree2_Joined),
@ -329,7 +413,7 @@ index_compare(_Config) ->
DL2_1 = leveled_tictac:find_dirtyleaves(TicTacTree2_Full, EmptyTree), DL2_1 = leveled_tictac:find_dirtyleaves(TicTacTree2_Full, EmptyTree),
true = DL2_0 == [], true = DL2_0 == [],
true = length(DL2_1) > 100, true = length(DL2_1) > 100,
IdxSpc = {add, "idx2_bin", "zz999"}, IdxSpc = {add, "idx2_bin", "zz999"},
{TestObj, TestSpc} = testutil:generate_testobject(BucketBin, {TestObj, TestSpc} = testutil:generate_testobject(BucketBin,
term_to_binary("K9.Z"), term_to_binary("K9.Z"),
@ -338,17 +422,17 @@ index_compare(_Config) ->
[{"MDK1", "MDV1"}]), [{"MDK1", "MDV1"}]),
ok = testutil:book_riakput(Book2C, TestObj, TestSpc), ok = testutil:book_riakput(Book2C, TestObj, TestSpc),
testutil:check_forobject(Book2C, TestObj), testutil:check_forobject(Book2C, TestObj),
TicTacTree3_Full = GetTicTacTreeFun(2, Book2A), TicTacTree3_Full = GetTicTacTreeFun(2, Book2A),
TicTacTree3_P1 = GetTicTacTreeFun(2, Book2B), TicTacTree3_P1 = GetTicTacTreeFun(2, Book2B),
TicTacTree3_P2 = GetTicTacTreeFun(2, Book2C), TicTacTree3_P2 = GetTicTacTreeFun(2, Book2C),
TicTacTree3_P3 = GetTicTacTreeFun(2, Book2D), TicTacTree3_P3 = GetTicTacTreeFun(2, Book2D),
% Merge the tree across the partitions % Merge the tree across the partitions
TicTacTree3_Joined = lists:foldl(fun leveled_tictac:merge_trees/2, TicTacTree3_Joined = lists:foldl(fun leveled_tictac:merge_trees/2,
TicTacTree3_P1, TicTacTree3_P1,
[TicTacTree3_P2, TicTacTree3_P3]), [TicTacTree3_P2, TicTacTree3_P3]),
% Find all keys index, and then just the last key % Find all keys index, and then just the last key
IdxQ1 = {index_query, IdxQ1 = {index_query,
BucketBin, BucketBin,
@ -357,7 +441,7 @@ index_compare(_Config) ->
{true, undefined}}, {true, undefined}},
{async, IdxFolder1} = leveled_bookie:book_returnfolder(Book2C, IdxQ1), {async, IdxFolder1} = leveled_bookie:book_returnfolder(Book2C, IdxQ1),
true = IdxFolder1() >= 1, true = IdxFolder1() >= 1,
DL_3to2B = leveled_tictac:find_dirtyleaves(TicTacTree2_P1, DL_3to2B = leveled_tictac:find_dirtyleaves(TicTacTree2_P1,
TicTacTree3_P1), TicTacTree3_P1),
DL_3to2C = leveled_tictac:find_dirtyleaves(TicTacTree2_P2, DL_3to2C = leveled_tictac:find_dirtyleaves(TicTacTree2_P2,
@ -366,23 +450,23 @@ index_compare(_Config) ->
TicTacTree3_P3), TicTacTree3_P3),
io:format("Individual tree comparison found dirty leaves of ~w ~w ~w~n", io:format("Individual tree comparison found dirty leaves of ~w ~w ~w~n",
[DL_3to2B, DL_3to2C, DL_3to2D]), [DL_3to2B, DL_3to2C, DL_3to2D]),
true = length(DL_3to2B) == 0, true = length(DL_3to2B) == 0,
true = length(DL_3to2C) == 1, true = length(DL_3to2C) == 1,
true = length(DL_3to2D) == 0, true = length(DL_3to2D) == 0,
% Go compare! Should find a difference in one leaf % Go compare! Should find a difference in one leaf
DL3_0 = leveled_tictac:find_dirtyleaves(TicTacTree3_Full, DL3_0 = leveled_tictac:find_dirtyleaves(TicTacTree3_Full,
TicTacTree3_Joined), TicTacTree3_Joined),
io:format("Different leaves count ~w~n", [length(DL3_0)]), io:format("Different leaves count ~w~n", [length(DL3_0)]),
true = length(DL3_0) == 1, true = length(DL3_0) == 1,
% Now we want to find for the {Term, Key} pairs that make up the segment % Now we want to find for the {Term, Key} pairs that make up the segment
% diferrence (there should only be one) % diferrence (there should only be one)
% %
% We want the database to filter on segment - so this doesn't have the % We want the database to filter on segment - so this doesn't have the
% overheads of key listing % overheads of key listing
FoldKeysIndexQFun = FoldKeysIndexQFun =
fun(_Bucket, {Term, Key}, Acc) -> fun(_Bucket, {Term, Key}, Acc) ->
Seg = leveled_tictac:get_segment(Key, SegmentCount), Seg = leveled_tictac:get_segment(Key, SegmentCount),
@ -393,7 +477,7 @@ index_compare(_Config) ->
Acc Acc
end end
end, end,
MismatchQ = {index_query, MismatchQ = {index_query,
BucketBin, BucketBin,
{FoldKeysIndexQFun, []}, {FoldKeysIndexQFun, []},
@ -403,21 +487,21 @@ index_compare(_Config) ->
{async, MMFldr_2B} = leveled_bookie:book_returnfolder(Book2B, MismatchQ), {async, MMFldr_2B} = leveled_bookie:book_returnfolder(Book2B, MismatchQ),
{async, MMFldr_2C} = leveled_bookie:book_returnfolder(Book2C, MismatchQ), {async, MMFldr_2C} = leveled_bookie:book_returnfolder(Book2C, MismatchQ),
{async, MMFldr_2D} = leveled_bookie:book_returnfolder(Book2D, MismatchQ), {async, MMFldr_2D} = leveled_bookie:book_returnfolder(Book2D, MismatchQ),
SWSS = os:timestamp(), SWSS = os:timestamp(),
SL_Joined = MMFldr_2B() ++ MMFldr_2C() ++ MMFldr_2D(), SL_Joined = MMFldr_2B() ++ MMFldr_2C() ++ MMFldr_2D(),
SL_Full = MMFldr_2A(), SL_Full = MMFldr_2A(),
io:format("Segment search across both clusters took ~w~n", io:format("Segment search across both clusters took ~w~n",
[timer:now_diff(os:timestamp(), SWSS)]), [timer:now_diff(os:timestamp(), SWSS)]),
io:format("Joined SegList ~w~n", [SL_Joined]), io:format("Joined SegList ~w~n", [SL_Joined]),
io:format("Full SegList ~w~n", [SL_Full]), io:format("Full SegList ~w~n", [SL_Full]),
Diffs = lists:subtract(SL_Full, SL_Joined) Diffs = lists:subtract(SL_Full, SL_Joined)
++ lists:subtract(SL_Joined, SL_Full), ++ lists:subtract(SL_Joined, SL_Full),
io:format("Differences between lists ~w~n", [Diffs]), io:format("Differences between lists ~w~n", [Diffs]),
% The actual difference is discovered % The actual difference is discovered
true = lists:member({"zz999", term_to_binary("K9.Z")}, Diffs), true = lists:member({"zz999", term_to_binary("K9.Z")}, Diffs),
% Without discovering too many others % Without discovering too many others
@ -433,11 +517,11 @@ index_compare(_Config) ->
recent_aae_noaae(_Config) -> recent_aae_noaae(_Config) ->
% Starts databases with recent_aae tables, and attempt to query to fetch % Starts databases with recent_aae tables, and attempt to query to fetch
% recent aae trees returns empty trees as no index entries are found. % recent aae trees returns empty trees as no index entries are found.
TreeSize = small, TreeSize = small,
% SegmentCount = 256 * 256, % SegmentCount = 256 * 256,
UnitMins = 2, UnitMins = 2,
% Test requires multiple different databases, so want to mount them all % Test requires multiple different databases, so want to mount them all
% on individual file paths % on individual file paths
RootPathA = testutil:reset_filestructure("testA"), RootPathA = testutil:reset_filestructure("testA"),
@ -448,28 +532,28 @@ recent_aae_noaae(_Config) ->
StartOptsB = aae_startopts(RootPathB, false), StartOptsB = aae_startopts(RootPathB, false),
StartOptsC = aae_startopts(RootPathC, false), StartOptsC = aae_startopts(RootPathC, false),
StartOptsD = aae_startopts(RootPathD, false), StartOptsD = aae_startopts(RootPathD, false),
% Book1A to get all objects % Book1A to get all objects
{ok, Book1A} = leveled_bookie:book_start(StartOptsA), {ok, Book1A} = leveled_bookie:book_start(StartOptsA),
% Book1B/C/D will have objects partitioned across it % Book1B/C/D will have objects partitioned across it
{ok, Book1B} = leveled_bookie:book_start(StartOptsB), {ok, Book1B} = leveled_bookie:book_start(StartOptsB),
{ok, Book1C} = leveled_bookie:book_start(StartOptsC), {ok, Book1C} = leveled_bookie:book_start(StartOptsC),
{ok, Book1D} = leveled_bookie:book_start(StartOptsD), {ok, Book1D} = leveled_bookie:book_start(StartOptsD),
{B1, K1, V1, S1, MD} = {"Bucket", {B1, K1, V1, S1, MD} = {"Bucket",
"Key1.1.4567.4321", "Key1.1.4567.4321",
"Value1", "Value1",
[], [],
[{"MDK1", "MDV1"}]}, [{"MDK1", "MDV1"}]},
{TestObject, TestSpec} = testutil:generate_testobject(B1, K1, V1, S1, MD), {TestObject, TestSpec} = testutil:generate_testobject(B1, K1, V1, S1, MD),
SW_StartLoad = os:timestamp(), SW_StartLoad = os:timestamp(),
ok = testutil:book_riakput(Book1A, TestObject, TestSpec), ok = testutil:book_riakput(Book1A, TestObject, TestSpec),
ok = testutil:book_riakput(Book1B, TestObject, TestSpec), ok = testutil:book_riakput(Book1B, TestObject, TestSpec),
testutil:check_forobject(Book1A, TestObject), testutil:check_forobject(Book1A, TestObject),
testutil:check_forobject(Book1B, TestObject), testutil:check_forobject(Book1B, TestObject),
{TicTacTreeJoined, TicTacTreeFull, EmptyTree, _LMDIndexes} = {TicTacTreeJoined, TicTacTreeFull, EmptyTree, _LMDIndexes} =
load_and_check_recentaae(Book1A, Book1B, Book1C, Book1D, load_and_check_recentaae(Book1A, Book1B, Book1C, Book1D,
SW_StartLoad, TreeSize, UnitMins, SW_StartLoad, TreeSize, UnitMins,
@ -477,7 +561,7 @@ recent_aae_noaae(_Config) ->
% Go compare! Also confirm we're not comparing empty trees % Go compare! Also confirm we're not comparing empty trees
DL1_0 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, DL1_0 = leveled_tictac:find_dirtyleaves(TicTacTreeFull,
TicTacTreeJoined), TicTacTreeJoined),
DL1_1 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, EmptyTree), DL1_1 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, EmptyTree),
true = DL1_0 == [], true = DL1_0 == [],
true = length(DL1_1) == 0, true = length(DL1_1) == 0,
@ -489,7 +573,7 @@ recent_aae_noaae(_Config) ->
recent_aae_allaae(_Config) -> recent_aae_allaae(_Config) ->
% Leveled is started in blacklisted mode with no buckets blacklisted. % Leveled is started in blacklisted mode with no buckets blacklisted.
% %
% A number of changes are then loaded into a store, and also partitioned % A number of changes are then loaded into a store, and also partitioned
% across a separate set of three stores. A merge tree is returned from % across a separate set of three stores. A merge tree is returned from
@ -503,12 +587,12 @@ recent_aae_allaae(_Config) ->
% The segment Id found is then used in a query to find the Keys that make % The segment Id found is then used in a query to find the Keys that make
% up that segment, and the delta discovered should be just that one key % up that segment, and the delta discovered should be just that one key
% which was known to have been changed % which was known to have been changed
TreeSize = small, TreeSize = small,
% SegmentCount = 256 * 256, % SegmentCount = 256 * 256,
UnitMins = 2, UnitMins = 2,
AAE = {blacklist, [], 60, UnitMins}, AAE = {blacklist, [], 60, UnitMins},
% Test requires multiple different databases, so want to mount them all % Test requires multiple different databases, so want to mount them all
% on individual file paths % on individual file paths
RootPathA = testutil:reset_filestructure("testA"), RootPathA = testutil:reset_filestructure("testA"),
@ -519,28 +603,28 @@ recent_aae_allaae(_Config) ->
StartOptsB = aae_startopts(RootPathB, AAE), StartOptsB = aae_startopts(RootPathB, AAE),
StartOptsC = aae_startopts(RootPathC, AAE), StartOptsC = aae_startopts(RootPathC, AAE),
StartOptsD = aae_startopts(RootPathD, AAE), StartOptsD = aae_startopts(RootPathD, AAE),
% Book1A to get all objects % Book1A to get all objects
{ok, Book1A} = leveled_bookie:book_start(StartOptsA), {ok, Book1A} = leveled_bookie:book_start(StartOptsA),
% Book1B/C/D will have objects partitioned across it % Book1B/C/D will have objects partitioned across it
{ok, Book1B} = leveled_bookie:book_start(StartOptsB), {ok, Book1B} = leveled_bookie:book_start(StartOptsB),
{ok, Book1C} = leveled_bookie:book_start(StartOptsC), {ok, Book1C} = leveled_bookie:book_start(StartOptsC),
{ok, Book1D} = leveled_bookie:book_start(StartOptsD), {ok, Book1D} = leveled_bookie:book_start(StartOptsD),
{B1, K1, V1, S1, MD} = {"Bucket", {B1, K1, V1, S1, MD} = {"Bucket",
"Key1.1.4567.4321", "Key1.1.4567.4321",
"Value1", "Value1",
[], [],
[{"MDK1", "MDV1"}]}, [{"MDK1", "MDV1"}]},
{TestObject, TestSpec} = testutil:generate_testobject(B1, K1, V1, S1, MD), {TestObject, TestSpec} = testutil:generate_testobject(B1, K1, V1, S1, MD),
SW_StartLoad = os:timestamp(), SW_StartLoad = os:timestamp(),
ok = testutil:book_riakput(Book1A, TestObject, TestSpec), ok = testutil:book_riakput(Book1A, TestObject, TestSpec),
ok = testutil:book_riakput(Book1B, TestObject, TestSpec), ok = testutil:book_riakput(Book1B, TestObject, TestSpec),
testutil:check_forobject(Book1A, TestObject), testutil:check_forobject(Book1A, TestObject),
testutil:check_forobject(Book1B, TestObject), testutil:check_forobject(Book1B, TestObject),
{TicTacTreeJoined, TicTacTreeFull, EmptyTree, LMDIndexes} = {TicTacTreeJoined, TicTacTreeFull, EmptyTree, LMDIndexes} =
load_and_check_recentaae(Book1A, Book1B, Book1C, Book1D, load_and_check_recentaae(Book1A, Book1B, Book1C, Book1D,
SW_StartLoad, TreeSize, UnitMins, SW_StartLoad, TreeSize, UnitMins,
@ -548,7 +632,7 @@ recent_aae_allaae(_Config) ->
% Go compare! Also confirm we're not comparing empty trees % Go compare! Also confirm we're not comparing empty trees
DL1_0 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, DL1_0 = leveled_tictac:find_dirtyleaves(TicTacTreeFull,
TicTacTreeJoined), TicTacTreeJoined),
DL1_1 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, EmptyTree), DL1_1 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, EmptyTree),
true = DL1_0 == [], true = DL1_0 == [],
true = length(DL1_1) > 100, true = length(DL1_1) > 100,
@ -557,14 +641,14 @@ recent_aae_allaae(_Config) ->
ok = leveled_bookie:book_close(Book1B), ok = leveled_bookie:book_close(Book1B),
ok = leveled_bookie:book_close(Book1C), ok = leveled_bookie:book_close(Book1C),
ok = leveled_bookie:book_close(Book1D), ok = leveled_bookie:book_close(Book1D),
% Book2A to get all objects % Book2A to get all objects
{ok, Book2A} = leveled_bookie:book_start(StartOptsA), {ok, Book2A} = leveled_bookie:book_start(StartOptsA),
% Book2B/C/D will have objects partitioned across it % Book2B/C/D will have objects partitioned across it
{ok, Book2B} = leveled_bookie:book_start(StartOptsB), {ok, Book2B} = leveled_bookie:book_start(StartOptsB),
{ok, Book2C} = leveled_bookie:book_start(StartOptsC), {ok, Book2C} = leveled_bookie:book_start(StartOptsC),
{ok, Book2D} = leveled_bookie:book_start(StartOptsD), {ok, Book2D} = leveled_bookie:book_start(StartOptsD),
{TicTacTreeJoined, TicTacTreeFull, EmptyTree, LMDIndexes} = {TicTacTreeJoined, TicTacTreeFull, EmptyTree, LMDIndexes} =
load_and_check_recentaae(Book2A, Book2B, Book2C, Book2D, load_and_check_recentaae(Book2A, Book2B, Book2C, Book2D,
SW_StartLoad, TreeSize, UnitMins, SW_StartLoad, TreeSize, UnitMins,
@ -572,21 +656,21 @@ recent_aae_allaae(_Config) ->
% Go compare! Also confirm we're not comparing empty trees % Go compare! Also confirm we're not comparing empty trees
DL1_0 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, DL1_0 = leveled_tictac:find_dirtyleaves(TicTacTreeFull,
TicTacTreeJoined), TicTacTreeJoined),
DL1_1 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, EmptyTree), DL1_1 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, EmptyTree),
true = DL1_0 == [], true = DL1_0 == [],
true = length(DL1_1) > 100, true = length(DL1_1) > 100,
V2 = "Value2", V2 = "Value2",
{TestObject2, TestSpec2} = {TestObject2, TestSpec2} =
testutil:generate_testobject(B1, K1, V2, S1, MD), testutil:generate_testobject(B1, K1, V2, S1, MD),
New_startTS = os:timestamp(), New_startTS = os:timestamp(),
ok = testutil:book_riakput(Book2B, TestObject2, TestSpec2), ok = testutil:book_riakput(Book2B, TestObject2, TestSpec2),
testutil:check_forobject(Book2B, TestObject2), testutil:check_forobject(Book2B, TestObject2),
testutil:check_forobject(Book2A, TestObject), testutil:check_forobject(Book2A, TestObject),
New_endTS = os:timestamp(), New_endTS = os:timestamp(),
NewLMDIndexes = determine_lmd_indexes(New_startTS, New_endTS, UnitMins), NewLMDIndexes = determine_lmd_indexes(New_startTS, New_endTS, UnitMins),
{TicTacTreeJoined2, TicTacTreeFull2, _EmptyTree, NewLMDIndexes} = {TicTacTreeJoined2, TicTacTreeFull2, _EmptyTree, NewLMDIndexes} =
@ -595,10 +679,10 @@ recent_aae_allaae(_Config) ->
NewLMDIndexes), NewLMDIndexes),
DL2_0 = leveled_tictac:find_dirtyleaves(TicTacTreeFull2, DL2_0 = leveled_tictac:find_dirtyleaves(TicTacTreeFull2,
TicTacTreeJoined2), TicTacTreeJoined2),
% DL2_1 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, EmptyTree), % DL2_1 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, EmptyTree),
true = length(DL2_0) == 1, true = length(DL2_0) == 1,
[DirtySeg] = DL2_0, [DirtySeg] = DL2_0,
TermPrefix = string:right(integer_to_list(DirtySeg), 8, $0), TermPrefix = string:right(integer_to_list(DirtySeg), 8, $0),
@ -631,18 +715,18 @@ recent_aae_allaae(_Config) ->
{KeysTerms2D, _} = lists:foldl(LMDSegFolder, {KeysTerms2D, _} = lists:foldl(LMDSegFolder,
{[], Book2D}, {[], Book2D},
lists:usort(LMDIndexes ++ NewLMDIndexes)), lists:usort(LMDIndexes ++ NewLMDIndexes)),
KeysTerms2Joined = KeysTerms2B ++ KeysTerms2C ++ KeysTerms2D, KeysTerms2Joined = KeysTerms2B ++ KeysTerms2C ++ KeysTerms2D,
DeltaX = lists:subtract(KeysTerms2A, KeysTerms2Joined), DeltaX = lists:subtract(KeysTerms2A, KeysTerms2Joined),
DeltaY = lists:subtract(KeysTerms2Joined, KeysTerms2A), DeltaY = lists:subtract(KeysTerms2Joined, KeysTerms2A),
io:format("DeltaX ~w~n", [DeltaX]), io:format("DeltaX ~w~n", [DeltaX]),
io:format("DeltaY ~w~n", [DeltaY]), io:format("DeltaY ~w~n", [DeltaY]),
true = length(DeltaX) == 0, % This hasn't seen any extra changes true = length(DeltaX) == 0, % This hasn't seen any extra changes
true = length(DeltaY) == 1, % This has seen an extra change true = length(DeltaY) == 1, % This has seen an extra change
[{_, {B1, K1}}] = DeltaY, [{_, {B1, K1}}] = DeltaY,
ok = leveled_bookie:book_close(Book2A), ok = leveled_bookie:book_close(Book2A),
ok = leveled_bookie:book_close(Book2B), ok = leveled_bookie:book_close(Book2B),
ok = leveled_bookie:book_close(Book2C), ok = leveled_bookie:book_close(Book2C),
@ -654,12 +738,12 @@ recent_aae_bucketaae(_Config) ->
% Configure AAE to work only on a single whitelisted bucket % Configure AAE to work only on a single whitelisted bucket
% Confirm that we can spot a delta in this bucket, but not % Confirm that we can spot a delta in this bucket, but not
% in another bucket % in another bucket
TreeSize = small, TreeSize = small,
% SegmentCount = 256 * 256, % SegmentCount = 256 * 256,
UnitMins = 2, UnitMins = 2,
AAE = {whitelist, [<<"Bucket">>], 60, UnitMins}, AAE = {whitelist, [<<"Bucket">>], 60, UnitMins},
% Test requires multiple different databases, so want to mount them all % Test requires multiple different databases, so want to mount them all
% on individual file paths % on individual file paths
RootPathA = testutil:reset_filestructure("testA"), RootPathA = testutil:reset_filestructure("testA"),
@ -670,28 +754,28 @@ recent_aae_bucketaae(_Config) ->
StartOptsB = aae_startopts(RootPathB, AAE), StartOptsB = aae_startopts(RootPathB, AAE),
StartOptsC = aae_startopts(RootPathC, AAE), StartOptsC = aae_startopts(RootPathC, AAE),
StartOptsD = aae_startopts(RootPathD, AAE), StartOptsD = aae_startopts(RootPathD, AAE),
% Book1A to get all objects % Book1A to get all objects
{ok, Book1A} = leveled_bookie:book_start(StartOptsA), {ok, Book1A} = leveled_bookie:book_start(StartOptsA),
% Book1B/C/D will have objects partitioned across it % Book1B/C/D will have objects partitioned across it
{ok, Book1B} = leveled_bookie:book_start(StartOptsB), {ok, Book1B} = leveled_bookie:book_start(StartOptsB),
{ok, Book1C} = leveled_bookie:book_start(StartOptsC), {ok, Book1C} = leveled_bookie:book_start(StartOptsC),
{ok, Book1D} = leveled_bookie:book_start(StartOptsD), {ok, Book1D} = leveled_bookie:book_start(StartOptsD),
{B1, K1, V1, S1, MD} = {<<"Bucket">>, {B1, K1, V1, S1, MD} = {<<"Bucket">>,
"Key1.1.4567.4321", "Key1.1.4567.4321",
"Value1", "Value1",
[], [],
[{"MDK1", "MDV1"}]}, [{"MDK1", "MDV1"}]},
{TestObject, TestSpec} = testutil:generate_testobject(B1, K1, V1, S1, MD), {TestObject, TestSpec} = testutil:generate_testobject(B1, K1, V1, S1, MD),
SW_StartLoad = os:timestamp(), SW_StartLoad = os:timestamp(),
ok = testutil:book_riakput(Book1A, TestObject, TestSpec), ok = testutil:book_riakput(Book1A, TestObject, TestSpec),
ok = testutil:book_riakput(Book1B, TestObject, TestSpec), ok = testutil:book_riakput(Book1B, TestObject, TestSpec),
testutil:check_forobject(Book1A, TestObject), testutil:check_forobject(Book1A, TestObject),
testutil:check_forobject(Book1B, TestObject), testutil:check_forobject(Book1B, TestObject),
{TicTacTreeJoined, TicTacTreeFull, EmptyTree, LMDIndexes} = {TicTacTreeJoined, TicTacTreeFull, EmptyTree, LMDIndexes} =
load_and_check_recentaae(Book1A, Book1B, Book1C, Book1D, load_and_check_recentaae(Book1A, Book1B, Book1C, Book1D,
SW_StartLoad, TreeSize, UnitMins, SW_StartLoad, TreeSize, UnitMins,
@ -699,7 +783,7 @@ recent_aae_bucketaae(_Config) ->
% Go compare! Also confirm we're not comparing empty trees % Go compare! Also confirm we're not comparing empty trees
DL1_0 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, DL1_0 = leveled_tictac:find_dirtyleaves(TicTacTreeFull,
TicTacTreeJoined), TicTacTreeJoined),
DL1_1 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, EmptyTree), DL1_1 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, EmptyTree),
true = DL1_0 == [], true = DL1_0 == [],
true = length(DL1_1) > 100, true = length(DL1_1) > 100,
@ -708,27 +792,27 @@ recent_aae_bucketaae(_Config) ->
ok = leveled_bookie:book_close(Book1B), ok = leveled_bookie:book_close(Book1B),
ok = leveled_bookie:book_close(Book1C), ok = leveled_bookie:book_close(Book1C),
ok = leveled_bookie:book_close(Book1D), ok = leveled_bookie:book_close(Book1D),
% Book2A to get all objects % Book2A to get all objects
{ok, Book2A} = leveled_bookie:book_start(StartOptsA), {ok, Book2A} = leveled_bookie:book_start(StartOptsA),
% Book2B/C/D will have objects partitioned across it % Book2B/C/D will have objects partitioned across it
{ok, Book2B} = leveled_bookie:book_start(StartOptsB), {ok, Book2B} = leveled_bookie:book_start(StartOptsB),
{ok, Book2C} = leveled_bookie:book_start(StartOptsC), {ok, Book2C} = leveled_bookie:book_start(StartOptsC),
{ok, Book2D} = leveled_bookie:book_start(StartOptsD), {ok, Book2D} = leveled_bookie:book_start(StartOptsD),
% Change the value for a key in another bucket % Change the value for a key in another bucket
% If we get trees for this period, no difference should be found % If we get trees for this period, no difference should be found
V2 = "Value2", V2 = "Value2",
{TestObject2, TestSpec2} = {TestObject2, TestSpec2} =
testutil:generate_testobject(<<"NotBucket">>, K1, V2, S1, MD), testutil:generate_testobject(<<"NotBucket">>, K1, V2, S1, MD),
New_startTS2 = os:timestamp(), New_startTS2 = os:timestamp(),
ok = testutil:book_riakput(Book2B, TestObject2, TestSpec2), ok = testutil:book_riakput(Book2B, TestObject2, TestSpec2),
testutil:check_forobject(Book2B, TestObject2), testutil:check_forobject(Book2B, TestObject2),
testutil:check_forobject(Book2A, TestObject), testutil:check_forobject(Book2A, TestObject),
New_endTS2 = os:timestamp(), New_endTS2 = os:timestamp(),
NewLMDIndexes2 = determine_lmd_indexes(New_startTS2, New_endTS2, UnitMins), NewLMDIndexes2 = determine_lmd_indexes(New_startTS2, New_endTS2, UnitMins),
{TicTacTreeJoined2, TicTacTreeFull2, _EmptyTree, NewLMDIndexes2} = {TicTacTreeJoined2, TicTacTreeFull2, _EmptyTree, NewLMDIndexes2} =
@ -738,19 +822,19 @@ recent_aae_bucketaae(_Config) ->
DL2_0 = leveled_tictac:find_dirtyleaves(TicTacTreeFull2, DL2_0 = leveled_tictac:find_dirtyleaves(TicTacTreeFull2,
TicTacTreeJoined2), TicTacTreeJoined2),
true = length(DL2_0) == 0, true = length(DL2_0) == 0,
% Now create an object that is a change to an existing key in the % Now create an object that is a change to an existing key in the
% monitored bucket. A differrence should be found % monitored bucket. A differrence should be found
{TestObject3, TestSpec3} = {TestObject3, TestSpec3} =
testutil:generate_testobject(B1, K1, V2, S1, MD), testutil:generate_testobject(B1, K1, V2, S1, MD),
New_startTS3 = os:timestamp(), New_startTS3 = os:timestamp(),
ok = testutil:book_riakput(Book2B, TestObject3, TestSpec3), ok = testutil:book_riakput(Book2B, TestObject3, TestSpec3),
testutil:check_forobject(Book2B, TestObject3), testutil:check_forobject(Book2B, TestObject3),
testutil:check_forobject(Book2A, TestObject), testutil:check_forobject(Book2A, TestObject),
New_endTS3 = os:timestamp(), New_endTS3 = os:timestamp(),
NewLMDIndexes3 = determine_lmd_indexes(New_startTS3, New_endTS3, UnitMins), NewLMDIndexes3 = determine_lmd_indexes(New_startTS3, New_endTS3, UnitMins),
{TicTacTreeJoined3, TicTacTreeFull3, _EmptyTree, NewLMDIndexes3} = {TicTacTreeJoined3, TicTacTreeFull3, _EmptyTree, NewLMDIndexes3} =
@ -759,15 +843,15 @@ recent_aae_bucketaae(_Config) ->
NewLMDIndexes3, <<"Bucket">>), NewLMDIndexes3, <<"Bucket">>),
DL3_0 = leveled_tictac:find_dirtyleaves(TicTacTreeFull3, DL3_0 = leveled_tictac:find_dirtyleaves(TicTacTreeFull3,
TicTacTreeJoined3), TicTacTreeJoined3),
% DL2_1 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, EmptyTree), % DL2_1 = leveled_tictac:find_dirtyleaves(TicTacTreeFull, EmptyTree),
true = length(DL3_0) == 1, true = length(DL3_0) == 1,
% Find the dirty segment, and use that to find the dirty key % Find the dirty segment, and use that to find the dirty key
% %
% Note that unlike when monitoring $all, fold_keys can be used as there % Note that unlike when monitoring $all, fold_keys can be used as there
% is no need to return the Bucket (as hte bucket is known) % is no need to return the Bucket (as hte bucket is known)
[DirtySeg] = DL3_0, [DirtySeg] = DL3_0,
TermPrefix = string:right(integer_to_list(DirtySeg), 8, $0), TermPrefix = string:right(integer_to_list(DirtySeg), 8, $0),
@ -800,18 +884,18 @@ recent_aae_bucketaae(_Config) ->
{KeysTerms2D, _} = lists:foldl(LMDSegFolder, {KeysTerms2D, _} = lists:foldl(LMDSegFolder,
{[], Book2D}, {[], Book2D},
lists:usort(LMDIndexes ++ NewLMDIndexes3)), lists:usort(LMDIndexes ++ NewLMDIndexes3)),
KeysTerms2Joined = KeysTerms2B ++ KeysTerms2C ++ KeysTerms2D, KeysTerms2Joined = KeysTerms2B ++ KeysTerms2C ++ KeysTerms2D,
DeltaX = lists:subtract(KeysTerms2A, KeysTerms2Joined), DeltaX = lists:subtract(KeysTerms2A, KeysTerms2Joined),
DeltaY = lists:subtract(KeysTerms2Joined, KeysTerms2A), DeltaY = lists:subtract(KeysTerms2Joined, KeysTerms2A),
io:format("DeltaX ~w~n", [DeltaX]), io:format("DeltaX ~w~n", [DeltaX]),
io:format("DeltaY ~w~n", [DeltaY]), io:format("DeltaY ~w~n", [DeltaY]),
true = length(DeltaX) == 0, % This hasn't seen any extra changes true = length(DeltaX) == 0, % This hasn't seen any extra changes
true = length(DeltaY) == 1, % This has seen an extra change true = length(DeltaY) == 1, % This has seen an extra change
[{_, K1}] = DeltaY, [{_, K1}] = DeltaY,
ok = leveled_bookie:book_close(Book2A), ok = leveled_bookie:book_close(Book2A),
ok = leveled_bookie:book_close(Book2B), ok = leveled_bookie:book_close(Book2B),
ok = leveled_bookie:book_close(Book2C), ok = leveled_bookie:book_close(Book2C),
@ -820,21 +904,21 @@ recent_aae_bucketaae(_Config) ->
recent_aae_expiry(_Config) -> recent_aae_expiry(_Config) ->
% Proof that the index entries are indeed expired % Proof that the index entries are indeed expired
TreeSize = small, TreeSize = small,
% SegmentCount = 256 * 256, % SegmentCount = 256 * 256,
UnitMins = 1, UnitMins = 1,
TotalMins = 2, TotalMins = 2,
AAE = {blacklist, [], TotalMins, UnitMins}, AAE = {blacklist, [], TotalMins, UnitMins},
% Test requires multiple different databases, so want to mount them all % Test requires multiple different databases, so want to mount them all
% on individual file paths % on individual file paths
RootPathA = testutil:reset_filestructure("testA"), RootPathA = testutil:reset_filestructure("testA"),
StartOptsA = aae_startopts(RootPathA, AAE), StartOptsA = aae_startopts(RootPathA, AAE),
% Book1A to get all objects % Book1A to get all objects
{ok, Book1A} = leveled_bookie:book_start(StartOptsA), {ok, Book1A} = leveled_bookie:book_start(StartOptsA),
GenMapFun = GenMapFun =
fun(_X) -> fun(_X) ->
V = testutil:get_compressiblevalue(), V = testutil:get_compressiblevalue(),
@ -845,13 +929,13 @@ recent_aae_expiry(_Config) ->
V, V,
Indexes) Indexes)
end, end,
ObjLists = lists:map(GenMapFun, lists:seq(1, 3)), ObjLists = lists:map(GenMapFun, lists:seq(1, 3)),
SW0 = os:timestamp(), SW0 = os:timestamp(),
% Load all nine lists into Book1A % Load all nine lists into Book1A
lists:foreach(fun(ObjL) -> testutil:riakload(Book1A, ObjL) end, lists:foreach(fun(ObjL) -> testutil:riakload(Book1A, ObjL) end,
ObjLists), ObjLists),
SW1 = os:timestamp(), SW1 = os:timestamp(),
% sleep for two minutes, so all index entries will have expired % sleep for two minutes, so all index entries will have expired
GetTicTacTreeFun = GetTicTacTreeFun =
@ -860,19 +944,19 @@ recent_aae_expiry(_Config) ->
end, end,
EmptyTree = leveled_tictac:new_tree(empty, TreeSize), EmptyTree = leveled_tictac:new_tree(empty, TreeSize),
LMDIndexes = determine_lmd_indexes(SW0, SW1, UnitMins), LMDIndexes = determine_lmd_indexes(SW0, SW1, UnitMins),
% Should get a non-empty answer to the query % Should get a non-empty answer to the query
TicTacTree1_Full = TicTacTree1_Full =
lists:foldl(GetTicTacTreeFun(Book1A), EmptyTree, LMDIndexes), lists:foldl(GetTicTacTreeFun(Book1A), EmptyTree, LMDIndexes),
DL3_0 = leveled_tictac:find_dirtyleaves(TicTacTree1_Full, EmptyTree), DL3_0 = leveled_tictac:find_dirtyleaves(TicTacTree1_Full, EmptyTree),
io:format("Dirty leaves found before expiry ~w~n", [length(DL3_0)]), io:format("Dirty leaves found before expiry ~w~n", [length(DL3_0)]),
true = length(DL3_0) > 0, true = length(DL3_0) > 0,
SecondsSinceLMD = timer:now_diff(os:timestamp(), SW0) div 1000000, SecondsSinceLMD = timer:now_diff(os:timestamp(), SW0) div 1000000,
SecondsToExpiry = (TotalMins + UnitMins) * 60, SecondsToExpiry = (TotalMins + UnitMins) * 60,
io:format("SecondsToExpiry ~w SecondsSinceLMD ~w~n", io:format("SecondsToExpiry ~w SecondsSinceLMD ~w~n",
[SecondsToExpiry, SecondsSinceLMD]), [SecondsToExpiry, SecondsSinceLMD]),
io:format("LMDIndexes ~w~n", [LMDIndexes]), io:format("LMDIndexes ~w~n", [LMDIndexes]),
@ -882,17 +966,17 @@ recent_aae_expiry(_Config) ->
false -> false ->
timer:sleep(1000) timer:sleep(1000)
end, end,
% Should now get an empty answer - all entries have expired % Should now get an empty answer - all entries have expired
TicTacTree2_Full = TicTacTree2_Full =
lists:foldl(GetTicTacTreeFun(Book1A), EmptyTree, LMDIndexes), lists:foldl(GetTicTacTreeFun(Book1A), EmptyTree, LMDIndexes),
DL4_0 = leveled_tictac:find_dirtyleaves(TicTacTree2_Full, EmptyTree), DL4_0 = leveled_tictac:find_dirtyleaves(TicTacTree2_Full, EmptyTree),
io:format("Dirty leaves found after expiry ~w~n", [length(DL4_0)]), io:format("Dirty leaves found after expiry ~w~n", [length(DL4_0)]),
timer:sleep(10000), timer:sleep(10000),
TicTacTree3_Full = TicTacTree3_Full =
lists:foldl(GetTicTacTreeFun(Book1A), EmptyTree, LMDIndexes), lists:foldl(GetTicTacTreeFun(Book1A), EmptyTree, LMDIndexes),
DL5_0 = leveled_tictac:find_dirtyleaves(TicTacTree3_Full, EmptyTree), DL5_0 = leveled_tictac:find_dirtyleaves(TicTacTree3_Full, EmptyTree),
io:format("Dirty leaves found after expiry plus 10s ~w~n", [length(DL5_0)]), io:format("Dirty leaves found after expiry plus 10s ~w~n", [length(DL5_0)]),
@ -913,7 +997,7 @@ load_and_check_recentaae(Book1A, Book1B, Book1C, Book1D,
load_and_check_recentaae(Book1A, Book1B, Book1C, Book1D, load_and_check_recentaae(Book1A, Book1B, Book1C, Book1D,
SW_StartLoad, TreeSize, UnitMins, SW_StartLoad, TreeSize, UnitMins,
LMDIndexes_Loaded, Bucket) -> LMDIndexes_Loaded, Bucket) ->
LMDIndexes = LMDIndexes =
case LMDIndexes_Loaded of case LMDIndexes_Loaded of
false -> false ->
% Generate nine lists of objects % Generate nine lists of objects
@ -928,13 +1012,13 @@ load_and_check_recentaae(Book1A, Book1B, Book1C, Book1D,
V, V,
Indexes) Indexes)
end, end,
ObjLists = lists:map(GenMapFun, lists:seq(1, 9)), ObjLists = lists:map(GenMapFun, lists:seq(1, 9)),
% Load all nine lists into Book1A % Load all nine lists into Book1A
lists:foreach(fun(ObjL) -> testutil:riakload(Book1A, ObjL) end, lists:foreach(fun(ObjL) -> testutil:riakload(Book1A, ObjL) end,
ObjLists), ObjLists),
% Split nine lists across Book1B to Book1D, three object lists % Split nine lists across Book1B to Book1D, three object lists
% in each % in each
lists:foreach(fun(ObjL) -> testutil:riakload(Book1B, ObjL) end, lists:foreach(fun(ObjL) -> testutil:riakload(Book1B, ObjL) end,
@ -943,38 +1027,38 @@ load_and_check_recentaae(Book1A, Book1B, Book1C, Book1D,
lists:sublist(ObjLists, 4, 3)), lists:sublist(ObjLists, 4, 3)),
lists:foreach(fun(ObjL) -> testutil:riakload(Book1D, ObjL) end, lists:foreach(fun(ObjL) -> testutil:riakload(Book1D, ObjL) end,
lists:sublist(ObjLists, 7, 3)), lists:sublist(ObjLists, 7, 3)),
SW_EndLoad = os:timestamp(), SW_EndLoad = os:timestamp(),
determine_lmd_indexes(SW_StartLoad, SW_EndLoad, UnitMins); determine_lmd_indexes(SW_StartLoad, SW_EndLoad, UnitMins);
_ -> _ ->
LMDIndexes_Loaded LMDIndexes_Loaded
end, end,
EmptyTree = leveled_tictac:new_tree(empty, TreeSize), EmptyTree = leveled_tictac:new_tree(empty, TreeSize),
GetTicTacTreeFun = GetTicTacTreeFun =
fun(Bookie) -> fun(Bookie) ->
get_tictactree_fun(Bookie, Bucket, TreeSize) get_tictactree_fun(Bookie, Bucket, TreeSize)
end, end,
% Get a TicTac tree representing one of the indexes in Bucket A % Get a TicTac tree representing one of the indexes in Bucket A
TicTacTree1_Full = TicTacTree1_Full =
lists:foldl(GetTicTacTreeFun(Book1A), EmptyTree, LMDIndexes), lists:foldl(GetTicTacTreeFun(Book1A), EmptyTree, LMDIndexes),
TicTacTree1_P1 = TicTacTree1_P1 =
lists:foldl(GetTicTacTreeFun(Book1B), EmptyTree, LMDIndexes), lists:foldl(GetTicTacTreeFun(Book1B), EmptyTree, LMDIndexes),
TicTacTree1_P2 = TicTacTree1_P2 =
lists:foldl(GetTicTacTreeFun(Book1C), EmptyTree, LMDIndexes), lists:foldl(GetTicTacTreeFun(Book1C), EmptyTree, LMDIndexes),
TicTacTree1_P3 = TicTacTree1_P3 =
lists:foldl(GetTicTacTreeFun(Book1D), EmptyTree, LMDIndexes), lists:foldl(GetTicTacTreeFun(Book1D), EmptyTree, LMDIndexes),
% Merge the tree across the partitions % Merge the tree across the partitions
TicTacTree1_Joined = lists:foldl(fun leveled_tictac:merge_trees/2, TicTacTree1_Joined = lists:foldl(fun leveled_tictac:merge_trees/2,
TicTacTree1_P1, TicTacTree1_P1,
[TicTacTree1_P2, TicTacTree1_P3]), [TicTacTree1_P2, TicTacTree1_P3]),
{TicTacTree1_Full, TicTacTree1_Joined, EmptyTree, LMDIndexes}. {TicTacTree1_Full, TicTacTree1_Joined, EmptyTree, LMDIndexes}.
aae_startopts(RootPath, AAE) -> aae_startopts(RootPath, AAE) ->
LS = 2000, LS = 2000,
@ -992,7 +1076,7 @@ determine_lmd_indexes(StartTS, EndTS, UnitMins) ->
EndDT = calendar:now_to_datetime(EndTS), EndDT = calendar:now_to_datetime(EndTS),
StartTimeStr = get_strtime(StartDT, UnitMins), StartTimeStr = get_strtime(StartDT, UnitMins),
EndTimeStr = get_strtime(EndDT, UnitMins), EndTimeStr = get_strtime(EndDT, UnitMins),
AddTimeFun = AddTimeFun =
fun(X, Acc) -> fun(X, Acc) ->
case lists:member(EndTimeStr, Acc) of case lists:member(EndTimeStr, Acc) of
@ -1007,10 +1091,10 @@ determine_lmd_indexes(StartTS, EndTS, UnitMins) ->
Acc ++ [get_strtime(NextDT, UnitMins)] Acc ++ [get_strtime(NextDT, UnitMins)]
end end
end, end,
lists:foldl(AddTimeFun, [StartTimeStr], lists:seq(1, 10)). lists:foldl(AddTimeFun, [StartTimeStr], lists:seq(1, 10)).
get_strtime(DateTime, UnitMins) -> get_strtime(DateTime, UnitMins) ->
{{Y, M, D}, {Hour, Minute, _Second}} = DateTime, {{Y, M, D}, {Hour, Minute, _Second}} = DateTime,
RoundMins = RoundMins =