summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorDouglas B. Rumbaugh <douglas@t480.douglasrumbaugh.com>2023-11-02 08:01:45 -0400
committerDouglas B. Rumbaugh <douglas@t480.douglasrumbaugh.com>2023-11-02 08:01:45 -0400
commit0b723322a611de83872dd83b55d2e10e8886a283 (patch)
tree3cf3f6d11f23b9743c28585627de341d8fe7848e
parent83ca486048a5053d8c75bb5041091edb1b183a85 (diff)
downloaddynamic-extension-0b723322a611de83872dd83b55d2e10e8886a283.tar.gz
started refactoring queries interface
m---------external/psudb-common0
-rw-r--r--include/framework/QueryRequirements.h17
-rw-r--r--include/framework/interface/Shard.h6
-rw-r--r--include/query/irs.h216
-rw-r--r--include/query/rangequery.h161
-rw-r--r--include/shard/MemISAM.h361
6 files changed, 400 insertions, 361 deletions
diff --git a/external/psudb-common b/external/psudb-common
-Subproject b85686b50ab767e5c06eedb975686923aa79dd3
+Subproject 7005ad856c941d8485843c53a3b08d53ccc3d98
diff --git a/include/framework/QueryRequirements.h b/include/framework/QueryRequirements.h
new file mode 100644
index 0000000..ff4eaff
--- /dev/null
+++ b/include/framework/QueryRequirements.h
@@ -0,0 +1,17 @@
+/*
+ * include/framework/QueryRequirements.h
+ *
+ * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu>
+ *
+ * All rights reserved. Published under the Modified BSD License.
+ *
+ * A header file containing the necessary includes for Shard
+ * development.
+ *
+ */
+#pragma once
+
+#include "framework/structure/MutableBuffer.h"
+#include "framework/interface/Record.h"
+#include "framework/interface/Shard.h"
+#include "framework/interface/Query.h"
diff --git a/include/framework/interface/Shard.h b/include/framework/interface/Shard.h
index d3a6cf8..40a696b 100644
--- a/include/framework/interface/Shard.h
+++ b/include/framework/interface/Shard.h
@@ -33,4 +33,10 @@ concept ShardInterface = requires(S s, S **spp, void *p, bool b, size_t i) {
{s.get_aux_memory_usage()} -> std::convertible_to<size_t>;
};
+template <typename S, typename R>
+concept SortedShardInterface = ShardInterface<S> && requires(S s, R r, R *rp) {
+ {s.lower_bound(r)} -> std::convertible_to<size_t>;
+ {s.upper_bound(r)} -> std::convertible_to<size_t>;
+}
+
}
diff --git a/include/query/irs.h b/include/query/irs.h
new file mode 100644
index 0000000..5b09e73
--- /dev/null
+++ b/include/query/irs.h
@@ -0,0 +1,216 @@
+/*
+ * include/query/irs.h
+ *
+ * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu>
+ *
+ * All rights reserved. Published under the Modified BSD License.
+ *
+ */
+#pragma once
+
+#include "framework/QueryRequirements.h"
+
+namespace de { namespace irs {
+
+template <RecordInterface R>
+struct Parms {
+ decltype(R::key) lower_bound;
+ decltype(R::key) upper_bound;
+ size_t sample_size;
+ gsl_rng *rng;
+};
+
+
+template <RecordInterface R>
+struct State {
+ size_t lower_bound;
+ size_t upper_bound;
+ size_t sample_size;
+ size_t total_weight;
+};
+
+template <RecordInterface R>
+struct BufferState {
+ size_t cutoff;
+ std::vector<Wrapped<R>> records;
+ size_t sample_size;
+};
+
+template <ShardInterface S, RecordInterface R, bool Rejection=true>
+class Query {
+public:
+ constexpr static bool EARLY_ABORT=false;
+ constexpr static bool SKIP_DELETE_FILTER=false;
+
+ static void *get_query_state(S *shard, void *parms) {
+ auto res = new State<R>();
+ decltype(R::key) lower_key = ((PARMS<R> *) parms)->lower_bound;
+ decltype(R::key) upper_key = (PARMS<R> *) parms)->upper_bound;
+
+ res->lower_bound = shard->get_lower_bound(lower_key);
+ res->upper_bound = shard->get_upper_bound(upper_key);
+
+ if (res->lower_bound == shard->get_record_count()) {
+ res->total_weight = 0;
+ } else {
+ res->total_weight = res->upper_bound - res->lower_bound;
+ }
+
+ res->sample_size = 0;
+ return res;
+ }
+
+ static void* get_buffer_query_state(MutableBuffer<R> *buffer, void *parms) {
+ auto res = new BufferState<R>();
+
+ res->cutoff = buffer->get_record_count();
+ res->sample_size = 0;
+
+ if constexpr (Rejection) {
+ return res;
+ }
+
+ auto lower_key = ((Parms<R> *) parms)->lower_bound;
+ auto upper_key = ((Parms<R> *) parms)->upper_bound;
+
+ for (size_t i=0; i<res->cutoff; i++) {
+ if (((buffer->get_data() + i)->rec.key >= lower_key) && ((buffer->get_data() + i)->rec.key <= upper_key)) {
+ res->records.emplace_back(*(buffer->get_data() + i));
+ }
+ }
+
+ return res;
+ }
+
+ static void process_query_states(void *query_parms, std::vector<void*> &shard_states, void *buff_state) {
+ auto p = (Parms<R> *) query_parms;
+ auto bs = (buff_state) ? (BufferState<R> *) buff_state : nullptr;
+
+ std::vector<size_t> shard_sample_sizes(shard_states.size()+1, 0);
+ size_t buffer_sz = 0;
+
+ std::vector<size_t> weights;
+ if constexpr (Rejection) {
+ weights.push_back((bs) ? bs->cutoff : 0);
+ } else {
+ weights.push_back((bs) ? bs->records.size() : 0);
+ }
+
+ size_t total_weight = 0;
+ for (auto &s : shard_states) {
+ auto state = (State<R> *) s;
+ total_weight += state->total_weight;
+ weights.push_back(state->total_weight);
+ }
+
+ // if no valid records fall within the query range, just
+ // set all of the sample sizes to 0 and bail out.
+ if (total_weight == 0) {
+ for (size_t i=0; i<shard_states.size(); i++) {
+ auto state = (State<R> *) shard_states[i];
+ state->sample_size = 0;
+ }
+
+ return;
+ }
+
+ std::vector<double> normalized_weights;
+ for (auto w : weights) {
+ normalized_weights.push_back((double) w / (double) total_weight);
+ }
+
+ auto shard_alias = Alias(normalized_weights);
+ for (size_t i=0; i<p->sample_size; i++) {
+ auto idx = shard_alias.get(p->rng);
+ if (idx == 0) {
+ buffer_sz++;
+ } else {
+ shard_sample_sizes[idx - 1]++;
+ }
+ }
+
+ if (bs) {
+ bs->sample_size = buffer_sz;
+ }
+ for (size_t i=0; i<shard_states.size(); i++) {
+ auto state = (State<R> *) shard_states[i];
+ state->sample_size = shard_sample_sizes[i+1];
+ }
+ }
+
+ static std::vector<Wrapped<R>> query(S *shard, void *q_state, void *parms) {
+ auto lower_key = ((Parms<R> *) parms)->lower_bound;
+ auto upper_key = ((Parms<R> *) parms)->upper_bound;
+ auto rng = ((Parms<R> *) parms)->rng;
+
+ auto state = (State<R> *) q_state;
+ auto sample_sz = state->sample_size;
+
+ std::vector<Wrapped<R>> result_set;
+
+ if (sample_sz == 0 || state->lower_bound == shard->get_record_count()) {
+ return result_set;
+ }
+
+ size_t attempts = 0;
+ size_t range_length = state->upper_bound - state->lower_bound;
+ do {
+ attempts++;
+ size_t idx = (range_length > 0) ? gsl_rng_uniform_int(rng, range_length) : 0;
+ result_set.emplace_back(*shard->get_record_at(state->lower_bound + idx));
+ } while (attempts < sample_sz);
+
+ return result_set;
+ }
+
+ static std::vector<Wrapped<R>> buffer_query(MutableBuffer<R> *buffer, void *state, void *parms) {
+ auto st = (BufferState<R> *) state;
+ auto p = (Parms<R> *) parms;
+
+ std::vector<Wrapped<R>> result;
+ result.reserve(st->sample_size);
+
+ if constexpr (Rejection) {
+ for (size_t i=0; i<st->sample_size; i++) {
+ auto idx = gsl_rng_uniform_int(p->rng, st->cutoff);
+ auto rec = buffer->get_data() + idx;
+
+ if (rec->rec.key >= p->lower_bound && rec->rec.key <= p->upper_bound) {
+ result.emplace_back(*rec);
+ }
+ }
+
+ return result;
+ }
+
+ for (size_t i=0; i<st->sample_size; i++) {
+ auto idx = gsl_rng_uniform_int(p->rng, st->records.size());
+ result.emplace_back(st->records[idx]);
+ }
+
+ return result;
+ }
+
+ static std::vector<R> merge(std::vector<std::vector<Wrapped<R>>> &results, void *parms) {
+ std::vector<R> output;
+
+ for (size_t i=0; i<results.size(); i++) {
+ for (size_t j=0; j<results[i].size(); j++) {
+ output.emplace_back(results[i][j].rec);
+ }
+ }
+
+ return output;
+ }
+
+ static void delete_query_state(void *state) {
+ auto s = (State<R> *) state;
+ delete s;
+ }
+
+ static void delete_buffer_query_state(void *state) {
+ auto s = (BufferState<R> *) state;
+ delete s;
+ }
+};
+}}
diff --git a/include/query/rangequery.h b/include/query/rangequery.h
new file mode 100644
index 0000000..f9a34d9
--- /dev/null
+++ b/include/query/rangequery.h
@@ -0,0 +1,161 @@
+/*
+ * include/query/rangequery.h
+ *
+ * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu>
+ *
+ * All rights reserved. Published under the Modified BSD License.
+ *
+ */
+#pragma once
+
+namespace de { namespace rq {
+
+template <RecordInterface R>
+struct Parms {
+ decltype(R::key) lower_bound;
+ decltype(R::key) upper_bound;
+};
+
+template <RecordInterface R>
+struct State {
+ size_t start_idx;
+ size_t stop_idx;
+};
+
+template <RecordInterface R>
+struct BufferState {
+ size_t cutoff;
+};
+
+template <RecordInterface R>
+class Query {
+public:
+ constexpr static bool EARLY_ABORT=false;
+ constexpr static bool SKIP_DELETE_FILTER=true;
+
+ static void *get_query_state(S *shard, void *parms) {
+ auto res = new State<R>();
+ auto p = (Parms<R> *) parms;
+
+ res->start_idx = shard->get_lower_bound(p->lower_bound);
+ res->stop_idx = shard->get_record_count();
+
+ return res;
+ }
+
+ static void* get_buffer_query_state(MutableBuffer<R> *buffer, void *parms) {
+ auto res = new BufferState<R>();
+ res->cutoff = buffer->get_record_count();
+
+ return res;
+ }
+
+ static void process_query_states(void *query_parms, std::vector<void*> &shard_states, std::vector<void*> &buffer_states) {
+ return;
+ }
+
+ static std::vector<Wrapped<R>> query(S *shard, void *q_state, void *parms) {
+ std::vector<Wrapped<R>> records;
+ auto p = (Parms<R> *) parms;
+ auto s = (State<R> *) q_state;
+
+ // if the returned index is one past the end of the
+ // records for the PGM, then there are not records
+ // in the index falling into the specified range.
+ if (s->start_idx == shard->get_record_count()) {
+ return records;
+ }
+
+ auto ptr = shard->get_record_at(s->start_idx);
+
+ // roll the pointer forward to the first record that is
+ // greater than or equal to the lower bound.
+ while(ptr->rec.key < p->lower_bound) {
+ ptr++;
+ }
+
+ while (ptr->rec.key <= p->upper_bound && ptr < shard->m_data + s->stop_idx) {
+ records.emplace_back(*ptr);
+ ptr++;
+ }
+
+ return records;
+ }
+
+ static std::vector<Wrapped<R>> buffer_query(MutableBuffer<R> *buffer, void *state, void *parms) {
+ auto p = (Parms<R> *) parms;
+ auto s = (BufferState<R> *) state;
+
+ std::vector<Wrapped<R>> records;
+ for (size_t i=0; i<s->cutoff; i++) {
+ auto rec = buffer->get_data() + i;
+ if (rec->rec.key >= p->lower_bound && rec->rec.key <= p->upper_bound) {
+ records.emplace_back(*rec);
+ }
+ }
+
+ return records;
+ }
+
+ static std::vector<R> merge(std::vector<std::vector<Wrapped<R>>> &results, void *parms) {
+ std::vector<Cursor<Wrapped<R>>> cursors;
+ cursors.reserve(results.size());
+
+ PriorityQueue<Wrapped<R>> pq(results.size());
+ size_t total = 0;
+ size_t tmp_n = results.size();
+
+
+ for (size_t i = 0; i < tmp_n; ++i)
+ if (results[i].size() > 0){
+ auto base = results[i].data();
+ cursors.emplace_back(Cursor{base, base + results[i].size(), 0, results[i].size()});
+ assert(i == cursors.size() - 1);
+ total += results[i].size();
+ pq.push(cursors[i].ptr, tmp_n - i - 1);
+ } else {
+ cursors.emplace_back(Cursor<Wrapped<R>>{nullptr, nullptr, 0, 0});
+ }
+
+ if (total == 0) {
+ return std::vector<R>();
+ }
+
+ std::vector<R> output;
+ output.reserve(total);
+
+ while (pq.size()) {
+ auto now = pq.peek();
+ auto next = pq.size() > 1 ? pq.peek(1) : queue_record<Wrapped<R>>{nullptr, 0};
+ if (!now.data->is_tombstone() && next.data != nullptr &&
+ now.data->rec == next.data->rec && next.data->is_tombstone()) {
+
+ pq.pop(); pq.pop();
+ auto& cursor1 = cursors[tmp_n - now.version - 1];
+ auto& cursor2 = cursors[tmp_n - next.version - 1];
+ if (advance_cursor<Wrapped<R>>(cursor1)) pq.push(cursor1.ptr, now.version);
+ if (advance_cursor<Wrapped<R>>(cursor2)) pq.push(cursor2.ptr, next.version);
+ } else {
+ auto& cursor = cursors[tmp_n - now.version - 1];
+ if (!now.data->is_tombstone()) output.push_back(cursor.ptr->rec);
+ pq.pop();
+
+ if (advance_cursor<Wrapped<R>>(cursor)) pq.push(cursor.ptr, now.version);
+ }
+ }
+
+ return output;
+ }
+
+ static void delete_query_state(void *state) {
+ auto s = (State<R> *) state;
+ delete s;
+ }
+
+ static void delete_buffer_query_state(void *state) {
+ auto s = (BufferState<R> *) state;
+ delete s;
+ }
+};
+
+}}
diff --git a/include/shard/MemISAM.h b/include/shard/MemISAM.h
index 00fb467..6962c19 100644
--- a/include/shard/MemISAM.h
+++ b/include/shard/MemISAM.h
@@ -32,52 +32,6 @@ namespace de {
thread_local size_t mrun_cancelations = 0;
template <RecordInterface R>
-struct irs_query_parms {
- decltype(R::key) lower_bound;
- decltype(R::key) upper_bound;
- size_t sample_size;
- gsl_rng *rng;
-};
-
-template <RecordInterface R, bool Rejection>
-class IRSQuery;
-
-template <RecordInterface R>
-struct IRSState {
- size_t lower_bound;
- size_t upper_bound;
- size_t sample_size;
- size_t total_weight;
-};
-
-template <RecordInterface R>
-struct IRSBufferState {
- size_t cutoff;
- std::vector<Wrapped<R>> records;
- size_t sample_size;
-};
-
-template <RecordInterface R>
-struct ISAMRangeQueryParms {
- decltype(R::key) lower_bound;
- decltype(R::key) upper_bound;
-};
-
-template <RecordInterface R>
-class ISAMRangeQuery;
-
-template <RecordInterface R>
-struct ISAMRangeQueryState {
- size_t start_idx;
- size_t stop_idx;
-};
-
-template <RecordInterface R>
-struct RangeQueryBufferState {
- size_t cutoff;
-};
-
-template <RecordInterface R>
class MemISAM {
private:
friend class IRSQuery<R, true>;
@@ -384,319 +338,4 @@ private:
size_t m_deleted_cnt;
size_t m_alloc_size;
};
-
-template <RecordInterface R, bool Rejection=true>
-class IRSQuery {
-public:
-
- constexpr static bool EARLY_ABORT=false;
- constexpr static bool SKIP_DELETE_FILTER=false;
-
- static void *get_query_state(MemISAM<R> *isam, void *parms) {
- auto res = new IRSState<R>();
- decltype(R::key) lower_key = ((irs_query_parms<R> *) parms)->lower_bound;
- decltype(R::key) upper_key = ((irs_query_parms<R> *) parms)->upper_bound;
-
- res->lower_bound = isam->get_lower_bound(lower_key);
- res->upper_bound = isam->get_upper_bound(upper_key);
-
- if (res->lower_bound == isam->get_record_count()) {
- res->total_weight = 0;
- } else {
- res->total_weight = res->upper_bound - res->lower_bound;
- }
-
- res->sample_size = 0;
- return res;
- }
-
- static void* get_buffer_query_state(MutableBuffer<R> *buffer, void *parms) {
- auto res = new IRSBufferState<R>();
-
- res->cutoff = buffer->get_record_count();
- res->sample_size = 0;
-
- if constexpr (Rejection) {
- return res;
- }
-
- auto lower_key = ((irs_query_parms<R> *) parms)->lower_bound;
- auto upper_key = ((irs_query_parms<R> *) parms)->upper_bound;
-
- for (size_t i=0; i<res->cutoff; i++) {
- if (((buffer->get_data() + i)->rec.key >= lower_key) && ((buffer->get_data() + i)->rec.key <= upper_key)) {
- res->records.emplace_back(*(buffer->get_data() + i));
- }
- }
-
- return res;
- }
-
- static void process_query_states(void *query_parms, std::vector<void*> &shard_states, void *buff_state) {
- auto p = (irs_query_parms<R> *) query_parms;
- auto bs = (buff_state) ? (IRSBufferState<R> *) buff_state : nullptr;
-
- std::vector<size_t> shard_sample_sizes(shard_states.size()+1, 0);
- size_t buffer_sz = 0;
-
- std::vector<size_t> weights;
- if constexpr (Rejection) {
- weights.push_back((bs) ? bs->cutoff : 0);
- } else {
- weights.push_back((bs) ? bs->records.size() : 0);
- }
-
- size_t total_weight = 0;
- for (auto &s : shard_states) {
- auto state = (IRSState<R> *) s;
- total_weight += state->total_weight;
- weights.push_back(state->total_weight);
- }
-
- // if no valid records fall within the query range, just
- // set all of the sample sizes to 0 and bail out.
- if (total_weight == 0) {
- for (size_t i=0; i<shard_states.size(); i++) {
- auto state = (IRSState<R> *) shard_states[i];
- state->sample_size = 0;
- }
-
- return;
- }
-
- std::vector<double> normalized_weights;
- for (auto w : weights) {
- normalized_weights.push_back((double) w / (double) total_weight);
- }
-
- auto shard_alias = Alias(normalized_weights);
- for (size_t i=0; i<p->sample_size; i++) {
- auto idx = shard_alias.get(p->rng);
- if (idx == 0) {
- buffer_sz++;
- } else {
- shard_sample_sizes[idx - 1]++;
- }
- }
-
- if (bs) {
- bs->sample_size = buffer_sz;
- }
- for (size_t i=0; i<shard_states.size(); i++) {
- auto state = (IRSState<R> *) shard_states[i];
- state->sample_size = shard_sample_sizes[i+1];
- }
- }
-
- static std::vector<Wrapped<R>> query(MemISAM<R> *isam, void *q_state, void *parms) {
- auto lower_key = ((irs_query_parms<R> *) parms)->lower_bound;
- auto upper_key = ((irs_query_parms<R> *) parms)->upper_bound;
- auto rng = ((irs_query_parms<R> *) parms)->rng;
-
- auto state = (IRSState<R> *) q_state;
- auto sample_sz = state->sample_size;
-
- std::vector<Wrapped<R>> result_set;
-
- if (sample_sz == 0 || state->lower_bound == isam->get_record_count()) {
- return result_set;
- }
-
- size_t attempts = 0;
- size_t range_length = state->upper_bound - state->lower_bound;
- do {
- attempts++;
- size_t idx = (range_length > 0) ? gsl_rng_uniform_int(rng, range_length) : 0;
- result_set.emplace_back(*isam->get_record_at(state->lower_bound + idx));
- } while (attempts < sample_sz);
-
- return result_set;
- }
-
- static std::vector<Wrapped<R>> buffer_query(MutableBuffer<R> *buffer, void *state, void *parms) {
- auto st = (IRSBufferState<R> *) state;
- auto p = (irs_query_parms<R> *) parms;
-
- std::vector<Wrapped<R>> result;
- result.reserve(st->sample_size);
-
- if constexpr (Rejection) {
- for (size_t i=0; i<st->sample_size; i++) {
- auto idx = gsl_rng_uniform_int(p->rng, st->cutoff);
- auto rec = buffer->get_data() + idx;
-
- if (rec->rec.key >= p->lower_bound && rec->rec.key <= p->upper_bound) {
- result.emplace_back(*rec);
- }
- }
-
- return result;
- }
-
- for (size_t i=0; i<st->sample_size; i++) {
- auto idx = gsl_rng_uniform_int(p->rng, st->records.size());
- result.emplace_back(st->records[idx]);
- }
-
- return result;
- }
-
- static std::vector<R> merge(std::vector<std::vector<Wrapped<R>>> &results, void *parms) {
- std::vector<R> output;
-
- for (size_t i=0; i<results.size(); i++) {
- for (size_t j=0; j<results[i].size(); j++) {
- output.emplace_back(results[i][j].rec);
- }
- }
-
- return output;
- }
-
- static void delete_query_state(void *state) {
- auto s = (IRSState<R> *) state;
- delete s;
- }
-
- static void delete_buffer_query_state(void *state) {
- auto s = (IRSBufferState<R> *) state;
- delete s;
- }
-};
-
-
-template <RecordInterface R>
-class ISAMRangeQuery {
-public:
-
- constexpr static bool EARLY_ABORT=false;
- constexpr static bool SKIP_DELETE_FILTER=true;
-
- static void *get_query_state(MemISAM<R> *ts, void *parms) {
- auto res = new ISAMRangeQueryState<R>();
- auto p = (ISAMRangeQueryParms<R> *) parms;
-
- res->start_idx = ts->get_lower_bound(p->lower_bound);
- res->stop_idx = ts->get_record_count();
-
- return res;
- }
-
- static void* get_buffer_query_state(MutableBuffer<R> *buffer, void *parms) {
- auto res = new RangeQueryBufferState<R>();
- res->cutoff = buffer->get_record_count();
-
- return res;
- }
-
- static void process_query_states(void *query_parms, std::vector<void*> &shard_states, std::vector<void*> &buffer_states) {
- return;
- }
-
- static std::vector<Wrapped<R>> query(MemISAM<R> *ts, void *q_state, void *parms) {
- std::vector<Wrapped<R>> records;
- auto p = (ISAMRangeQueryParms<R> *) parms;
- auto s = (ISAMRangeQueryState<R> *) q_state;
-
- // if the returned index is one past the end of the
- // records for the PGM, then there are not records
- // in the index falling into the specified range.
- if (s->start_idx == ts->get_record_count()) {
- return records;
- }
-
- auto ptr = ts->get_record_at(s->start_idx);
-
- // roll the pointer forward to the first record that is
- // greater than or equal to the lower bound.
- while(ptr->rec.key < p->lower_bound) {
- ptr++;
- }
-
- while (ptr->rec.key <= p->upper_bound && ptr < ts->m_data + s->stop_idx) {
- records.emplace_back(*ptr);
- ptr++;
- }
-
- return records;
- }
-
- static std::vector<Wrapped<R>> buffer_query(MutableBuffer<R> *buffer, void *state, void *parms) {
- auto p = (ISAMRangeQueryParms<R> *) parms;
- auto s = (RangeQueryBufferState<R> *) state;
-
- std::vector<Wrapped<R>> records;
- for (size_t i=0; i<s->cutoff; i++) {
- auto rec = buffer->get_data() + i;
- if (rec->rec.key >= p->lower_bound && rec->rec.key <= p->upper_bound) {
- records.emplace_back(*rec);
- }
- }
-
- return records;
- }
-
- static std::vector<R> merge(std::vector<std::vector<Wrapped<R>>> &results, void *parms) {
- std::vector<Cursor<Wrapped<R>>> cursors;
- cursors.reserve(results.size());
-
- PriorityQueue<Wrapped<R>> pq(results.size());
- size_t total = 0;
- size_t tmp_n = results.size();
-
-
- for (size_t i = 0; i < tmp_n; ++i)
- if (results[i].size() > 0){
- auto base = results[i].data();
- cursors.emplace_back(Cursor{base, base + results[i].size(), 0, results[i].size()});
- assert(i == cursors.size() - 1);
- total += results[i].size();
- pq.push(cursors[i].ptr, tmp_n - i - 1);
- } else {
- cursors.emplace_back(Cursor<Wrapped<R>>{nullptr, nullptr, 0, 0});
- }
-
- if (total == 0) {
- return std::vector<R>();
- }
-
- std::vector<R> output;
- output.reserve(total);
-
- while (pq.size()) {
- auto now = pq.peek();
- auto next = pq.size() > 1 ? pq.peek(1) : queue_record<Wrapped<R>>{nullptr, 0};
- if (!now.data->is_tombstone() && next.data != nullptr &&
- now.data->rec == next.data->rec && next.data->is_tombstone()) {
-
- pq.pop(); pq.pop();
- auto& cursor1 = cursors[tmp_n - now.version - 1];
- auto& cursor2 = cursors[tmp_n - next.version - 1];
- if (advance_cursor<Wrapped<R>>(cursor1)) pq.push(cursor1.ptr, now.version);
- if (advance_cursor<Wrapped<R>>(cursor2)) pq.push(cursor2.ptr, next.version);
- } else {
- auto& cursor = cursors[tmp_n - now.version - 1];
- if (!now.data->is_tombstone()) output.push_back(cursor.ptr->rec);
- pq.pop();
-
- if (advance_cursor<Wrapped<R>>(cursor)) pq.push(cursor.ptr, now.version);
- }
- }
-
- return output;
- }
-
- static void delete_query_state(void *state) {
- auto s = (ISAMRangeQueryState<R> *) state;
- delete s;
- }
-
- static void delete_buffer_query_state(void *state) {
- auto s = (RangeQueryBufferState<R> *) state;
- delete s;
- }
-};
-
-
-
}