Optimize reference filtering.

include/collection.h

@@ -463,6 +463,10 @@ public:
 
     Option<bool> validate_reference_filter(const std::string& filter_query) const;
 
+    Option<bool> get_reference_filter_ids(const std::string & filter_query,
+                                          const std::string & collection_name,
+                                          std::pair<uint32_t, uint32_t*>& reference_index_ids) const;
+
     Option<bool> validate_reference_filter(const std::string& filter_query) const;
 
     Option<nlohmann::json> get(const std::string & id) const;

include/index.h

@@ -706,6 +706,10 @@ public:
                                                   filter_result_t& filter_result,
                                                   const std::string & reference_helper_field_name) const;
 
+    void do_reference_filtering_with_lock(std::pair<uint32_t, uint32_t*>& reference_index_ids,
+                                          filter_node_t const* const& filter_tree_root,
+                                          const std::string& reference_field_name) const;
+
     void refresh_schemas(const std::vector<field>& new_fields, const std::vector<field>& del_fields);
 
     // the following methods are not synchronized because their parent calls are synchronized or they are const/static

src/collection.cpp

@@ -2580,6 +2580,41 @@ Option<bool> Collection::validate_reference_filter(const std::string& filter_que
     return Option<bool>(true);
 }
 
+Option<bool> Collection::get_reference_filter_ids(const std::string & filter_query,
+                                                  const std::string & collection_name,
+                                                  std::pair<uint32_t, uint32_t*>& reference_index_ids) const {
+    std::shared_lock lock(mutex);
+
+    std::string reference_field_name;
+    for (auto const& field: fields) {
+        if (!field.reference.empty() &&
+            field.reference.find(collection_name) == 0 &&
+            field.reference.find('.') == collection_name.size()) {
+            reference_field_name = field.name;
+            break;
+        }
+    }
+
+    if (reference_field_name.empty()) {
+        return Option<bool>(400, "Could not find any field in `" + name + "` referencing the collection `"
+                                                    + collection_name + "`.");
+    }
+
+    const std::string doc_id_prefix = std::to_string(collection_id) + "_" + DOC_ID_PREFIX + "_";
+    filter_node_t* filter_tree_root = nullptr;
+    Option<bool> filter_op = filter::parse_filter_query(filter_query, search_schema,
+                                                        store, doc_id_prefix, filter_tree_root);
+    if(!filter_op.ok()) {
+        return filter_op;
+    }
+
+    reference_field_name += "_sequence_id";
+    index->do_reference_filtering_with_lock(reference_index_ids, filter_tree_root, reference_field_name);
+
+    delete filter_tree_root;
+    return Option<bool>(true);
+}
+
 Option<bool> Collection::validate_reference_filter(const std::string& filter_query) const {
     std::shared_lock lock(mutex);
 

src/index.cpp

@@ -1497,7 +1497,7 @@ Option<bool> Index::do_filtering(filter_node_t* const root,
                                  const uint32_t& context_ids_length,
                                  const uint32_t* context_ids) const {
     // auto begin = std::chrono::high_resolution_clock::now();
-    const filter a_filter = root->filter_exp;
+/**/    const filter a_filter = root->filter_exp;
 
     bool is_referenced_filter = !a_filter.referenced_collection_name.empty();
     if (is_referenced_filter) {

test/collection_join_test.cpp

@@ -284,9 +284,6 @@ TEST_F(CollectionJoinTest, IndexDocumentHavingReferenceField) {
     ASSERT_TRUE(add_doc_op.ok());
     ASSERT_EQ(customer_collection->get("0").get().count("reference_id_sequence_id"), 1);
 
-    // Referenced document should be accessible from Customers collection.
-    auto sequence_id = collectionManager.get_collection("Products")->get_seq_id_collection_prefix() + "_" +
-                                customer_collection->get("0").get()["product_id_sequence_id"].get<std::string>();
     nlohmann::json document;
     // Referenced document's sequence_id must be valid.
     auto get_op = collectionManager.get_collection("Products")->get_document_from_store(