From 0b723322a611de83872dd83b55d2e10e8886a283 Mon Sep 17 00:00:00 2001
From: "Douglas B. Rumbaugh" <douglas@t480.douglasrumbaugh.com>
Date: Thu, 2 Nov 2023 08:01:45 -0400
Subject: started refactoring queries interface

---
 include/query/irs.h        | 216 +++++++++++++++++++++++++++++++++++++++++++++
 include/query/rangequery.h | 161 +++++++++++++++++++++++++++++++++
 2 files changed, 377 insertions(+)
 create mode 100644 include/query/irs.h
 create mode 100644 include/query/rangequery.h

(limited to 'include/query')

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;
+    }
+};
+
+}}
-- 
cgit v1.2.3


From e02742b07540dd5a9bcbb44dae14856bf10955ed Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Mon, 6 Nov 2023 15:18:53 -0500
Subject: Refactoring progress

---
 include/query/irs.h        |   6 +-
 include/query/rangequery.h |  14 +++-
 include/query/wss.h        | 204 +++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 217 insertions(+), 7 deletions(-)
 create mode 100644 include/query/wss.h

(limited to 'include/query')

diff --git a/include/query/irs.h b/include/query/irs.h
index 5b09e73..4cb69b0 100644
--- a/include/query/irs.h
+++ b/include/query/irs.h
@@ -44,8 +44,8 @@ public:
 
     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;
+        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);
@@ -119,7 +119,7 @@ public:
             normalized_weights.push_back((double) w / (double) total_weight);
         }
 
