started refactoring queries interface

1 files changed, 0 insertions, 361 deletions

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;
-    }
-};
-
-
-
 }