From d6e08e9d8d3ac9b356ac50cee22b41f828160247 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Thu, 27 Jul 2023 18:21:26 -0400
Subject: Expanded query interface

Query interface now enables skipping of delete processing and stopping
query processing when first match is found.
---
 include/framework/DynamicExtension.h | 41 ++++++++++++++++++++++++++++++------
 include/framework/QueryInterface.h   | 10 ++++++++-
 2 files changed, 43 insertions(+), 8 deletions(-)

(limited to 'include/framework')

diff --git a/include/framework/DynamicExtension.h b/include/framework/DynamicExtension.h
index bd09e1f..5c903b9 100644
--- a/include/framework/DynamicExtension.h
+++ b/include/framework/DynamicExtension.h
@@ -121,16 +121,39 @@ public:
 
         Q::process_query_states(parms, states, buffer_state);
 
-        std::vector<std::vector<R>> query_results(shards.size() + 1);
+        std::vector<std::vector<Wrapped<R>>> query_results(shards.size() + 1);
 
         // Execute the query for the buffer
         auto buffer_results = Q::buffer_query(buffer, buffer_state, parms);
-        query_results[0] = filter_deletes(buffer_results, {-1, -1}, buffer);
+        query_results[0] = std::move(filter_deletes(buffer_results, {-1, -1}, buffer));
+        if constexpr (Q::EARLY_ABORT) {
+            if (query_results[0].size() > 0) {
+                auto result = Q::merge(query_results, parms);
+                for (size_t i=0; i<states.size(); i++) {
+                    Q::delete_query_state(states[i]);
+                }
+
+                Q::delete_buffer_query_state(buffer_state);
+                return result;
+            }
+        }
 
         // Execute the query for each shard
         for (size_t i=0; i<shards.size(); i++) {
             auto shard_results = Q::query(shards[i].second, states[i], parms);
-            query_results[i+1] = filter_deletes(shard_results, shards[i].first, buffer);
+            query_results[i+1] = std::move(filter_deletes(shard_results, shards[i].first, buffer));
+            if constexpr (Q::EARLY_ABORT) {
+                if (query_results[i].size() > 0) {
+                    auto result = Q::merge(query_results, parms);
+                    for (size_t i=0; i<states.size(); i++) {
+                        Q::delete_query_state(states[i]);
+                    }
+
+                    Q::delete_buffer_query_state(buffer_state);
+
+                    return result;
+                }
+            }
         }
         
         // Merge the results together
@@ -264,8 +287,12 @@ private:
         return buffer->append(rec, ts);
     }
 
-    std::vector<R> filter_deletes(std::vector<Wrapped<R>> &records, ShardID shid, Buffer *buffer) {
-        std::vector<R> processed_records;
+    std::vector<Wrapped<R>> filter_deletes(std::vector<Wrapped<R>> &records, ShardID shid, Buffer *buffer) {
+        if constexpr (!Q::SKIP_DELETE_FILTER) {
+            return records;
+        }
+
+        std::vector<Wrapped<R>> processed_records;
         processed_records.reserve(records.size());
 
         // For delete tagging, we just need to check the delete bit on each
@@ -276,7 +303,7 @@ private:
                     continue;
                 }
 
-                processed_records.emplace_back(rec.rec);
+                processed_records.emplace_back(rec);
             }
 
             return processed_records;
@@ -305,7 +332,7 @@ private:
                 }
             }
 
-            processed_records.emplace_back(rec.rec);
+            processed_records.emplace_back(rec);
         }
 
         return processed_records;
diff --git a/include/framework/QueryInterface.h b/include/framework/QueryInterface.h
index 886bdc8..46a1ce1 100644
--- a/include/framework/QueryInterface.h
+++ b/include/framework/QueryInterface.h
@@ -8,11 +8,12 @@
  */
 #pragma once
 
+#include <vector>
 #include <concepts>
 #include "util/types.h"
 
 template <typename Q>
-concept QueryInterface = requires(Q q, void *p) {
+concept QueryInterface = requires(Q q, void *p, std::vector<void*> &s) {
 
 /*
     {q.get_query_state(p, p)} -> std::convertible_to<void*>;
@@ -22,6 +23,13 @@ concept QueryInterface = requires(Q q, void *p) {
     {q.merge()};
     {q.delete_query_state(p)};
 */
+    {Q::EARLY_ABORT} -> std::convertible_to<bool>;
+    {Q::SKIP_DELETE_FILTER} -> std::convertible_to<bool>;
+    //{Q::get_query_state(p, p)} -> std::convertible_to<void*>;
+    //{Q::get_buffer_query_state(p, p)} -> std::convertible_to<void*>;
+    {Q::process_query_states(p, s, p)};
 
     {Q::delete_query_state(std::declval<void*>())} -> std::same_as<void>;
+    {Q::delete_buffer_query_state(p)};
+
 };
-- 
cgit v1.2.3