-        auto shard_alias = Alias(normalized_weights);
+        auto shard_alias = psudb::Alias(normalized_weights);
         for (size_t i=0; i<p->sample_size; i++) {
             auto idx = shard_alias.get(p->rng);            
             if (idx == 0) {
diff --git a/include/query/rangequery.h b/include/query/rangequery.h
index f9a34d9..b9ac9db 100644
--- a/include/query/rangequery.h
+++ b/include/query/rangequery.h
@@ -8,6 +8,12 @@
  */
 #pragma once
 
+#include "framework/interface/Record.h"
+#include "framework/interface/Shard.h"
+#include "framework/structure/MutableBuffer.h"
+#include "psu-ds/PriorityQueue.h"
+#include "util/Cursor.h"
+
 namespace de { namespace rq {
 
 template <RecordInterface R>
@@ -27,7 +33,7 @@ struct BufferState {
     size_t cutoff;
 };
 
-template <RecordInterface R>
+template <ShardInterface S, RecordInterface R>
 class Query {
 public:
     constexpr static bool EARLY_ABORT=false;
@@ -74,7 +80,7 @@ public:
             ptr++;
         }
 
-        while (ptr->rec.key <= p->upper_bound && ptr < shard->m_data + s->stop_idx) {
+        while (ptr->rec.key <= p->upper_bound && ptr < shard->get_data() + s->stop_idx) {
             records.emplace_back(*ptr);
             ptr++;
         }
@@ -101,7 +107,7 @@ public:
         std::vector<Cursor<Wrapped<R>>> cursors;
         cursors.reserve(results.size());
 
-        PriorityQueue<Wrapped<R>> pq(results.size());
+        psudb::PriorityQueue<Wrapped<R>> pq(results.size());
         size_t total = 0;
 		size_t tmp_n = results.size();
         
@@ -126,7 +132,7 @@ public:
 
         while (pq.size()) {
             auto now = pq.peek();
-            auto next = pq.size() > 1 ? pq.peek(1) : queue_record<Wrapped<R>>{nullptr, 0};
+            auto next = pq.size() > 1 ? pq.peek(1) : psudb::queue_record<Wrapped<R>>{nullptr, 0};
             if (!now.data->is_tombstone() && next.data != nullptr &&
                 now.data->rec == next.data->rec && next.data->is_tombstone()) {
                 
diff --git a/include/query/wss.h b/include/query/wss.h
new file mode 100644
index 0000000..b8a5d54
--- /dev/null
+++ b/include/query/wss.h
@@ -0,0 +1,204 @@
+/*
+ * include/query/rangequery.h
+ *
+ * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu> 
+ *
+ * All rights reserved. Published under the Modified BSD License.
+ *
+ */
+#pragma once
+
+#include "framework/interface/Record.h"
+#include "framework/interface/Shard.h"
+#include "framework/structure/MutableBuffer.h"
+
+namespace de { namespace wss {
+
+template <WeightedRecordInterface R>
+struct Parms {
+    size_t sample_size;
+    gsl_rng *rng;
+};
+
+template <WeightedRecordInterface R>
+struct State {
+    decltype(R::weight) total_weight;
+    size_t sample_size;
+
+    State() {
+        total_weight = 0;
+    }
+};
+
+template <RecordInterface R>
+struct BufferState {
+    size_t cutoff;
+    size_t sample_size;
+    psudb::Alias *alias;
+    decltype(R::weight) max_weight;
+    decltype(R::weight) total_weight;
+
+    ~BufferState() {
+        delete alias;
+    }
+};
+
+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>();
+        res->total_weight = shard->get_total_weight();
+        res->sample_size = 0;
+
+        return res;
+    }
+
+    static void* get_buffer_query_state(MutableBuffer<R> *buffer, void *parms) {
+        BufferState<R> *state = new BufferState<R>();
+        auto parameters = (Parms<R>*) parms;
+        if constexpr (Rejection) {
+            state->cutoff = buffer->get_record_count() - 1;
+            state->max_weight = buffer->get_max_weight();
+            state->total_weight = buffer->get_total_weight();
+            return state;
+        }
+
+        std::vector<double> weights;
+
+        state->cutoff = buffer->get_record_count() - 1;
+        double total_weight = 0.0;
+
+        for (size_t i = 0; i <= state->cutoff; i++) {
+            auto rec = buffer->get_data() + i;
+            weights.push_back(rec->rec.weight);
+            total_weight += rec->rec.weight;
+        }
+
+        for (size_t i = 0; i < weights.size(); i++) {
+            weights[i] = weights[i] / total_weight;
+        }
+
+        state->alias = new psudb::Alias(weights);
+        state->total_weight = total_weight;
+
+        return state;
+    }
+
+    static void process_query_states(void *query_parms, std::vector<void*> &shard_states, std::vector<void*> &buffer_states) {
+        auto p = (Parms<R> *) query_parms;
+        auto bs = (BufferState<R> *) buffer_states[0];
+
+        std::vector<size_t> shard_sample_sizes(shard_states.size()+1, 0);
+        size_t buffer_sz = 0;
+
+        std::vector<decltype(R::weight)> weights;
+        weights.push_back(bs->total_weight);
+
+        decltype(R::weight) total_weight = 0;
+        for (auto &s : shard_states) {
+            auto state = (State<R> *) s;
+            total_weight += state->total_weight;
+            weights.push_back(state->total_weight);
+        }
+
+        std::vector<double> normalized_weights;
+        for (auto w : weights) {
+            normalized_weights.push_back((double) w / (double) total_weight);
+        }
+
+        auto shard_alias = psudb::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]++;
+            }
+        }
+
+
+        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 rng = ((Parms<R> *) parms)->rng;
+
+        auto state = (State<R> *) q_state;
+        auto sample_size = state->sample_size;
+
+        std::vector<Wrapped<R>> result_set;
+
+        if (sample_size == 0) {
+            return result_set;
+        }
+        size_t attempts = 0;
+        do {
+            attempts++;
+            size_t idx = shard->m_alias->get(rng);
+            result_set.emplace_back(*shard->get_record_at(idx));
+        } while (attempts < sample_size);
+
+        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;
+
+                auto test = gsl_rng_uniform(p->rng) * st->max_weight;
+
+                if (test <= rec->rec.weight) {
+                    result.emplace_back(*rec);
+                }
+            }
+            return result;
+        }
+
+        for (size_t i=0; i<st->sample_size; i++) {
+            auto idx = st->alias->get(p->rng);
+            result.emplace_back(*(buffer->get_data() + 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;
+    }
+};
+
+}}
-- 
cgit v1.2.3


From a2fe4b1616a1b2318f70e842382818ee44aea9e6 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Tue, 7 Nov 2023 12:29:03 -0500
Subject: Alias shard fixes

---
 include/query/wss.h | 28 ++++++++++++++--------------
 1 file changed, 14 insertions(+), 14 deletions(-)

(limited to 'include/query')

diff --git a/include/query/wss.h b/include/query/wss.h
index b8a5d54..794485c 100644
--- a/include/query/wss.h
+++ b/include/query/wss.h
@@ -90,15 +90,19 @@ public:
 
     static void process_query_states(void *query_parms, std::vector<void*> &shard_states, std::vector<void*> &buffer_states) {
         auto p = (Parms<R> *) query_parms;
-        auto bs = (BufferState<R> *) buffer_states[0];
 
-        std::vector<size_t> shard_sample_sizes(shard_states.size()+1, 0);
+        std::vector<size_t> shard_sample_sizes(shard_states.size()+buffer_states.size(), 0);
         size_t buffer_sz = 0;
 
         std::vector<decltype(R::weight)> weights;
-        weights.push_back(bs->total_weight);
 
         decltype(R::weight) total_weight = 0;
+        for (auto &s : buffer_states) {
+            auto bs = (BufferState<R> *) s;
+            total_weight += bs->total_weight;
+            weights.push_back(bs->total_weight);
+        }
+
         for (auto &s : shard_states) {
             auto state = (State<R> *) s;
             total_weight += state->total_weight;
@@ -113,19 +117,15 @@ public:
         auto shard_alias = psudb::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++;
+
+            if (idx < buffer_states.size()) {
+                auto state = (BufferState<R> *) buffer_states[idx];
+                state->sample_size++;
             } else {
-                shard_sample_sizes[idx - 1]++;
+                auto state = (State<R> *) shard_states[idx - buffer_states.size()];
+                state->sample_size++;
             }
         }
-
-
-        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) {
@@ -142,7 +142,7 @@ public:
         size_t attempts = 0;
         do {
             attempts++;
-            size_t idx = shard->m_alias->get(rng);
+            size_t idx = shard->get_weighted_sample(rng);
             result_set.emplace_back(*shard->get_record_at(idx));
         } while (attempts < sample_size);
 
-- 
cgit v1.2.3


From 9e1c1b1b930031896851b1ed4a15152508327d73 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Tue, 7 Nov 2023 13:35:54 -0500
Subject: Converted WIRS to the new interface

---
 include/query/wirs.h | 240 +++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 240 insertions(+)
 create mode 100644 include/query/wirs.h

(limited to 'include/query')

diff --git a/include/query/wirs.h b/include/query/wirs.h
new file mode 100644
index 0000000..1113b1d
--- /dev/null
+++ b/include/query/wirs.h
@@ -0,0 +1,240 @@
+/*
+ * include/query/wirs.h
+ *
+ * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu> 
+ *
+ * All rights reserved. Published under the Modified BSD License.
+ *
+ */
+#pragma once
+
+#include "framework/interface/Record.h"
+#include "framework/interface/Shard.h"
+#include "framework/structure/MutableBuffer.h"
+#include "psu-ds/Alias.h"
+
+namespace de { namespace wirs {
+
+template <WeightedRecordInterface R>
+struct Parms {
+    decltype(R::key) lower_bound;
+    decltype(R::key) upper_bound;
+    size_t sample_size;
+    gsl_rng *rng;
+};
+
+template <WeightedRecordInterface R>
+struct State {
+    decltype(R::weight) total_weight;
+    std::vector<void*> nodes;
+    psudb::Alias* top_level_alias;
+    size_t sample_size;
+
+    State() {
+        total_weight = 0;
+        top_level_alias = nullptr;
+    }
+
+    ~State() {
+        if (top_level_alias) delete top_level_alias;
+    }
+};
+
+template <RecordInterface R>
+struct BufferState {
+    size_t cutoff;
+    psudb::Alias* alias;
+    std::vector<Wrapped<R>> records;
+    decltype(R::weight) max_weight;
+    size_t sample_size;
+    decltype(R::weight) total_weight;
+
+    ~BufferState() {
+        delete alias;
+    }
+};
+
+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;
+
+        std::vector<decltype(R::weight)> weights;
+        res->total_weight = shard->find_covering_nodes(lower_key, upper_key, res->nodes, weights);
+
+        std::vector<double> normalized_weights;
+        for (auto weight : weights) {
+            normalized_weights.emplace_back(weight / res->total_weight);
+        }
+
+        res->top_level_alias = new psudb::Alias(normalized_weights);
+        res->sample_size = 0;
+
+        return res;
+    }
+
+    static void* get_buffer_query_state(MutableBuffer<R> *buffer, void *parms) {
+        BufferState<R> *state = new BufferState<R>();
+        auto parameters = (Parms<R>*) parms;
+
+        if constexpr (Rejection) {
+            state->cutoff = buffer->get_record_count() - 1;
+            state->max_weight = buffer->get_max_weight();
+            state->total_weight = buffer->get_total_weight();
+            state->sample_size = 0;
+            return state;
+        }
+
+        std::vector<decltype(R::weight)> weights;
+
+        state->cutoff = buffer->get_record_count() - 1;
+        decltype(R::weight) total_weight = 0;
+
+        for (size_t i = 0; i <= state->cutoff; i++) {
+            auto rec = buffer->get_data() + i;
+
+            if (rec->rec.key >= parameters->lower_bound && rec->rec.key <= parameters->upper_bound && !rec->is_tombstone() && !rec->is_deleted()) {
+              weights.push_back(rec->rec.weight);
+              state->records.push_back(*rec);
+              total_weight += rec->rec.weight;
+            }
+        }
+
+        std::vector<double> normalized_weights;
+        for (size_t i = 0; i < weights.size(); i++) {
+            normalized_weights.push_back(weights[i] / total_weight);
+        }
+
+        state->total_weight = total_weight;
+        state->alias = new psudb::Alias(normalized_weights);
+        state->sample_size = 0;
+
+        return state;
+    }
+
+    static void process_query_states(void *query_parms, std::vector<void*> &shard_states, std::vector<void*> &buffer_states) {
+        auto p = (Parms<R> *) query_parms;
+
+        std::vector<size_t> shard_sample_sizes(shard_states.size()+buffer_states.size(), 0);
+        size_t buffer_sz = 0;
+
+        std::vector<decltype(R::weight)> weights;
+
+        decltype(R::weight) total_weight = 0;
+        for (auto &s : buffer_states) {
+            auto bs = (BufferState<R> *) s;
+            total_weight += bs->total_weight;
+            weights.push_back(bs->total_weight);
+        }
+
+        for (auto &s : shard_states) {
+            auto state = (State<R> *) s;
+            total_weight += state->total_weight;
+            weights.push_back(state->total_weight);
+        }
+
+        std::vector<double> normalized_weights;
+        for (auto w : weights) {
+            normalized_weights.push_back((double) w / (double) total_weight);
+        }
+
+        auto shard_alias = psudb::Alias(normalized_weights);
+        for (size_t i=0; i<p->sample_size; i++) {
+            auto idx = shard_alias.get(p->rng);            
+
+            if (idx < buffer_states.size()) {
+                auto state = (BufferState<R> *) buffer_states[idx];
+                state->sample_size++;
+            } else {
+                auto state = (State<R> *) shard_states[idx - buffer_states.size()];
+                state->sample_size++;
+            }
+        }
+    }
+
+    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_size = state->sample_size;
+
+        std::vector<Wrapped<R>> result_set;
+
+        if (sample_size == 0) {
+            return result_set;
+        }
+        size_t cnt = 0;
+        size_t attempts = 0;
+
+        for (size_t i=0; i<sample_size; i++) {
+            auto rec = shard->get_weighted_sample(lower_key, upper_key,
+                                                  state->nodes[state->top_level_alias->get(rng)],
+                                                  rng);
+            if (rec) {
+                result_set.emplace_back(*rec);
+            }
+        }
+
+        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;
+
+                auto test = gsl_rng_uniform(p->rng) * st->max_weight;
+
+                if (test <= rec->rec.weight && 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 = st->alias->get(p->rng);
+            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;
+    }
+};
+}}
-- 
cgit v1.2.3


From 357cab549c2ed33970562b84ff6f83923742343d Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Tue, 7 Nov 2023 15:34:24 -0500
Subject: Comment and License updates

---
 include/query/irs.h        | 6 +++++-
 include/query/rangequery.h | 6 ++++--
 include/query/wirs.h       | 6 +++++-
 include/query/wss.h        | 9 ++++++---
 4 files changed, 20 insertions(+), 7 deletions(-)

(limited to 'include/query')

diff --git a/include/query/irs.h b/include/query/irs.h
index 4cb69b0..fa69ea1 100644
--- a/include/query/irs.h
+++ b/include/query/irs.h
@@ -3,7 +3,11 @@
  *
  * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu> 
  *
- * All rights reserved. Published under the Modified BSD License.
+ * Distributed under the Modified BSD License.
+ *
+ * A query class for independent range sampling. This query requires 
+ * that the shard support get_lower_bound(key), get_upper_bound(key), 
+ * and get_record_at(index).
  *
  */
 #pragma once
diff --git a/include/query/rangequery.h b/include/query/rangequery.h
index b9ac9db..16dcd86 100644
--- a/include/query/rangequery.h
+++ b/include/query/rangequery.h
@@ -1,10 +1,12 @@
 /*
  * include/query/rangequery.h
  *
- * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu> 
+ * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu>
  *
- * All rights reserved. Published under the Modified BSD License.
+ * Distributed under the Modified BSD License.
  *
+ * A query class for single dimensional range queries. This query requires 
+ * that the shard support get_lower_bound(key) and get_record_at(index).
  */
 #pragma once
 
diff --git a/include/query/wirs.h b/include/query/wirs.h
index 1113b1d..9b3d2ad 100644
--- a/include/query/wirs.h
+++ b/include/query/wirs.h
@@ -3,7 +3,11 @@
  *
  * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu> 
  *
- * All rights reserved. Published under the Modified BSD License.
+ * Distributed under the Modified BSD License.
+ *
+ * A query class for weighted independent range sampling. This
+ * class is tightly coupled with include/shard/AugBTree.h, and
+ * so is probably of limited general utility.
  *
  */
 #pragma once
diff --git a/include/query/wss.h b/include/query/wss.h
index 794485c..4c8861e 100644
--- a/include/query/wss.h
+++ b/include/query/wss.h
@@ -1,10 +1,13 @@
 /*
- * include/query/rangequery.h
+ * include/query/wss.h
  *
- * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu> 
+ * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu>
  *
- * All rights reserved. Published under the Modified BSD License.
+ * Distributed under the Modified BSD License.
  *
+ * A query class for weighted set sampling. This
+ * class is tightly coupled with include/shard/Alias.h, 
+ * and so is probably of limited general utility.
  */
 #pragma once
 
-- 
cgit v1.2.3


From aac0bb661af8fae38d3ce08d6078cb4d9dfcb575 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Fri, 12 Jan 2024 14:10:11 -0500
Subject: Initial integration of new buffering scheme into framework

It isn't working right now (lotsa test failures), but we're to the
debugging phase now.
---
 include/query/rangequery.h | 34 ++++++++++++++++++++--------------
 1 file changed, 20 insertions(+), 14 deletions(-)

(limited to 'include/query')

diff --git a/include/query/rangequery.h b/include/query/rangequery.h
index 16dcd86..ad5b767 100644
--- a/include/query/rangequery.h
+++ b/include/query/rangequery.h
@@ -12,7 +12,7 @@
 
 #include "framework/interface/Record.h"
 #include "framework/interface/Shard.h"
-#include "framework/structure/MutableBuffer.h"
+#include "framework/structure/BufferView.h"
 #include "psu-ds/PriorityQueue.h"
 #include "util/Cursor.h"
 
@@ -32,7 +32,10 @@ struct State {
 
 template <RecordInterface R>
 struct BufferState {
-    size_t cutoff;
+    BufferView<R> buffer;
+
+    BufferState(BufferView<R> buffer) 
+        : buffer(std::move(buffer)) {}
 };
 
 template <ShardInterface S, RecordInterface R>
@@ -51,14 +54,13 @@ public:
         return res;
     }
 
-    static void* get_buffer_query_state(MutableBuffer<R> *buffer, void *parms) {
-        auto res = new BufferState<R>();
-        res->cutoff = buffer->get_record_count();
+    static void* get_buffer_query_state(BufferView<R> buffer, void *parms) {
+        auto res = new BufferState<R>(std::move(buffer));
 
         return res;
     }
 
-    static void process_query_states(void *query_parms, std::vector<void*> &shard_states, std::vector<void*> &buffer_states) {
+    static void process_query_states(void *query_parms, std::vector<void*> &shard_states, void* buffer_states) {
         return;
     }
 
@@ -67,17 +69,21 @@ public:
         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 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.
+        /*
+         * 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++;
         }
@@ -90,13 +96,13 @@ public:
         return records;
     }
 
-    static std::vector<Wrapped<R>> buffer_query(MutableBuffer<R> *buffer, void *state, void *parms) {
+    static std::vector<Wrapped<R>> buffer_query(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;
+        for (size_t i=0; i<s->buffer.get_record_count(); i++) {
+            auto rec = s->buffer.get(i);
             if (rec->rec.key >= p->lower_bound && rec->rec.key <= p->upper_bound) {
                 records.emplace_back(*rec);
             }
-- 
cgit v1.2.3


From 138c793b0a58577713d98c98bb140cf1d9c79bee Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Wed, 17 Jan 2024 18:22:00 -0500
Subject: Multiple concurrency bug fixes

A poorly organized commit with fixes for a variety of bugs that were
causing missing records. The core problems all appear to be fixed,
though there is an outstanding problem with tombstones not being
completely canceled. A very small number are appearing in the wrong
order during the static structure test.
---
 include/query/rangequery.h | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

(limited to 'include/query')

diff --git a/include/query/rangequery.h b/include/query/rangequery.h
index ad5b767..c44f5d7 100644
--- a/include/query/rangequery.h
+++ b/include/query/rangequery.h
@@ -84,11 +84,11 @@ public:
          * 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) {
+        while(ptr < shard->get_data() + s->stop_idx && ptr->rec.key < p->lower_bound) {
             ptr++;
         }
 
-        while (ptr->rec.key <= p->upper_bound && ptr < shard->get_data() + s->stop_idx) {
+        while (ptr < shard->get_data() + s->stop_idx && ptr->rec.key <= p->upper_bound) {
             records.emplace_back(*ptr);
             ptr++;
         }
@@ -152,6 +152,7 @@ public:
             } 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);
-- 
cgit v1.2.3


From 4ac2e14d24a1fdd3f9bf777775b16bf6a677f487 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Mon, 22 Jan 2024 10:14:05 -0500
Subject: Added RangeCount query

---
 include/query/rangecount.h | 169 +++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 169 insertions(+)
 create mode 100644 include/query/rangecount.h

(limited to 'include/query')

diff --git a/include/query/rangecount.h b/include/query/rangecount.h
new file mode 100644
index 0000000..7d88b1d
--- /dev/null
+++ b/include/query/rangecount.h
@@ -0,0 +1,169 @@
+/*
+ * include/query/rangecount.h
+ *
+ * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu>
+ *
+ * Distributed under the Modified BSD License.
+ *
+ * A query class for single dimensional range count queries. This query 
+ * requires that the shard support get_lower_bound(key) and 
+ * get_record_at(index).
+ */
+#pragma once
+
+#include "framework/interface/Record.h"
+#include "framework/interface/Shard.h"
+#include "framework/structure/BufferView.h"
+#include "psu-ds/PriorityQueue.h"
+#include "util/Cursor.h"
+
+namespace de { namespace rc {
+
+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 {
+    BufferView<R> buffer;
+
+    BufferState(BufferView<R> buffer) 
+        : buffer(std::move(buffer)) {}
+};
+
+template <ShardInterface S, KVPInterface 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(BufferView<R> buffer, void *parms) {
+        auto res = new BufferState<R>(std::move(buffer));
+
+        return res;
+    }
+
+    static void process_query_states(void *query_parms, std::vector<void*> &shard_states, 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;
+
+        size_t reccnt = 0;
+        size_t tscnt = 0;
+
+        Wrapped<R> res;
+        res.rec.key= 0; // records
+        res.rec.value = 0; // tombstones
+        records.emplace_back(res);
+
+        /* 
+         * 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 < shard->get_data() + s->stop_idx && ptr->rec.key < p->lower_bound) {
+            ptr++;
+        }
+
+        while (ptr < shard->get_data() + s->stop_idx && ptr->rec.key <= p->upper_bound) {
+            if (!ptr->is_deleted()) {
+                if (ptr->is_tombstone()) {
+                    records[0].rec.value++;
+                } else {
+                    records[0].rec.key++;
+                }
+            }
+
+            ptr++;
+        }
+
+        return records;
+    }
+
+    static std::vector<Wrapped<R>> buffer_query(void *state, void *parms) {
+        auto p = (Parms<R> *) parms;
+        auto s = (BufferState<R> *) state;
+
+        std::vector<Wrapped<R>> records;
+
+        Wrapped<R> res;
+        res.rec.key= 0; // records
+        res.rec.value = 0; // tombstones
+        records.emplace_back(res);
+
+        for (size_t i=0; i<s->buffer.get_record_count(); i++) {
+            auto rec = s->buffer.get(i);
+            if (rec->rec.key >= p->lower_bound && rec->rec.key <= p->upper_bound
+                    && !rec->is_deleted()) {
+                if (rec->is_tombstone()) {
+                    records[0].rec.value++;
+                } else {
+                    records[0].rec.key++;
+                }
+            }
+        }
+
+        return records;
+    }
+
+    static std::vector<R> merge(std::vector<std::vector<Wrapped<R>>> &results, void *parms) {
+
+        R res;
+        res.key = 0;
+        res.value = 0;
+        std::vector<R> output;
+        output.emplace_back(res);
+
+        for (size_t i=0; i<results.size(); i++) {
+            output[0].key += results[i][0].rec.key; // records
+            output[0].value += results[i][0].rec.value; // tombstones
+        }
+
+        output[0].key -= output[0].value;
+        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;
+    }
+};
+
+}}
-- 
cgit v1.2.3


From 1e226fc415d7674de0ecde51199d89e9042c6a22 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Wed, 31 Jan 2024 16:57:41 -0500
Subject: Updated insert query throughput to use IRS queries

---
 include/query/irs.h | 18 +++++++++++-------
 1 file changed, 11 insertions(+), 7 deletions(-)

(limited to 'include/query')

diff --git a/include/query/irs.h b/include/query/irs.h
index fa69ea1..7ef5069 100644
--- a/include/query/irs.h
+++ b/include/query/irs.h
@@ -13,6 +13,7 @@
 #pragma once
 
 #include "framework/QueryRequirements.h"
+#include "psu-ds/Alias.h"
 
 namespace de { namespace irs {
 
@@ -38,6 +39,9 @@ struct BufferState {
     size_t cutoff;
     std::vector<Wrapped<R>> records;
     size_t sample_size;
+    BufferView<R> buffer;
+
+    BufferState(BufferView<R> buffer) : buffer(std::move(buffer)) {}
 };
 
 template <ShardInterface S, RecordInterface R, bool Rejection=true>
@@ -64,10 +68,10 @@ public:
         return res;
     }
 
-    static void* get_buffer_query_state(MutableBuffer<R> *buffer, void *parms) {
-        auto res = new BufferState<R>();
+    static void* get_buffer_query_state(BufferView<R> buffer, void *parms) {
+        auto res = new BufferState<R>(std::move(buffer));
 
-        res->cutoff = buffer->get_record_count();
+        res->cutoff = res->buffer.get_record_count();
         res->sample_size = 0;
 
         if constexpr (Rejection) {
@@ -78,8 +82,8 @@ public:
         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));
+            if ((res->buffer.get(i)->rec.key >= lower_key) && (buffer.get(i)->rec.key <= upper_key)) { 
+                res->records.emplace_back(*(res->buffer.get(i)));
             }
         }
 
@@ -167,7 +171,7 @@ public:
         return result_set;
     }
 
-    static std::vector<Wrapped<R>> buffer_query(MutableBuffer<R> *buffer, void *state, void *parms) {
+    static std::vector<Wrapped<R>> buffer_query(void *state, void *parms) {
         auto st = (BufferState<R> *) state;
         auto p = (Parms<R> *) parms;
 
@@ -177,7 +181,7 @@ public:
         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;
+                auto rec = st->buffer.get(idx);
 
                 if (rec->rec.key >= p->lower_bound && rec->rec.key <= p->upper_bound) {
                     result.emplace_back(*rec);
-- 
cgit v1.2.3


From db4806d9dd9757273a14e6c3ea92e5a087239145 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Mon, 5 Feb 2024 15:17:25 -0500
Subject: Set up tombstone deletes properly

---
 include/query/irs.h        | 16 ++++++++--------
 include/query/rangecount.h | 14 +++++++-------
 include/query/rangequery.h | 14 +++++++-------
 3 files changed, 22 insertions(+), 22 deletions(-)

(limited to 'include/query')

diff --git a/include/query/irs.h b/include/query/irs.h
index 7ef5069..7eea14b 100644
--- a/include/query/irs.h
+++ b/include/query/irs.h
@@ -39,9 +39,9 @@ struct BufferState {
     size_t cutoff;
     std::vector<Wrapped<R>> records;
     size_t sample_size;
-    BufferView<R> buffer;
+    BufferView<R> *buffer;
 
-    BufferState(BufferView<R> buffer) : buffer(std::move(buffer)) {}
+    BufferState(BufferView<R> *buffer) : buffer(buffer) {}
 };
 
 template <ShardInterface S, RecordInterface R, bool Rejection=true>
@@ -68,10 +68,10 @@ public:
         return res;
     }
 
-    static void* get_buffer_query_state(BufferView<R> buffer, void *parms) {
-        auto res = new BufferState<R>(std::move(buffer));
+    static void* get_buffer_query_state(BufferView<R> *buffer, void *parms) {
+        auto res = new BufferState<R>(buffer);
 
-        res->cutoff = res->buffer.get_record_count();
+        res->cutoff = res->buffer->get_record_count();
         res->sample_size = 0;
 
         if constexpr (Rejection) {
@@ -82,8 +82,8 @@ public:
         auto upper_key = ((Parms<R> *) parms)->upper_bound;
 
         for (size_t i=0; i<res->cutoff; i++) {
-            if ((res->buffer.get(i)->rec.key >= lower_key) && (buffer.get(i)->rec.key <= upper_key)) { 
-                res->records.emplace_back(*(res->buffer.get(i)));
+            if ((res->buffer->get(i)->rec.key >= lower_key) && (buffer->get(i)->rec.key <= upper_key)) { 
+                res->records.emplace_back(*(res->buffer->get(i)));
             }
         }
 
@@ -181,7 +181,7 @@ public:
         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 = st->buffer.get(idx);
+                auto rec = st->buffer->get(idx);
 
                 if (rec->rec.key >= p->lower_bound && rec->rec.key <= p->upper_bound) {
                     result.emplace_back(*rec);
diff --git a/include/query/rangecount.h b/include/query/rangecount.h
index 7d88b1d..70d57d8 100644
--- a/include/query/rangecount.h
+++ b/include/query/rangecount.h
@@ -33,10 +33,10 @@ struct State {
 
 template <RecordInterface R>
 struct BufferState {
-    BufferView<R> buffer;
+    BufferView<R> *buffer;
 
-    BufferState(BufferView<R> buffer) 
-        : buffer(std::move(buffer)) {}
+    BufferState(BufferView<R> *buffer) 
+        : buffer(buffer) {}
 };
 
 template <ShardInterface S, KVPInterface R>
@@ -55,8 +55,8 @@ public:
         return res;
     }
 
-    static void* get_buffer_query_state(BufferView<R> buffer, void *parms) {
-        auto res = new BufferState<R>(std::move(buffer));
+    static void* get_buffer_query_state(BufferView<R> *buffer, void *parms) {
+        auto res = new BufferState<R>(buffer);
 
         return res;
     }
@@ -123,8 +123,8 @@ public:
         res.rec.value = 0; // tombstones
         records.emplace_back(res);
 
-        for (size_t i=0; i<s->buffer.get_record_count(); i++) {
-            auto rec = s->buffer.get(i);
+        for (size_t i=0; i<s->buffer->get_record_count(); i++) {
+            auto rec = s->buffer->get(i);
             if (rec->rec.key >= p->lower_bound && rec->rec.key <= p->upper_bound
                     && !rec->is_deleted()) {
                 if (rec->is_tombstone()) {
diff --git a/include/query/rangequery.h b/include/query/rangequery.h
index c44f5d7..1a42265 100644
--- a/include/query/rangequery.h
+++ b/include/query/rangequery.h
@@ -32,10 +32,10 @@ struct State {
 
 template <RecordInterface R>
 struct BufferState {
-    BufferView<R> buffer;
+    BufferView<R> *buffer;
 
-    BufferState(BufferView<R> buffer) 
-        : buffer(std::move(buffer)) {}
+    BufferState(BufferView<R> *buffer) 
+        : buffer(buffer) {}
 };
 
 template <ShardInterface S, RecordInterface R>
@@ -54,8 +54,8 @@ public:
         return res;
     }
 
-    static void* get_buffer_query_state(BufferView<R> buffer, void *parms) {
-        auto res = new BufferState<R>(std::move(buffer));
+    static void* get_buffer_query_state(BufferView<R> *buffer, void *parms) {
+        auto res = new BufferState<R>(buffer);
 
         return res;
     }
@@ -101,8 +101,8 @@ public:
         auto s = (BufferState<R> *) state;
 
         std::vector<Wrapped<R>> records;
-        for (size_t i=0; i<s->buffer.get_record_count(); i++) {
-            auto rec = s->buffer.get(i);
+        for (size_t i=0; i<s->buffer->get_record_count(); i++) {
+            auto rec = s->buffer->get(i);
             if (rec->rec.key >= p->lower_bound && rec->rec.key <= p->upper_bound) {
                 records.emplace_back(*rec);
             }
-- 
cgit v1.2.3


From 10b4425e842d10b7fbfa85978969ed4591d6b98e Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Wed, 7 Feb 2024 10:56:52 -0500
Subject: Fully implemented Query concept and adjusted queries to use it

---
 include/query/irs.h        |  4 ++--
 include/query/rangecount.h |  8 ++------
 include/query/rangequery.h |  6 ++----
 include/query/wirs.h       | 19 ++++++++++---------
 include/query/wss.h        | 20 +++++++++++---------
 5 files changed, 27 insertions(+), 30 deletions(-)

(limited to 'include/query')

diff --git a/include/query/irs.h b/include/query/irs.h
index 7eea14b..bef75bf 100644
--- a/include/query/irs.h
+++ b/include/query/irs.h
@@ -90,9 +90,9 @@ public:
         return res;
     }
 
-    static void process_query_states(void *query_parms, std::vector<void*> &shard_states, void *buff_state) {
+    static void process_query_states(void *query_parms, std::vector<void*> &shard_states, void *buffer_state) {
         auto p = (Parms<R> *) query_parms;
-        auto bs = (buff_state) ? (BufferState<R> *) buff_state : nullptr;
+        auto bs = (buffer_state) ? (BufferState<R> *) buffer_state : nullptr;
 
         std::vector<size_t> shard_sample_sizes(shard_states.size()+1, 0);
         size_t buffer_sz = 0;
diff --git a/include/query/rangecount.h b/include/query/rangecount.h
index 70d57d8..a09ad64 100644
--- a/include/query/rangecount.h
+++ b/include/query/rangecount.h
@@ -11,11 +11,7 @@
  */
 #pragma once
 
-#include "framework/interface/Record.h"
-#include "framework/interface/Shard.h"
-#include "framework/structure/BufferView.h"
-#include "psu-ds/PriorityQueue.h"
-#include "util/Cursor.h"
+#include "framework/QueryRequirements.h"
 
 namespace de { namespace rc {
 
@@ -61,7 +57,7 @@ public:
         return res;
     }
 
-    static void process_query_states(void *query_parms, std::vector<void*> &shard_states, void* buffer_states) {
+    static void process_query_states(void *query_parms, std::vector<void*> &shard_states, void* buffer_state) {
         return;
     }
 
diff --git a/include/query/rangequery.h b/include/query/rangequery.h
index 1a42265..c3985fa 100644
--- a/include/query/rangequery.h
+++ b/include/query/rangequery.h
@@ -10,9 +10,7 @@
  */
 #pragma once
 
-#include "framework/interface/Record.h"
-#include "framework/interface/Shard.h"
-#include "framework/structure/BufferView.h"
+#include "framework/QueryRequirements.h"
 #include "psu-ds/PriorityQueue.h"
 #include "util/Cursor.h"
 
@@ -60,7 +58,7 @@ public:
         return res;
     }
 
-    static void process_query_states(void *query_parms, std::vector<void*> &shard_states, void* buffer_states) {
+    static void process_query_states(void *query_parms, std::vector<void*> &shard_states, void* buffer_state) {
         return;
     }
 
diff --git a/include/query/wirs.h b/include/query/wirs.h
index 9b3d2ad..07c5292 100644
--- a/include/query/wirs.h
+++ b/include/query/wirs.h
@@ -12,9 +12,7 @@
  */
 #pragma once
 
-#include "framework/interface/Record.h"
-#include "framework/interface/Shard.h"
-#include "framework/structure/MutableBuffer.h"
+#include "framework/QueryRequirements.h"
 #include "psu-ds/Alias.h"
 
 namespace de { namespace wirs {
@@ -52,6 +50,7 @@ struct BufferState {
     decltype(R::weight) max_weight;
     size_t sample_size;
     decltype(R::weight) total_weight;
+    BufferView<R> *buffer;
 
     ~BufferState() {
         delete alias;
@@ -83,7 +82,7 @@ public:
         return res;
     }
 
-    static void* get_buffer_query_state(MutableBuffer<R> *buffer, void *parms) {
+    static void* get_buffer_query_state(BufferView<R> *buffer, void *parms) {
         BufferState<R> *state = new BufferState<R>();
         auto parameters = (Parms<R>*) parms;
 
@@ -92,16 +91,17 @@ public:
             state->max_weight = buffer->get_max_weight();
             state->total_weight = buffer->get_total_weight();
             state->sample_size = 0;
+            state->buffer = buffer;
             return state;
         }
 
         std::vector<decltype(R::weight)> weights;
 
-        state->cutoff = buffer->get_record_count() - 1;
+        state->buffer = buffer;
         decltype(R::weight) total_weight = 0;
 
-        for (size_t i = 0; i <= state->cutoff; i++) {
-            auto rec = buffer->get_data() + i;
+        for (size_t i = 0; i <= buffer->get_record_count(); i++) {
+            auto rec = buffer->get(i);
 
             if (rec->rec.key >= parameters->lower_bound && rec->rec.key <= parameters->upper_bound && !rec->is_tombstone() && !rec->is_deleted()) {
               weights.push_back(rec->rec.weight);
@@ -190,9 +190,10 @@ public:
         return result_set;
     }
 
-    static std::vector<Wrapped<R>> buffer_query(MutableBuffer<R> *buffer, void *state, void *parms) {
+    static std::vector<Wrapped<R>> buffer_query(void *state, void *parms) {
         auto st = (BufferState<R> *) state;
         auto p = (Parms<R> *) parms;
+        auto buffer = st->buffer;
 
         std::vector<Wrapped<R>> result;
         result.reserve(st->sample_size);
@@ -200,7 +201,7 @@ public:
         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;
+                auto rec = buffer->get(idx);
 
                 auto test = gsl_rng_uniform(p->rng) * st->max_weight;
 
diff --git a/include/query/wss.h b/include/query/wss.h
index 4c8861e..9f192ee 100644
--- a/include/query/wss.h
+++ b/include/query/wss.h
@@ -11,9 +11,8 @@
  */
 #pragma once
 
-#include "framework/interface/Record.h"
-#include "framework/interface/Shard.h"
-#include "framework/structure/MutableBuffer.h"
+#include "framework/QueryRequirements.h"
+#include "psu-ds/Alias.h"
 
 namespace de { namespace wss {
 
@@ -40,6 +39,7 @@ struct BufferState {
     psudb::Alias *alias;
     decltype(R::weight) max_weight;
     decltype(R::weight) total_weight;
+    BufferView<R> *buffer;
 
     ~BufferState() {
         delete alias;
@@ -60,23 +60,24 @@ public:
         return res;
     }
 
-    static void* get_buffer_query_state(MutableBuffer<R> *buffer, void *parms) {
+    static void* get_buffer_query_state(BufferState<R> *buffer, void *parms) {
         BufferState<R> *state = new BufferState<R>();
         auto parameters = (Parms<R>*) parms;
         if constexpr (Rejection) {
             state->cutoff = buffer->get_record_count() - 1;
             state->max_weight = buffer->get_max_weight();
             state->total_weight = buffer->get_total_weight();
+            state->buffer = buffer;
             return state;
         }
 
         std::vector<double> weights;
 
-        state->cutoff = buffer->get_record_count() - 1;
         double total_weight = 0.0;
+        state->buffer = buffer;
 
-        for (size_t i = 0; i <= state->cutoff; i++) {
-            auto rec = buffer->get_data() + i;
+        for (size_t i = 0; i <= buffer->get_record_count(); i++) {
+            auto rec = buffer->get_data(i);
             weights.push_back(rec->rec.weight);
             total_weight += rec->rec.weight;
         }
@@ -152,9 +153,10 @@ public:
         return result_set;
     }
 
-    static std::vector<Wrapped<R>> buffer_query(MutableBuffer<R> *buffer, void *state, void *parms) {
+    static std::vector<Wrapped<R>> buffer_query(void *state, void *parms) {
         auto st = (BufferState<R> *) state;
         auto p = (Parms<R> *) parms;
+        auto buffer = st->buffer;
 
         std::vector<Wrapped<R>> result;
         result.reserve(st->sample_size);
@@ -162,7 +164,7 @@ public:
         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;
+                auto rec = buffer->get(idx);
 
                 auto test = gsl_rng_uniform(p->rng) * st->max_weight;
 
-- 
cgit v1.2.3


From 2c5d549b3618b9ea72e6eece4cb4f3da5a6811a8 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Wed, 7 Feb 2024 13:42:34 -0500
Subject: Fully realized shard concept interface

---
 include/query/irs.h        | 2 +-
 include/query/rangecount.h | 2 +-
 include/query/rangequery.h | 2 +-
 include/query/wirs.h       | 2 +-
 include/query/wss.h        | 2 +-
 5 files changed, 5 insertions(+), 5 deletions(-)

(limited to 'include/query')

diff --git a/include/query/irs.h b/include/query/irs.h
index bef75bf..c14d0cf 100644
--- a/include/query/irs.h
+++ b/include/query/irs.h
@@ -44,7 +44,7 @@ struct BufferState {
     BufferState(BufferView<R> *buffer) : buffer(buffer) {}
 };
 
-template <ShardInterface S, RecordInterface R, bool Rejection=true>
+template <RecordInterface R, ShardInterface<R> S, bool Rejection=true>
 class Query {
 public:
     constexpr static bool EARLY_ABORT=false;
diff --git a/include/query/rangecount.h b/include/query/rangecount.h
index a09ad64..6c57809 100644
--- a/include/query/rangecount.h
+++ b/include/query/rangecount.h
@@ -35,7 +35,7 @@ struct BufferState {
         : buffer(buffer) {}
 };
 
-template <ShardInterface S, KVPInterface R>
+template <KVPInterface R, ShardInterface<R> S>
 class Query {
 public:
     constexpr static bool EARLY_ABORT=false;
diff --git a/include/query/rangequery.h b/include/query/rangequery.h
index c3985fa..24b38ec 100644
--- a/include/query/rangequery.h
+++ b/include/query/rangequery.h
@@ -36,7 +36,7 @@ struct BufferState {
         : buffer(buffer) {}
 };
 
-template <ShardInterface S, RecordInterface R>
+template <RecordInterface R, ShardInterface<R> S>
 class Query {
 public:
     constexpr static bool EARLY_ABORT=false;
diff --git a/include/query/wirs.h b/include/query/wirs.h
index 07c5292..4fac7e7 100644
--- a/include/query/wirs.h
+++ b/include/query/wirs.h
@@ -57,7 +57,7 @@ struct BufferState {
     }
 };
 
-template <ShardInterface S, RecordInterface R, bool Rejection=true>
+template <RecordInterface R, ShardInterface<R> S, bool Rejection=true>
 class Query {
 public:
     constexpr static bool EARLY_ABORT=false;
diff --git a/include/query/wss.h b/include/query/wss.h
index 9f192ee..ea36cb2 100644
--- a/include/query/wss.h
+++ b/include/query/wss.h
@@ -46,7 +46,7 @@ struct BufferState {
     }
 };
 
-template <ShardInterface S, RecordInterface R, bool Rejection=true>
+template <RecordInterface R, ShardInterface<R> S, bool Rejection=true>
 class Query {
 public:
     constexpr static bool EARLY_ABORT=false;
-- 
cgit v1.2.3


From bd74e27b28bd95267ce50d2e4b6f12b51d9b6aae Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Wed, 7 Feb 2024 17:23:23 -0500
Subject: Cleaned up shard files (except VPTree)

Cleaned up shard implementations, fixed a few bugs, and set up some
tests. There's still some work to be done in creating tests for the
weighted sampling operations for the alias and aug btree shards.
---
 include/query/wss.h | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'include/query')

diff --git a/include/query/wss.h b/include/query/wss.h
index ea36cb2..8797035 100644
--- a/include/query/wss.h
+++ b/include/query/wss.h
@@ -60,7 +60,7 @@ public:
         return res;
     }
 
-    static void* get_buffer_query_state(BufferState<R> *buffer, void *parms) {
+    static void* get_buffer_query_state(BufferView<R> *buffer, void *parms) {
         BufferState<R> *state = new BufferState<R>();
         auto parameters = (Parms<R>*) parms;
         if constexpr (Rejection) {
-- 
cgit v1.2.3


From 711769574e647839677739192698e400529efe75 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Thu, 8 Feb 2024 16:38:44 -0500
Subject: Updated VPTree to new shard/query interfaces

---
 include/query/knn.h | 159 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 159 insertions(+)
 create mode 100644 include/query/knn.h

(limited to 'include/query')

diff --git a/include/query/knn.h b/include/query/knn.h
new file mode 100644
index 0000000..19dcf5c
--- /dev/null
+++ b/include/query/knn.h
@@ -0,0 +1,159 @@
+/*
+ * include/query/knn.h
+ *
+ * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu>
+ *
+ * Distributed under the Modified BSD License.
+ *
+ * A query class for k-NN queries, designed for use with the VPTree
+ * shard. 
+ *
+ * FIXME: no support for tombstone deletes just yet. This would require a
+ * query resumption mechanism, most likely.
+ */
+#pragma once
+
+#include "framework/QueryRequirements.h"
+#include "psu-ds/PriorityQueue.h"
+
+namespace de { namespace knn {
+
+using psudb::PriorityQueue;
+
+template <NDRecordInterface R>
+struct Parms {
+    R point;
+    size_t k;
+};
+
+template <NDRecordInterface R>
+struct State {
+    size_t k;
+};
+
+template <NDRecordInterface R>
+struct BufferState {
+    BufferView<R> *buffer;
+
+    BufferState(BufferView<R> *buffer) 
+        : buffer(buffer) {}
+};
+
+template <NDRecordInterface R, ShardInterface<R> S>
+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) {
+        return nullptr;
+    }
+
+    static void* get_buffer_query_state(BufferView<R> *buffer, void *parms) {
+        return new BufferState<R>(buffer);
+    }
+
+    static void process_query_states(void *query_parms, std::vector<void*> &shard_states, void* buffer_state) {
+        return;
+    }
+
+    static std::vector<Wrapped<R>> query(S *shard, void *q_state, void *parms) {
+        std::vector<Wrapped<R>> results;
+        Parms<R> *p = (Parms<R> *) parms;
+        Wrapped<R> wrec;
+        wrec.rec = p->point;
+        wrec.header = 0;
+
+        PriorityQueue<Wrapped<R>, DistCmpMax<Wrapped<R>>> pq(p->k, &wrec);
+
+        shard->search(p->point, p->k, pq);
+
+        while (pq.size() > 0) {
+            results.emplace_back(*pq.peek().data);
+            pq.pop();
+        }
+
+        return results;
+    }
+
+    static std::vector<Wrapped<R>> buffer_query(void *state, void *parms) {
+        Parms<R> *p = (Parms<R> *) parms;
+        BufferState<R> *s = (BufferState<R> *) state;
+        Wrapped<R> wrec;
+        wrec.rec = p->point;
+        wrec.header = 0;
+
+        size_t k = p->k;
+
+        PriorityQueue<Wrapped<R>, DistCmpMax<Wrapped<R>>> pq(k, &wrec);
+        for (size_t i=0; i<s->buffer->get_record_count(); i++) {
+            // Skip over deleted records (under tagging)
+            if (s->buffer->get(i)->is_deleted()) {
+                continue;
+            }
+
+            if (pq.size() < k) {
+                pq.push(s->buffer->get(i));
+            } else {
+                double head_dist = pq.peek().data->rec.calc_distance(wrec.rec);
+                double cur_dist = (s->buffer->get(i))->rec.calc_distance(wrec.rec);
+
+                if (cur_dist < head_dist) {
+                    pq.pop();
+                    pq.push(s->buffer->get(i));
+                }
+            }
+        }
+
+        std::vector<Wrapped<R>> results;
+        while (pq.size() > 0) {
+            results.emplace_back(*(pq.peek().data));
+            pq.pop();
+        }
+
+        return results;
+    }
+
+    static std::vector<R> merge(std::vector<std::vector<Wrapped<R>>> &results, void *parms) {
+        Parms<R> *p = (Parms<R> *) parms;
+        R rec = p->point;
+        size_t k = p->k;
+
+        PriorityQueue<R, DistCmpMax<R>> pq(k, &rec);
+        for (size_t i=0; i<results.size(); i++) {
+            for (size_t j=0; j<results[i].size(); j++) {
+                if (pq.size() < k) {
+                    pq.push(&results[i][j].rec);
+                } else {
+                    double head_dist = pq.peek().data->calc_distance(rec);
+                    double cur_dist = results[i][j].rec.calc_distance(rec);
+
+                    if (cur_dist < head_dist) {
+                        pq.pop();
+                        pq.push(&results[i][j].rec);
+                    }
+                }
+            }
+        }
+
+        std::vector<R> output;
+        while (pq.size() > 0) {
+            output.emplace_back(*pq.peek().data);
+            pq.pop();
+        }
+
+        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;
+    }
+};
+
+}}
-- 
cgit v1.2.3


From 402fc269c0aaa671d84a6d15918735ad4b90e6b2 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Fri, 9 Feb 2024 12:30:21 -0500
Subject: Comment updates/fixes

---
 include/query/irs.h  | 1 -
 include/query/wirs.h | 1 -
 2 files changed, 2 deletions(-)

(limited to 'include/query')

diff --git a/include/query/irs.h b/include/query/irs.h
index c14d0cf..e2d9325 100644
--- a/include/query/irs.h
+++ b/include/query/irs.h
@@ -8,7 +8,6 @@
  * A query class for independent range sampling. This query requires 
  * that the shard support get_lower_bound(key), get_upper_bound(key), 
  * and get_record_at(index).
- *
  */
 #pragma once
 
diff --git a/include/query/wirs.h b/include/query/wirs.h
index 4fac7e7..ae82194 100644
--- a/include/query/wirs.h
+++ b/include/query/wirs.h
@@ -8,7 +8,6 @@
  * A query class for weighted independent range sampling. This
  * class is tightly coupled with include/shard/AugBTree.h, and
  * so is probably of limited general utility.
- *
  */
 #pragma once
 
-- 
cgit v1.2.3