Merge pull request #374 from singnet/senna-361-1

[#361] Add optional filtering of QueryAnswers which can be considered duplicates

Author: andre-senna

src/link_creation_agent/link_creation_agent.cc

@@ -104,8 +104,8 @@ void LinkCreationAgent::run() {
 shared_ptr<PatternMatchingQueryProxy> LinkCreationAgent::query(vector<string>& query_tokens,
                                                                string context,
                                                                bool update_attention_broker) {
-    shared_ptr<PatternMatchingQueryProxy> proxy =
-        make_shared<PatternMatchingQueryProxy>(query_tokens, context, update_attention_broker);
+    shared_ptr<PatternMatchingQueryProxy> proxy = make_shared<PatternMatchingQueryProxy>(
+        query_tokens, context, false, update_attention_broker, false);
     service_bus->issue_bus_command(proxy);
     return proxy;
 }

src/main/query_client_main.cc

@@ -65,7 +65,7 @@ int main(int argc, char* argv[]) {
 
     shared_ptr<ServiceBus> service_bus = ServiceBusSingleton::get_instance();
     shared_ptr<PatternMatchingQueryProxy> proxy =
-        make_shared<PatternMatchingQueryProxy>(query, "", update_attention_broker);
+        make_shared<PatternMatchingQueryProxy>(query, "", true, update_attention_broker, false);
     service_bus->issue_bus_command(proxy);
     shared_ptr<QueryAnswer> query_answer;
     int count = 0;
@@ -79,6 +79,7 @@ int main(int argc, char* argv[]) {
             }
         }
     }
+    cout << count << " answers" << endl;
     if (count == 0) {
         cout << "No match for query" << endl;
     }

src/main/word_query_main.cc

@@ -127,7 +127,7 @@ void run(const string& context, const string& word_tag) {
                                  "\"" + word_tag + "\""};
 
     shared_ptr<PatternMatchingQueryProxy> proxy =
-        make_shared<PatternMatchingQueryProxy>(query_word, context, true);
+        make_shared<PatternMatchingQueryProxy>(query_word, context, false, true, false);
     service_bus->issue_bus_command(proxy);
 
     shared_ptr<QueryAnswer> query_answer;
@@ -171,7 +171,7 @@ void run(const string& context, const string& word_tag) {
     }
 
     shared_ptr<PatternMatchingQueryProxy> proxy2 =
-        make_shared<PatternMatchingQueryProxy>(query_word, context, true, true);
+        make_shared<PatternMatchingQueryProxy>(query_word, context, true, true, false);
     service_bus->issue_bus_command(proxy2);
     while (!proxy2->finished()) {
         Utils::sleep();

src/query_engine/PatternMatchingQueryProcessor.cc

@@ -8,6 +8,7 @@
 #include "ServiceBus.h"
 #include "Sink.h"
 #include "Terminal.h"
+#include "UniqueAssignmentFilter.h"
 
 #define LOG_LEVEL INFO_LEVEL
 #include "Logger.h"
@@ -137,10 +138,13 @@ void PatternMatchingQueryProcessor::thread_process_one_query(
     if (proxy->args.size() < 2) {
         Utils::error("Syntax error in query command. Missing implicit parameters.");
     }
-    proxy->set_context(proxy->args[0]);
-    proxy->set_attention_update_flag(proxy->args[1] == "1");
-    proxy->set_count_flag(proxy->args[2] == "1");
-    proxy->query_tokens.insert(proxy->query_tokens.begin(), proxy->args.begin() + 3, proxy->args.end());
+    unsigned int skip_arg = 0;
+    proxy->set_context(proxy->args[skip_arg++]);
+    proxy->set_unique_assignment_flag(proxy->args[skip_arg++] == "1");
+    proxy->set_attention_update_flag(proxy->args[skip_arg++] == "1");
+    proxy->set_count_flag(proxy->args[skip_arg++] == "1");
+    proxy->query_tokens.insert(
+        proxy->query_tokens.begin(), proxy->args.begin() + skip_arg, proxy->args.end());
     LOG_DEBUG("Setting up query tree");
     shared_ptr<QueryElement> root_query_element = setup_query_tree(proxy);
     set<string> joint_answer;  // used to stimulate attention broker
@@ -212,8 +216,14 @@ shared_ptr<QueryElement> PatternMatchingQueryProcessor::setup_query_tree(
             element_stack.push(build_link_template(proxy, cursor, element_stack));
         } else if (proxy->query_tokens[cursor] == "AND") {
             element_stack.push(build_and(proxy, cursor, element_stack));
+            if (proxy->get_unique_assignment_flag()) {
+                element_stack.push(build_unique_assignment_filter(proxy, cursor, element_stack));
+            }
         } else if (proxy->query_tokens[cursor] == "OR") {
             element_stack.push(build_or(proxy, cursor, element_stack));
+            if (proxy->get_unique_assignment_flag()) {
+                element_stack.push(build_unique_assignment_filter(proxy, cursor, element_stack));
+            }
         } else {
             Utils::error("Invalid token " + proxy->query_tokens[cursor] +
                          " in PATTERN_MATCHING_QUERY message");
@@ -630,3 +640,18 @@ shared_ptr<QueryElement> PatternMatchingQueryProcessor::build_link(
     }
     return NULL;  // Just to avoid warnings. This is not actually reachable.
 }
+
+shared_ptr<QueryElement> PatternMatchingQueryProcessor::build_unique_assignment_filter(
+    shared_ptr<PatternMatchingQueryProxy> proxy,
+    unsigned int cursor,
+    stack<shared_ptr<QueryElement>>& element_stack) {
+    if (element_stack.size() < 1) {
+        Utils::error(
+            "PATTERN_MATCHING_QUERY message: parse error in tokens - too few arguments for "
+            "UniqueAssignmentFilter");
+    }
+
+    shared_ptr<QueryElement> input = element_stack.top();
+    element_stack.pop();
+    return make_shared<UniqueAssignmentFilter>(input);
+}

src/query_engine/PatternMatchingQueryProcessor.h

@@ -68,6 +68,11 @@ class PatternMatchingQueryProcessor : public BusCommandProcessor {
                                         unsigned int cursor,
                                         stack<shared_ptr<QueryElement>>& element_stack);
 
+    shared_ptr<QueryElement> build_unique_assignment_filter(
+        shared_ptr<PatternMatchingQueryProxy> proxy,
+        unsigned int cursor,
+        stack<shared_ptr<QueryElement>>& element_stack);
+
     vector<thread*> query_threads;
     mutex query_threads_mutex;
     shared_ptr<PatternMatchingQueryProxy> proxy;

src/query_engine/PatternMatchingQueryProxy.cc

@@ -23,6 +23,7 @@ PatternMatchingQueryProxy::PatternMatchingQueryProxy() {
 
 PatternMatchingQueryProxy::PatternMatchingQueryProxy(const vector<string>& tokens,
                                                      const string& context,
+                                                     bool unique_assignment,
                                                      bool update_attention_broker,
                                                      bool count_only)
     : BusCommandProxy() {
@@ -31,13 +32,18 @@ PatternMatchingQueryProxy::PatternMatchingQueryProxy(const vector<string>& token
     init();
     this->command = ServiceBus::PATTERN_MATCHING_QUERY;
     this->count_flag = count_only;
-    this->args = {context, to_string(update_attention_broker), to_string(count_flag)};
+    this->unique_assignment_flag = unique_assignment;
+    this->args = {context,
+                  to_string(unique_assignment),
+                  to_string(update_attention_broker),
+                  to_string(count_flag)};
     this->args.insert(this->args.end(), tokens.begin(), tokens.end());
 }
 
 void PatternMatchingQueryProxy::init() {
     this->answer_flow_finished = false;
     this->abort_flag = false;
+    this->unique_assignment_flag = false;
     this->update_attention_broker = false;
     this->answer_count = 0;
     this->count_flag = false;
@@ -123,6 +129,16 @@ void PatternMatchingQueryProxy::set_count_flag(bool flag) {
     this->count_flag = flag;
 }
 
+bool PatternMatchingQueryProxy::get_unique_assignment_flag() {
+    lock_guard<mutex> semaphore(this->api_mutex);
+    return this->unique_assignment_flag;
+}
+
+void PatternMatchingQueryProxy::set_unique_assignment_flag(bool flag) {
+    lock_guard<mutex> semaphore(this->api_mutex);
+    this->unique_assignment_flag = flag;
+}
+
 // ---------------------------------------------------------------------------------------------
 // Virtual superclass API from_remote_peer() and the piggyback methods called by it
 

src/query_engine/PatternMatchingQueryProxy.h

@@ -44,16 +44,19 @@ class PatternMatchingQueryProxy : public BusCommandProxy {
      *
      * @param tokens Query tokens.
      * @param context AttentionBroker context
+     * @param unique_assignment When true, operators (e.g. And, Or) don't output more than one
+     * QueryAnswer with the same variable assignment.
      * @param update_attention_broker Flag to trigger AttentionBroker update based on this query
      * results
      * @param count_only Flag to indicate that this query is supposed to count the results and not
      * actually provide the query answers (i.e. no QueryAnswer is sent from the command executor and
      * the caller of the query).
      */
     PatternMatchingQueryProxy(const vector<string>& tokens,
-                              const string& context = "",
-                              bool update_attention_broker = false,
-                              bool count_only = false);
+                              const string& context,
+                              bool unique_assignment,
+                              bool update_attention_broker,
+                              bool count_only);
 
     /**
      * Destructor.
@@ -162,6 +165,24 @@ class PatternMatchingQueryProxy : public BusCommandProxy {
      */
     void set_count_flag(bool flag);
 
+    /**
+     * Getter for unique_assignment_flag
+     *
+     * unique_assignment_flag prevents duplicated variable assignment in Operators' output
+     *
+     * @return unique_assignment_flag
+     */
+    bool get_unique_assignment_flag();
+
+    /**
+     * Setter for unique_assignment_flag
+     *
+     * unique_assignment_flag prevents duplicated variable assignment in Operators' output
+     *
+     * @param flag Flag
+     */
+    void set_unique_assignment_flag(bool flag);
+
     /**
      * Getter for query_tokens
      *
@@ -221,6 +242,7 @@ class PatternMatchingQueryProxy : public BusCommandProxy {
     string context;
     bool update_attention_broker;
     bool count_flag;
+    bool unique_assignment_flag;
 
     void init();
 };

src/query_engine/QueryAnswer.cc

@@ -94,6 +94,27 @@ string Assignment::to_string() {
     return answer;
 }
 
+bool Assignment::operator==(const Assignment& other) const {
+    if (this->size != other.size) {
+        return false;
+    }
+    unsigned int n = this->size;
+    for (unsigned int i = 0; i < n; i++) {
+        unsigned int j = 0;
+        while ((j != n) && strncmp(this->labels[i], other.labels[j], MAX_VARIABLE_NAME_SIZE)) {
+            j++;
+        }
+        if (j == n) {
+            // There's a variable in "this" which doesn't exist in "other"
+            return false;
+        } else if (strncmp(this->values[i], other.values[j], HANDLE_HASH_SIZE)) {
+            // There same variable have different values in "this" and "other"
+            return false;
+        }
+    }
+    return true;
+}
+
 // -------------------------------------------------------------------------------------------------
 // QueryAnswer
 

src/query_engine/QueryAnswer.h

@@ -26,6 +26,7 @@ namespace query_engine {
  */
 class Assignment {
     friend class QueryAnswer;
+    friend struct std::hash<Assignment>;
 
    public:
     /**
@@ -115,6 +116,12 @@ class Assignment {
      */
     string to_string();
 
+    /**
+     * Two Assignments are equal iff they have the exact set of variable lables with
+     * the exact same assigned values.
+     */
+    bool operator==(const Assignment& other) const;
+
    private:
     const char* labels[MAX_NUMBER_OF_VARIABLES_IN_QUERY];
     const char* values[MAX_NUMBER_OF_VARIABLES_IN_QUERY];
@@ -260,3 +267,21 @@ class QueryAnswer {
 };
 
 }  // namespace query_engine
+
+template <>
+struct std::hash<query_engine::Assignment> {
+    std::size_t operator()(const query_engine::Assignment& k) const {
+        if (k.size == 0) {
+            return 0;
+        }
+
+        std::size_t hash_value = 1;
+        for (unsigned int i = 0; i < k.size; i++) {
+            hash_value = hash_value ^ ((std::hash<string>()(string(k.labels[i])) ^
+                                        (std::hash<string>()(string(k.values[i])) << 1)) >>
+                                       1);
+        }
+
+        return hash_value;
+    }
+};

src/query_engine/query_element/BUILD

@@ -13,6 +13,7 @@ cc_library(
     ":query_element",
     ":link_template",
     ":source",
+    ":unique_assignment_filter",
   ],
 )
 
@@ -103,6 +104,15 @@ cc_library(
 )
 
 
+cc_library(
+  name = "unique_assignment_filter",
+  srcs = ["UniqueAssignmentFilter.cc"],
+  hdrs = ["UniqueAssignmentFilter.h"],
+  deps = [
+    ":query_element",
+  ],
+)
+
 cc_library(
   name = "source",
   srcs = ["Source.cc"],

src/query_engine/query_element/UniqueAssignmentFilter.cc

@@ -0,0 +1,67 @@
+#include "UniqueAssignmentFilter.h"
+
+#define LOG_LEVEL INFO_LEVEL
+#include "Logger.h"
+
+using namespace std;
+using namespace query_element;
+
+// -------------------------------------------------------------------------------------------------
+// Constructors, destructors and initialization
+
+UniqueAssignmentFilter::UniqueAssignmentFilter(const shared_ptr<QueryElement>& input)
+    : Operator<1>({input}) {
+    initialize(input);
+}
+
+UniqueAssignmentFilter::~UniqueAssignmentFilter() { graceful_shutdown(); }
+
+void UniqueAssignmentFilter::initialize(const shared_ptr<QueryElement>& input) {
+    this->id = "UniqueAssignmentFilter(" + input->id + ")";
+    LOG_INFO(this->id);
+}
+
+// -------------------------------------------------------------------------------------------------
+// QueryElement API
+
+void UniqueAssignmentFilter::setup_buffers() {
+    Operator<1>::setup_buffers();
+    this->operator_thread = new thread(&UniqueAssignmentFilter::thread_filter, this);
+}
+
+void UniqueAssignmentFilter::graceful_shutdown() {
+    Operator<1>::graceful_shutdown();
+    if (this->operator_thread != NULL) {
+        this->operator_thread->join();
+        delete this->operator_thread;
+        this->operator_thread = NULL;
+    }
+}
+
+// -------------------------------------------------------------------------------------------------
+// Private methods
+
+void UniqueAssignmentFilter::thread_filter() {
+    unordered_set<Assignment> already_used;
+
+    while (true) {
+        if (this->input_buffer[0]->is_query_answers_finished() &&
+            this->input_buffer[0]->is_query_answers_empty()) {
+            this->output_buffer->query_answers_finished();
+            break;
+        }
+        QueryAnswer* answer = dynamic_cast<QueryAnswer*>(this->input_buffer[0]->pop_query_answer());
+        if (answer != NULL) {
+            if (already_used.find(answer->assignment) == already_used.end()) {
+                // New assignment. Let the QueryAnswer pass.
+                already_used.insert(answer->assignment);
+                this->output_buffer->add_query_answer(answer);
+            } else {
+                // Assignment has already been processed. Delete duplicate QueryAnswer.
+                delete answer;
+            }
+        } else {
+            Utils::sleep();
+        }
+    }
+}