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 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 216 insertions(+)
 create mode 100644 include/query/irs.h

(limited to 'include/query/irs.h')

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;
+    }
+};
+}}
-- 
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 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

(limited to 'include/query/irs.h')

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) {
-- 
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 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

(limited to 'include/query/irs.h')

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
-- 
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/irs.h')

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 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

(limited to 'include/query/irs.h')

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);
-- 
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 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

(limited to 'include/query/irs.h')

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;
-- 
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 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'include/query/irs.h')

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;
-- 
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 -
 1 file changed, 1 deletion(-)

(limited to 'include/query/irs.h')

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