From 438feac7e56fee425d9c6f1a43298ff9dc5b71d1 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Fri, 19 Apr 2024 17:38:16 -0400 Subject: Properly implemented support for iteratively decomposable problems --- include/framework/DynamicExtension.h | 40 ++++++++++++++++++++---------------- include/framework/interface/Query.h | 6 ++++-- 2 files changed, 26 insertions(+), 20 deletions(-) (limited to 'include/framework') diff --git a/include/framework/DynamicExtension.h b/include/framework/DynamicExtension.h index b154be8..6fd95c6 100644 --- a/include/framework/DynamicExtension.h +++ b/include/framework/DynamicExtension.h @@ -546,30 +546,34 @@ private: std::vector> shards; std::vector states = vers->get_query_states(shards, parms); + std::vector results; Q::process_query_states(parms, states, buffer_state); - std::vector>> query_results(shards.size() + 1); - for (size_t i=0; i> local_results; - ShardID shid; - - if (i == 0) { /* process the buffer first */ - local_results = Q::buffer_query(buffer_state, parms); - shid = INVALID_SHID; - } else { - local_results = Q::query(shards[i - 1].second, states[i - 1], parms); - shid = shards[i - 1].first; - } + do { + std::vector>> query_results(shards.size() + 1); + for (size_t i=0; i> local_results; + ShardID shid; + + if (i == 0) { /* process the buffer first */ + local_results = Q::buffer_query(buffer_state, parms); + shid = INVALID_SHID; + } else { + local_results = Q::query(shards[i - 1].second, states[i - 1], parms); + shid = shards[i - 1].first; + } - query_results[i] = std::move(filter_deletes(local_results, shid, vers, &buffer)); + query_results[i] = std::move(filter_deletes(local_results, shid, vers, &buffer)); - if constexpr (Q::EARLY_ABORT) { - if (query_results[i].size() > 0) break; + if constexpr (Q::EARLY_ABORT) { + if (query_results[i].size() > 0) break; + } } - } + Q::merge(query_results, parms, results); + + } while (Q::repeat(parms, results, states, buffer_state)); - auto result = Q::merge(query_results, parms); - args->result_set.set_value(std::move(result)); + args->result_set.set_value(std::move(results)); ((DynamicExtension *) args->extension)->end_job(epoch); diff --git a/include/framework/interface/Query.h b/include/framework/interface/Query.h index 3d487f0..577d6cd 100644 --- a/include/framework/interface/Query.h +++ b/include/framework/interface/Query.h @@ -13,17 +13,19 @@ namespace de{ template -concept QueryInterface = requires(void *p, S *sh, std::vector &s, std::vector>> &rv, BufferView *bv) { +concept QueryInterface = requires(void *p, S *sh, std::vector &s, std::vector>> &rv, BufferView *bv, std::vector &resv) { {Q::get_query_state(sh, p)} -> std::convertible_to; {Q::get_buffer_query_state(bv, p)} -> std::convertible_to; {Q::process_query_states(p, s, p)}; {Q::query(sh, p, p)} -> std::convertible_to>>; {Q::buffer_query(p, p)} -> std::convertible_to>>; - {Q::merge(rv, p)} -> std::convertible_to>; + {Q::merge(rv, p, resv)}; {Q::delete_query_state(p)} -> std::same_as; {Q::delete_buffer_query_state(p)} -> std::same_as; + {Q::repeat(p, resv, s, p)} -> std::same_as; + {Q::EARLY_ABORT} -> std::convertible_to; {Q::SKIP_DELETE_FILTER} -> std::convertible_to; }; -- cgit v1.2.3