diff --git a/src/agents/atomdb_broker/AtomDBProxy.cc b/src/agents/atomdb_broker/AtomDBProxy.cc index defe8cfe2..216f59f34 100644 --- a/src/agents/atomdb_broker/AtomDBProxy.cc +++ b/src/agents/atomdb_broker/AtomDBProxy.cc @@ -167,7 +167,7 @@ void AtomDBProxy::add_atoms_callback(const vector& tokens) { for (auto& atom : atoms) { buffer.push_back(atom.get()); } - this->atomdb->add_atoms(buffer, false, true); + this->atomdb->add_atoms(buffer, true); } catch (const exception& e) { LOG_ERROR("Error processing batch: " << e.what()); } @@ -215,7 +215,7 @@ void AtomDBProxy::process_atom_batches() { this->pending_atoms_count -= atoms.size(); lock.unlock(); auto job = [this, atoms = std::move(atoms)]() { - this->atomdb->add_atoms(atoms, false, true); + this->atomdb->add_atoms(atoms, true); for (auto& atom : atoms) { delete atom; } diff --git a/src/agents/link_creation_agent/MettaTemplateProcessor.cc b/src/agents/link_creation_agent/MettaTemplateProcessor.cc index a94f06452..e24c3859d 100644 --- a/src/agents/link_creation_agent/MettaTemplateProcessor.cc +++ b/src/agents/link_creation_agent/MettaTemplateProcessor.cc @@ -40,7 +40,7 @@ static void create_missing_atoms_in_atomdb(shared_ptr parser for (const auto& element : parser_actions->handle_to_atom) { if (dynamic_pointer_cast(element.second) != nullptr) { try { - atomdb->add_node(dynamic_pointer_cast(element.second).get(), false); + atomdb->add_node(dynamic_pointer_cast(element.second).get()); LOG_DEBUG("Node added to AtomDB: " << element.second->to_string()); } catch (const std::exception& e) { LOG_ERROR("Error adding node to AtomDB: " << e.what()); @@ -68,7 +68,7 @@ static void create_missing_atoms_in_atomdb(shared_ptr parser RAISE_ERROR("Parsed atom is not a Link for metta expression: " + metta_expression_cp); continue; } - atomdb->add_link(dynamic_pointer_cast(link).get(), false); + atomdb->add_link(dynamic_pointer_cast(link).get()); LOG_DEBUG("Link added to AtomDB: " << metta_expression_cp); } catch (const std::exception& e) { LOG_ERROR("Error adding link to AtomDB: " << e.what()); diff --git a/src/atomdb/AtomDB.h b/src/atomdb/AtomDB.h index 25256312b..babcc9135 100644 --- a/src/atomdb/AtomDB.h +++ b/src/atomdb/AtomDB.h @@ -7,6 +7,7 @@ #include "AtomDBAPITypes.h" #include "HandleDecoder.h" #include "LinkSchema.h" +#include "Merger.h" #include "Properties.h" using namespace std; @@ -41,19 +42,32 @@ class AtomDB : public HandleDecoder { virtual set nodes_exist(const vector& handles) = 0; virtual set links_exist(const vector& handles) = 0; - virtual string add_atom(const atoms::Atom* atom, bool throw_if_exists = false) = 0; - virtual string add_node(const atoms::Node* node, bool throw_if_exists = false) = 0; - virtual string add_link(const atoms::Link* link, bool throw_if_exists = false) = 0; - - virtual vector add_atoms(const vector& atoms, - bool throw_if_exists = false, - bool is_transactional = false) = 0; + /** + * Add methods take an optional Merger. + * - merger == NULL: upsert (insert if missing, replace if present) + * - merger != NULL and atom exists: merge into a working copy, then persist only if + * merge() returns true; if merge() returns false, stored state is left unchanged + * and the returned handle (or batch slot) is "" + * - Use &ThrowIfExistsMerger::instance() to reject duplicates by throwing + * (former throw_if_exists=true); earlier items in a batch may already be applied + * - Use &SkipIfExistsMerger::instance() to soft-skip duplicates (merge returns false, + * returned handle/slot is "", batch continues) + * - Soft merge() returns false does not abort the rest of a batch + * - Merge-enabled adds assume a single writer per handle (no concurrent RMW atomicity) + */ + virtual string add_atom(const atoms::Atom* atom, const atoms::Merger* merger = NULL) = 0; + virtual string add_node(const atoms::Node* node, const atoms::Merger* merger = NULL) = 0; + virtual string add_link(const atoms::Link* link, const atoms::Merger* merger = NULL) = 0; + + virtual vector add_atoms(const vector& atom_list, + bool is_transactional = false, + const atoms::Merger* merger = NULL) = 0; virtual vector add_nodes(const vector& nodes, - bool throw_if_exists = false, - bool is_transactional = false) = 0; + bool is_transactional = false, + const atoms::Merger* merger = NULL) = 0; virtual vector add_links(const vector& links, - bool throw_if_exists = false, - bool is_transactional = false) = 0; + bool is_transactional = false, + const atoms::Merger* merger = NULL) = 0; virtual bool delete_atom(const string& handle, bool delete_link_targets = false) = 0; virtual bool delete_node(const string& handle, bool delete_link_targets = false) = 0; diff --git a/src/atomdb/BUILD b/src/atomdb/BUILD index fb0f072b7..68b602e48 100644 --- a/src/atomdb/BUILD +++ b/src/atomdb/BUILD @@ -24,6 +24,7 @@ cc_library( includes = ["."], deps = [ ":atomdb_api_types", + "//commons/atoms:atoms_lib", ], ) diff --git a/src/atomdb/adapterdb/AdapterDB.cc b/src/atomdb/adapterdb/AdapterDB.cc index 8669381bc..8b8d4916e 100644 --- a/src/atomdb/adapterdb/AdapterDB.cc +++ b/src/atomdb/adapterdb/AdapterDB.cc @@ -137,40 +137,40 @@ set AdapterDB::links_exist(const vector& handles) { return this->atomdb_backend->links_exist(handles); } -string AdapterDB::add_atom(const atoms::Atom* atom, bool throw_if_exists) { +string AdapterDB::add_atom(const atoms::Atom* atom, const atoms::Merger* merger) { this->ensure_backend_ready(); - return this->atomdb_backend->add_atom(atom, throw_if_exists); + return this->atomdb_backend->add_atom(atom, merger); } -string AdapterDB::add_node(const atoms::Node* node, bool throw_if_exists) { +string AdapterDB::add_node(const atoms::Node* node, const atoms::Merger* merger) { this->ensure_backend_ready(); - return this->atomdb_backend->add_node(node, throw_if_exists); + return this->atomdb_backend->add_node(node, merger); } -string AdapterDB::add_link(const atoms::Link* link, bool throw_if_exists) { +string AdapterDB::add_link(const atoms::Link* link, const atoms::Merger* merger) { this->ensure_backend_ready(); - return this->atomdb_backend->add_link(link, throw_if_exists); + return this->atomdb_backend->add_link(link, merger); } -vector AdapterDB::add_atoms(const vector& atoms, - bool throw_if_exists, - bool is_transactional) { +vector AdapterDB::add_atoms(const vector& atom_list, + bool is_transactional, + const atoms::Merger* merger) { this->ensure_backend_ready(); - return this->atomdb_backend->add_atoms(atoms, throw_if_exists, is_transactional); + return this->atomdb_backend->add_atoms(atom_list, is_transactional, merger); } vector AdapterDB::add_nodes(const vector& nodes, - bool throw_if_exists, - bool is_transactional) { + bool is_transactional, + const atoms::Merger* merger) { this->ensure_backend_ready(); - return this->atomdb_backend->add_nodes(nodes, throw_if_exists, is_transactional); + return this->atomdb_backend->add_nodes(nodes, is_transactional, merger); } vector AdapterDB::add_links(const vector& links, - bool throw_if_exists, - bool is_transactional) { + bool is_transactional, + const atoms::Merger* merger) { this->ensure_backend_ready(); - return this->atomdb_backend->add_links(links, throw_if_exists, is_transactional); + return this->atomdb_backend->add_links(links, is_transactional, merger); } bool AdapterDB::delete_atom(const string& handle, bool delete_link_targets) { diff --git a/src/atomdb/adapterdb/AdapterDB.h b/src/atomdb/adapterdb/AdapterDB.h index d7943bc5d..e5dc43104 100644 --- a/src/atomdb/adapterdb/AdapterDB.h +++ b/src/atomdb/adapterdb/AdapterDB.h @@ -82,19 +82,19 @@ class AdapterDB : public AtomDB { set nodes_exist(const vector& handles) override; set links_exist(const vector& handles) override; - string add_atom(const atoms::Atom* atom, bool throw_if_exists = false) override; - string add_node(const atoms::Node* node, bool throw_if_exists = false) override; - string add_link(const atoms::Link* link, bool throw_if_exists = false) override; + string add_atom(const atoms::Atom* atom, const atoms::Merger* merger = NULL) override; + string add_node(const atoms::Node* node, const atoms::Merger* merger = NULL) override; + string add_link(const atoms::Link* link, const atoms::Merger* merger = NULL) override; - vector add_atoms(const vector& atoms, - bool throw_if_exists = false, - bool is_transactional = false) override; + vector add_atoms(const vector& atom_list, + bool is_transactional = false, + const atoms::Merger* merger = NULL) override; vector add_nodes(const vector& nodes, - bool throw_if_exists = false, - bool is_transactional = false) override; + bool is_transactional = false, + const atoms::Merger* merger = NULL) override; vector add_links(const vector& links, - bool throw_if_exists = false, - bool is_transactional = false) override; + bool is_transactional = false, + const atoms::Merger* merger = NULL) override; bool delete_atom(const string& handle, bool delete_link_targets = false) override; bool delete_node(const string& handle, bool delete_link_targets = false) override; diff --git a/src/atomdb/inmemorydb/InMemoryDB.cc b/src/atomdb/inmemorydb/InMemoryDB.cc index 9f0c8e843..d1b77dbd2 100644 --- a/src/atomdb/inmemorydb/InMemoryDB.cc +++ b/src/atomdb/inmemorydb/InMemoryDB.cc @@ -7,6 +7,7 @@ #include "InMemoryDBAPITypes.h" #include "Link.h" #include "LinkSchema.h" +#include "Merger.h" #include "Node.h" #include "Utils.h" @@ -289,52 +290,53 @@ set InMemoryDB::links_exist(const vector& handles) { return existing; } -string InMemoryDB::add_atom(const atoms::Atom* atom, bool throw_if_exists) { +string InMemoryDB::add_atom(const atoms::Atom* atom, const atoms::Merger* merger) { if (atom->arity() == 0) { - return add_node(dynamic_cast(atom), throw_if_exists); + return add_node(dynamic_cast(atom), merger); } else { - return add_link(dynamic_cast(atom), throw_if_exists); + return add_link(dynamic_cast(atom), merger); } } -string InMemoryDB::add_node(const atoms::Node* node, bool throw_if_exists) { +string InMemoryDB::add_node(const atoms::Node* node, const atoms::Merger* merger) { string handle = node->handle(); - if (throw_if_exists && this->node_exists(handle)) { - RAISE_ERROR("Node already exists: " + handle); - return ""; - } - - // Check if already exists auto existing = atoms_trie_->lookup(handle); - if (existing != NULL && !throw_if_exists) { - return handle; // Already exists, return handle + if ((existing == NULL) || (merger == NULL)) { + // Insert or upsert/replace — HandleTrie insert calls AtomTrieValue::merge, + // which deletes the previous Atom (if any) and takes ownership of the new one. + Node* cloned_node = new Node(*node); + atoms_trie_->insert(handle, new AtomTrieValue(cloned_node)); + return handle; + } + + // Merge a copy; persist only when merge() returns true. + auto* atom_trie_value = dynamic_cast(existing); + unique_ptr working(new Node(*dynamic_cast(atom_trie_value->get_atom()))); + if (!merger->merge(working.get(), node)) { + return ""; } - - // Clone the node to store in trie - Node* cloned_node = new Node(*node); - auto atom_trie_value = new AtomTrieValue(cloned_node); - atoms_trie_->insert(handle, atom_trie_value); + atoms_trie_->insert(handle, new AtomTrieValue(working.release())); return handle; } -string InMemoryDB::add_link(const atoms::Link* link, bool throw_if_exists) { +string InMemoryDB::add_link(const atoms::Link* link, const atoms::Merger* merger) { vector links = {const_cast(link)}; - auto handles = this->add_links(links, throw_if_exists, false); + auto handles = this->add_links(links, false, merger); return handles.empty() ? "" : handles[0]; } -vector InMemoryDB::add_atoms(const vector& atoms, - bool throw_if_exists, - bool is_transactional) { - if (atoms.empty()) { +vector InMemoryDB::add_atoms(const vector& atom_list, + bool is_transactional, + const atoms::Merger* merger) { + if (atom_list.empty()) { return {}; } vector nodes; vector links; - for (const auto& atom : atoms) { + for (const auto& atom : atom_list) { LOG_DEBUG("Adding atom: " + atom->to_string()); if (atom->arity() == 0) { nodes.push_back(dynamic_cast(atom)); @@ -342,89 +344,71 @@ vector InMemoryDB::add_atoms(const vector& atoms, links.push_back(dynamic_cast(atom)); } } - auto node_handles = this->add_nodes(nodes, throw_if_exists, is_transactional); - auto link_handles = this->add_links(links, throw_if_exists, is_transactional); + auto node_handles = this->add_nodes(nodes, is_transactional, merger); + auto link_handles = this->add_links(links, is_transactional, merger); node_handles.insert(node_handles.end(), link_handles.begin(), link_handles.end()); return node_handles; } vector InMemoryDB::add_nodes(const vector& nodes, - bool throw_if_exists, - bool is_transactional) { + bool is_transactional, + const atoms::Merger* merger) { if (nodes.empty()) { return {}; } vector handles; + handles.reserve(nodes.size()); for (const auto& node : nodes) { - handles.push_back(node->handle()); + handles.push_back(this->add_node(node, merger)); } - - if (throw_if_exists) { - auto existing_handles = this->nodes_exist(handles); - if (!existing_handles.empty()) { - vector existing_handles_vector(existing_handles.begin(), existing_handles.end()); - RAISE_ERROR("Failed to insert nodes, some nodes already exist: " + - Utils::join(existing_handles_vector, ',')); - return {}; - } - } - - for (const auto& node : nodes) { - handles.push_back(this->add_node(node, throw_if_exists)); - } - return handles; } vector InMemoryDB::add_links(const vector& links, - bool throw_if_exists, - bool is_transactional) { + bool is_transactional, + const atoms::Merger* merger) { if (links.empty()) { return {}; } - if (throw_if_exists) { - vector handles; - for (const auto& link : links) { - handles.push_back(link->handle()); - } - auto existing_handles = this->links_exist(handles); - if (!existing_handles.empty()) { - vector existing_handles_vector(existing_handles.begin(), existing_handles.end()); - RAISE_ERROR("Failed to insert links, some links already exist: " + - Utils::join(existing_handles_vector, ',')); - return {}; - } - } - vector handles; + handles.reserve(links.size()); + for (const auto& link : links) { string link_handle = link->handle(); - handles.push_back(link_handle); - // Check if already exists auto existing = atoms_trie_->lookup(link_handle); - if (existing == NULL || !throw_if_exists) { - if (existing == NULL) { - // Clone the link to store in trie - Link* cloned_link = new Link(*link); - auto atom_trie_value = new AtomTrieValue(cloned_link); - atoms_trie_->insert(link_handle, atom_trie_value); + if ((existing == NULL) || (merger == NULL)) { + // Insert or upsert/replace — AtomTrieValue::merge frees the previous Atom. + Link* cloned_link = new Link(*link); + atoms_trie_->insert(link_handle, new AtomTrieValue(cloned_link)); + } else { + // Merge a copy; persist only when merge() returns true. + // On failure, skip incoming-set/pattern updates — the link was already + // indexed by whichever add created it. + auto* atom_trie_value = dynamic_cast(existing); + unique_ptr working(new Link(*dynamic_cast(atom_trie_value->get_atom()))); + if (!merger->merge(working.get(), link)) { + handles.push_back(""); + continue; } + atoms_trie_->insert(link_handle, new AtomTrieValue(working.release())); + } - // Update incoming sets for each target - for (const auto& target_handle : link->targets) { - this->add_incoming_set(target_handle, link_handle); - } + // Update incoming sets for each target + for (const auto& target_handle : link->targets) { + this->add_incoming_set(target_handle, link_handle); + } - // Index pattern - auto pattern_handles = this->match_pattern_index_schema(link); - for (const auto& pattern_handle : pattern_handles) { - this->add_pattern(pattern_handle, link_handle); - } + // Index pattern + auto pattern_handles = this->match_pattern_index_schema(link); + for (const auto& pattern_handle : pattern_handles) { + this->add_pattern(pattern_handle, link_handle); } + + handles.push_back(link_handle); } return handles; diff --git a/src/atomdb/inmemorydb/InMemoryDB.h b/src/atomdb/inmemorydb/InMemoryDB.h index cc683d1cb..174734432 100644 --- a/src/atomdb/inmemorydb/InMemoryDB.h +++ b/src/atomdb/inmemorydb/InMemoryDB.h @@ -44,19 +44,19 @@ class InMemoryDB : public AtomDB { set nodes_exist(const vector& handles) override; set links_exist(const vector& handles) override; - string add_atom(const atoms::Atom* atom, bool throw_if_exists = false) override; - string add_node(const atoms::Node* node, bool throw_if_exists = false) override; - string add_link(const atoms::Link* link, bool throw_if_exists = false) override; + string add_atom(const atoms::Atom* atom, const atoms::Merger* merger = NULL) override; + string add_node(const atoms::Node* node, const atoms::Merger* merger = NULL) override; + string add_link(const atoms::Link* link, const atoms::Merger* merger = NULL) override; - vector add_atoms(const vector& atoms, - bool throw_if_exists = false, - bool is_transactional = false) override; + vector add_atoms(const vector& atom_list, + bool is_transactional = false, + const atoms::Merger* merger = NULL) override; vector add_nodes(const vector& nodes, - bool throw_if_exists = false, - bool is_transactional = false) override; + bool is_transactional = false, + const atoms::Merger* merger = NULL) override; vector add_links(const vector& links, - bool throw_if_exists = false, - bool is_transactional = false) override; + bool is_transactional = false, + const atoms::Merger* merger = NULL) override; bool delete_atom(const string& handle, bool delete_link_targets = false) override; bool delete_node(const string& handle, bool delete_link_targets = false) override; diff --git a/src/atomdb/morkdb/MorkDB.cc b/src/atomdb/morkdb/MorkDB.cc index 9033e05b0..f7bde3d1e 100644 --- a/src/atomdb/morkdb/MorkDB.cc +++ b/src/atomdb/morkdb/MorkDB.cc @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -12,6 +13,7 @@ #include "HandleDecoder.h" #include "Hasher.h" #include "Logger.h" +#include "Merger.h" #include "MettaMapping.h" #include "MettaParser.h" #include "MettaParserActions.h" @@ -215,8 +217,8 @@ shared_ptr MorkDB::query_for_targets(const string& } vector MorkDB::add_links(const vector& links, - bool throw_if_exists, - bool is_transactional) { + bool is_transactional, + const atoms::Merger* merger) { if (links.empty()) { if (this->composite_type_enabled() && is_transactional) { lock_guard composite_type_hashes_map_lock(this->composite_type_hashes_map_mutex); @@ -225,41 +227,85 @@ vector MorkDB::add_links(const vector& links, return {}; } - if (throw_if_exists) { - vector handles; - for (const auto& link : links) { - handles.push_back(link->handle()); - } - auto existing_handles = this->links_exist(handles); - if (!existing_handles.empty()) { - vector existing_handles_vector(existing_handles.begin(), existing_handles.end()); - RAISE_ERROR("Failed to insert links, some links already exist: " + - Utils::join(existing_handles_vector, ',')); - return {}; - } - } - - map> composite_type_entries_map; - map composite_type_hashes_map_copy; - if (this->composite_type_enabled() && is_transactional) { - this->build_composite_type_entries_map(links, composite_type_entries_map); - lock_guard composite_type_hashes_map_lock(this->composite_type_hashes_map_mutex); - composite_type_hashes_map_copy = this->composite_type_hashes_map; - } else if (!is_transactional) { + if (!is_transactional) { this->check_existing_targets(links); } vector handles; + handles.reserve(links.size()); string metta_expressions; vector documents; + map> batch_merged; + vector unique_handles; + // Non-const: metta_expression may be filled in before persist (including caller's Link*). + vector links_to_persist; + vector> composite_keepalive; + vector links_for_composite; - uint count = 0; for (const auto& link : links) { auto link_handle = link->handle(); - string metta_expression = link->custom_attributes.get_or("metta_expression", ""); + + if (merger != NULL) { + auto it = batch_merged.find(link_handle); + if (it != batch_merged.end()) { + shared_ptr candidate = make_shared(*it->second); + if (merger->merge(candidate.get(), link)) { + it->second = candidate; + handles.push_back(link_handle); + } else { + handles.push_back(""); + } + } else { + shared_ptr working; + auto existing_link = get_link(link_handle); + if (existing_link != nullptr) { + if (!merger->merge(existing_link.get(), link)) { + // Do not persist, but keep existing in composite-type bookkeeping. + composite_keepalive.push_back(existing_link); + links_for_composite.push_back(existing_link.get()); + handles.push_back(""); + continue; + } + working = existing_link; + } else { + working = make_shared(*link); + } + batch_merged[link_handle] = working; + unique_handles.push_back(link_handle); + composite_keepalive.push_back(working); + links_for_composite.push_back(working.get()); + handles.push_back(link_handle); + } + } else { + links_to_persist.push_back(link); + links_for_composite.push_back(link); + handles.push_back(link_handle); + } + } + + if (merger != NULL) { + for (const auto& link_handle : unique_handles) { + links_to_persist.push_back(batch_merged[link_handle].get()); + } + } + + // Derive transactional composite-type metadata from bookkeeping links (final merged + // objects plus rejected-but-existing ones), not from the raw input batch alone. + map> composite_type_entries_map; + if (this->composite_type_enabled() && is_transactional) { + this->build_composite_type_entries_map(links_for_composite, composite_type_entries_map); + } + + uint count = 0; + for (auto* to_store : links_to_persist) { + auto link_handle = to_store->handle(); + + string metta_expression = to_store->custom_attributes.get_or("metta_expression", ""); if (metta_expression.empty()) { - metta_expression = link->metta_representation(*this); - link->custom_attributes["metta_expression"] = metta_expression; + metta_expression = to_store->metta_representation(*this); + // Persist metta on the atom we will store (may be incoming or merged existing). + // Callers whose Link* is reused here should expect metta_expression to be filled in. + to_store->custom_attributes["metta_expression"] = metta_expression; } metta_expressions += metta_expression + "\n"; @@ -273,18 +319,16 @@ vector MorkDB::add_links(const vector& links, optional mongodb_doc; if (!this->composite_type_enabled()) { static const vector empty_composite_type; - mongodb_doc.emplace(link, "", empty_composite_type, false); + mongodb_doc.emplace(to_store, "", empty_composite_type, false); } else if (is_transactional) { - mongodb_doc.emplace(link, - composite_type_hashes_map_copy[link_handle], - composite_type_entries_map[link_handle]); + string composite_type_hash = + Hasher::composite_handle(composite_type_entries_map[link_handle]); + mongodb_doc.emplace(to_store, composite_type_hash, composite_type_entries_map[link_handle]); } else { - mongodb_doc.emplace(link, *this); + mongodb_doc.emplace(to_store, *this); } documents.push_back(mongodb_doc->value()); - - handles.push_back(link_handle); } if (!documents.empty()) { @@ -350,7 +394,7 @@ void MorkDB::re_index_patterns(bool flush_patterns) { } } - this->add_links(links, false, true); + this->add_links(links, true); } // <-- diff --git a/src/atomdb/morkdb/MorkDB.h b/src/atomdb/morkdb/MorkDB.h index 76be9056b..5eacf7c48 100644 --- a/src/atomdb/morkdb/MorkDB.h +++ b/src/atomdb/morkdb/MorkDB.h @@ -49,8 +49,8 @@ class MorkDB : public RedisMongoDB { shared_ptr query_for_targets(const string& handle) override; vector add_links(const vector& links, - bool throw_if_exists = false, - bool is_transactional = false) override; + bool is_transactional = false, + const atoms::Merger* merger = NULL) override; // TODO: Implement this once MORK supports deleting links (S-Expressions) bool delete_link(const string& handle, bool delete_targets) override; diff --git a/src/atomdb/redis_mongodb/RedisMongoDB.cc b/src/atomdb/redis_mongodb/RedisMongoDB.cc index 24f5d890b..80e653ea0 100644 --- a/src/atomdb/redis_mongodb/RedisMongoDB.cc +++ b/src/atomdb/redis_mongodb/RedisMongoDB.cc @@ -6,6 +6,7 @@ #include #include #include +#include #include #include #include @@ -13,6 +14,7 @@ #include "Hasher.h" #include "Link.h" #include "Logger.h" +#include "Merger.h" #include "MongoInitializer.h" #include "Node.h" #include "Properties.h" @@ -715,11 +717,11 @@ set RedisMongoDB::links_exist(const vector& link_handles) { return documents_exist(link_handles, MONGODB_LINKS_COLLECTION_NAME); } -string RedisMongoDB::add_atom(const atoms::Atom* atom, bool throw_if_exists) { +string RedisMongoDB::add_atom(const atoms::Atom* atom, const atoms::Merger* merger) { if (atom->arity() == 0) { - return add_node(dynamic_cast(atom), throw_if_exists); + return add_node(dynamic_cast(atom), merger); } else { - return add_link(dynamic_cast(atom), throw_if_exists); + return add_link(dynamic_cast(atom), merger); } } @@ -802,13 +804,21 @@ uint RedisMongoDB::upsert_documents(const std::vector& return total_modified; } -string RedisMongoDB::add_node(const atoms::Node* node, bool throw_if_exists) { - if (throw_if_exists && node_exists(node->handle())) { - RAISE_ERROR("Node already exists: " + node->handle()); - return ""; +string RedisMongoDB::add_node(const atoms::Node* node, const atoms::Merger* merger) { + const atoms::Node* to_store = node; + shared_ptr working_node; + if (merger != NULL) { + auto existing_node = get_node(node->handle()); + if (existing_node != nullptr) { + if (!merger->merge(existing_node.get(), node)) { + return ""; + } + working_node = existing_node; + to_store = working_node.get(); + } } - auto mongodb_doc = atomdb_api_types::MongodbDocument(node); + auto mongodb_doc = atomdb_api_types::MongodbDocument(to_store); if (!this->upsert_document(mongodb_doc.value(), MONGODB_NODES_COLLECTION_NAME)) { RAISE_ERROR("Failed to insert node into MongoDB"); return ""; @@ -816,21 +826,21 @@ string RedisMongoDB::add_node(const atoms::Node* node, bool throw_if_exists) { return node->handle(); } -string RedisMongoDB::add_link(const atoms::Link* link, bool throw_if_exists) { +string RedisMongoDB::add_link(const atoms::Link* link, const atoms::Merger* merger) { vector links = {const_cast(link)}; - return add_links(links, throw_if_exists)[0]; + return add_links(links, false, merger)[0]; } -vector RedisMongoDB::add_atoms(const vector& atoms, - bool throw_if_exists, - bool is_transactional) { - if (atoms.empty()) { +vector RedisMongoDB::add_atoms(const vector& atom_list, + bool is_transactional, + const atoms::Merger* merger) { + if (atom_list.empty()) { return {}; } vector nodes; vector links; - for (const auto& atom : atoms) { + for (const auto& atom : atom_list) { LOG_DEBUG("Adding atom: " + atom->to_string()); if (atom->arity() == 0) { nodes.push_back(dynamic_cast(atom)); @@ -838,38 +848,81 @@ vector RedisMongoDB::add_atoms(const vector& atoms, links.push_back(dynamic_cast(atom)); } } - auto node_handles = add_nodes(nodes, throw_if_exists, is_transactional); - auto link_handles = add_links(links, throw_if_exists, is_transactional); + auto node_handles = add_nodes(nodes, is_transactional, merger); + auto link_handles = add_links(links, is_transactional, merger); node_handles.insert(node_handles.end(), link_handles.begin(), link_handles.end()); return node_handles; } vector RedisMongoDB::add_nodes(const vector& nodes, - bool throw_if_exists, - bool is_transactional) { + bool is_transactional, + const atoms::Merger* merger) { if (nodes.empty()) { return {}; } vector documents; vector handles; + handles.reserve(nodes.size()); + map> batch_merged; + vector unique_handles; for (const auto& node : nodes) { - auto mongodb_doc = atomdb_api_types::MongodbDocument(node); - documents.push_back(mongodb_doc.value()); - handles.push_back(node->handle()); - if (this->composite_type_enabled_ && is_transactional) { - lock_guard composite_type_hashes_map_lock(this->composite_type_hashes_map_mutex); - this->composite_type_hashes_map[node->handle()] = node->named_type_hash(); + if (merger != NULL) { + string handle = node->handle(); + auto it = batch_merged.find(handle); + if (it != batch_merged.end()) { + shared_ptr candidate = make_shared(*it->second); + if (merger->merge(candidate.get(), node)) { + it->second = candidate; + handles.push_back(handle); + } else { + handles.push_back(""); + } + } else { + shared_ptr working; + auto existing_node = get_node(handle); + if (existing_node != nullptr) { + if (!merger->merge(existing_node.get(), node)) { + // Do not persist, but keep existing in the transactional + // composite-type map so later links can resolve this target. + if (this->composite_type_enabled_ && is_transactional) { + lock_guard composite_type_hashes_map_lock( + this->composite_type_hashes_map_mutex); + this->composite_type_hashes_map[handle] = existing_node->named_type_hash(); + } + handles.push_back(""); + continue; + } + working = existing_node; + } else { + working = make_shared(*node); + } + batch_merged[handle] = working; + unique_handles.push_back(handle); + handles.push_back(handle); + } + } else { + auto mongodb_doc = atomdb_api_types::MongodbDocument(node); + documents.push_back(mongodb_doc.value()); + handles.push_back(node->handle()); + if (this->composite_type_enabled_ && is_transactional) { + lock_guard composite_type_hashes_map_lock(this->composite_type_hashes_map_mutex); + this->composite_type_hashes_map[node->handle()] = node->named_type_hash(); + } } } - if (throw_if_exists) { - auto existing_handles = this->nodes_exist(handles); - if (existing_handles.size() > 0) { - RAISE_ERROR("Failed to insert nodes, some nodes already exist."); - return {}; + if (merger != NULL) { + for (const auto& handle : unique_handles) { + const atoms::Node* to_store = batch_merged[handle].get(); + auto mongodb_doc = atomdb_api_types::MongodbDocument(to_store); + documents.push_back(mongodb_doc.value()); + if (this->composite_type_enabled_ && is_transactional) { + lock_guard composite_type_hashes_map_lock(this->composite_type_hashes_map_mutex); + this->composite_type_hashes_map[handle] = to_store->named_type_hash(); + } } } @@ -881,8 +934,8 @@ vector RedisMongoDB::add_nodes(const vector& nodes, } vector RedisMongoDB::add_links(const vector& links, - bool throw_if_exists, - bool is_transactional) { + bool is_transactional, + const atoms::Merger* merger) { if (links.empty()) { if (this->composite_type_enabled_ && is_transactional) { lock_guard composite_type_hashes_map_lock(this->composite_type_hashes_map_mutex); @@ -891,38 +944,83 @@ vector RedisMongoDB::add_links(const vector& links, return {}; } - if (throw_if_exists) { - vector handles; - for (const auto& link : links) { - handles.push_back(link->handle()); + if (!is_transactional) { + this->check_existing_targets(links); + } + + vector handles; + handles.reserve(links.size()); + vector documents; + map> batch_merged; + vector unique_handles; + vector links_to_persist; + // Links that contribute to transactional composite-type bookkeeping, in encounter + // order. Includes rejected-but-existing links (not persisted) so later links in the + // same batch can resolve them as targets. + vector> composite_keepalive; + vector links_for_composite; + + for (const auto& link : links) { + auto link_handle = link->handle(); + + if (merger != NULL) { + auto it = batch_merged.find(link_handle); + if (it != batch_merged.end()) { + shared_ptr candidate = make_shared(*it->second); + if (merger->merge(candidate.get(), link)) { + it->second = candidate; + handles.push_back(link_handle); + } else { + handles.push_back(""); + } + } else { + shared_ptr working; + auto existing_link = get_link(link_handle); + if (existing_link != nullptr) { + if (!merger->merge(existing_link.get(), link)) { + // Do not persist, but keep existing in composite-type bookkeeping. + composite_keepalive.push_back(existing_link); + links_for_composite.push_back(existing_link.get()); + handles.push_back(""); + continue; + } + working = existing_link; + } else { + working = make_shared(*link); + } + batch_merged[link_handle] = working; + unique_handles.push_back(link_handle); + composite_keepalive.push_back(working); + links_for_composite.push_back(working.get()); + handles.push_back(link_handle); + } + } else { + links_to_persist.push_back(link); + links_for_composite.push_back(link); + handles.push_back(link_handle); } - auto existing_handles = this->links_exist(handles); - if (!existing_handles.empty()) { - vector existing_handles_vector(existing_handles.begin(), existing_handles.end()); - RAISE_ERROR("Failed to insert links, some links already exist: " + - Utils::join(existing_handles_vector, ',')); - return {}; + } + + if (merger != NULL) { + for (const auto& link_handle : unique_handles) { + links_to_persist.push_back(batch_merged[link_handle].get()); } } + // Derive transactional composite-type metadata from bookkeeping links (final merged + // objects plus rejected-but-existing ones), not from the raw input batch alone. map> composite_type_entries_map; if (this->composite_type_enabled_ && is_transactional) { - this->build_composite_type_entries_map(links, composite_type_entries_map); - } else if (!is_transactional) { - this->check_existing_targets(links); + this->build_composite_type_entries_map(links_for_composite, composite_type_entries_map); } - vector handles; - vector documents; - shared_ptr ctx = this->redis_pool->acquire(); - for (const auto& link : links) { - auto link_handle = link->handle(); - - auto pattern_handles = match_pattern_index_schema(link); + for (const auto* to_store : links_to_persist) { + auto link_handle = to_store->handle(); + auto pattern_handles = match_pattern_index_schema(to_store); - for (const auto& target : link->targets) { + for (const auto& target : to_store->targets) { string incomming_set_cmd = "ZADD " + REDIS_INCOMING_PREFIX + ":" + target + " " + to_string(this->incoming_set_next_score.load()) + " " + link_handle; @@ -931,7 +1029,7 @@ vector RedisMongoDB::add_links(const vector& links, } string outgoing_set_cmd = "SET " + REDIS_OUTGOING_PREFIX + ":" + link_handle + " "; - for (const auto& outgoing_handle : link->targets) { + for (const auto& outgoing_handle : to_store->targets) { outgoing_set_cmd += outgoing_handle; } ctx->append_command(outgoing_set_cmd.c_str()); @@ -946,13 +1044,13 @@ vector RedisMongoDB::add_links(const vector& links, optional mongodb_doc; if (!this->composite_type_enabled_) { static const vector empty_composite_type; - mongodb_doc.emplace(link, "", empty_composite_type, false); + mongodb_doc.emplace(to_store, "", empty_composite_type, false); } else if (is_transactional) { string composite_type_hash = Hasher::composite_handle(composite_type_entries_map[link_handle]); - mongodb_doc.emplace(link, composite_type_hash, composite_type_entries_map[link_handle]); + mongodb_doc.emplace(to_store, composite_type_hash, composite_type_entries_map[link_handle]); } else { - mongodb_doc.emplace(link, *this); + mongodb_doc.emplace(to_store, *this); } if (static_cast(ctx->get_pending_commands_count()) >= REDIS_CHUNK_SIZE) { @@ -962,8 +1060,6 @@ vector RedisMongoDB::add_links(const vector& links, } documents.push_back(mongodb_doc->value()); - - handles.push_back(link_handle); } if (!documents.empty()) { diff --git a/src/atomdb/redis_mongodb/RedisMongoDB.h b/src/atomdb/redis_mongodb/RedisMongoDB.h index 9abd196d4..b90664bf3 100644 --- a/src/atomdb/redis_mongodb/RedisMongoDB.h +++ b/src/atomdb/redis_mongodb/RedisMongoDB.h @@ -93,19 +93,19 @@ class RedisMongoDB : public AtomDB { set nodes_exist(const vector& handles); set links_exist(const vector& handles); - string add_atom(const atoms::Atom* atom, bool throw_if_exists = false); - string add_node(const atoms::Node* node, bool throw_if_exists = false); - string add_link(const atoms::Link* link, bool throw_if_exists = false); + string add_atom(const atoms::Atom* atom, const atoms::Merger* merger = NULL); + string add_node(const atoms::Node* node, const atoms::Merger* merger = NULL); + string add_link(const atoms::Link* link, const atoms::Merger* merger = NULL); - vector add_atoms(const vector& atoms, - bool throw_if_exists = false, - bool is_transactional = false); + vector add_atoms(const vector& atom_list, + bool is_transactional = false, + const atoms::Merger* merger = NULL); vector add_nodes(const vector& nodes, - bool throw_if_exists = false, - bool is_transactional = false); + bool is_transactional = false, + const atoms::Merger* merger = NULL); vector add_links(const vector& links, - bool throw_if_exists = false, - bool is_transactional = false); + bool is_transactional = false, + const atoms::Merger* merger = NULL); bool delete_atom(const string& handle, bool delete_link_targets = false); bool delete_node(const string& handle, bool delete_link_targets = false); diff --git a/src/atomdb/remotedb/RemoteAtomDB.cc b/src/atomdb/remotedb/RemoteAtomDB.cc index 82d9569b7..1e4806613 100644 --- a/src/atomdb/remotedb/RemoteAtomDB.cc +++ b/src/atomdb/remotedb/RemoteAtomDB.cc @@ -79,7 +79,7 @@ RemoteAtomDB::~RemoteAtomDB() = default; bool RemoteAtomDB::composite_type_enabled() const { LOG_ERROR( "RemoteAtomDB derives composite_type_enabled() from peers (true if any peer has it enabled)"); - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { if (peer->composite_type_enabled()) { return true; } @@ -92,7 +92,7 @@ void RemoteAtomDB::derive_nested_indexing() { // cannot describe a heterogeneous result set, so mixed configurations are normalized to the // lowest common denominator (false: the query engine re-matches every handle locally). unsigned int nested_peers = 0; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { if (peer->allow_nested_indexing()) nested_peers++; } if (!remote_db_.empty() && nested_peers == remote_db_.size()) { @@ -114,12 +114,12 @@ bool RemoteAtomDB::allow_nested_indexing() { return nested_indexing_; } shared_ptr RemoteAtomDB::get_atom(const string& handle) { // Phase 1: probe every peer's in-memory cache first (no network). Silent: this is the hot path. - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { auto atom = peer->get_cached_atom(handle); if (atom) return atom; } // Phase 2: escalate to peers (local_persistence + remote backend) only when no cache has it. - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { auto atom = peer->get_atom(handle); if (atom) { LOG_DEBUG("get_atom(" << handle << ") fetched from [" << uid << "]"); @@ -131,11 +131,11 @@ shared_ptr RemoteAtomDB::get_atom(const string& handle) { } shared_ptr RemoteAtomDB::get_node(const string& handle) { - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { auto node = peer->get_cached_node(handle); if (node) return node; } - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { auto node = peer->get_node(handle); if (node) { LOG_DEBUG("get_node(" << handle << ") fetched from [" << uid << "]"); @@ -147,11 +147,11 @@ shared_ptr RemoteAtomDB::get_node(const string& handle) { } shared_ptr RemoteAtomDB::get_link(const string& handle) { - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { auto link = peer->get_cached_link(handle); if (link) return link; } - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { auto link = peer->get_link(handle); if (link) { LOG_DEBUG("get_link(" << handle << ") fetched from [" << uid << "]"); @@ -166,7 +166,7 @@ vector> RemoteAtomDB::get_matching_atoms(bool is_toplevel, Atom vector> result; set seen; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { auto atoms = peer->get_matching_atoms(is_toplevel, key); for (const auto& atom : atoms) { string h = atom->handle(); @@ -185,7 +185,7 @@ shared_ptr RemoteAtomDB::query_for_pattern(const Li LOG_DEBUG("query_for_pattern(" << link_schema.handle() << ") fan-out to " << remote_db_.size() << " peers"); - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { auto handle_set = peer->query_for_pattern(link_schema); if (!handle_set) continue; @@ -219,7 +219,7 @@ shared_ptr RemoteAtomDB::query_for_pattern(const Li } shared_ptr RemoteAtomDB::query_for_targets(const string& handle) { - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { auto list = peer->query_for_targets(handle); if (list) { LOG_DEBUG("query_for_targets(" << handle << ") served by peer [" << uid << "]"); @@ -235,7 +235,7 @@ shared_ptr RemoteAtomDB::query_for_incoming_set(con set seen; LOG_DEBUG("query_for_incoming_set(" << handle << ") fan-out to " << remote_db_.size() << " peers"); - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { auto handle_set = peer->query_for_incoming_set(handle); if (!handle_set) continue; @@ -257,21 +257,21 @@ shared_ptr RemoteAtomDB::query_for_incoming_set(con } bool RemoteAtomDB::atom_exists(const string& handle) { - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { if (peer->atom_exists(handle)) return true; } return false; } bool RemoteAtomDB::node_exists(const string& handle) { - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { if (peer->node_exists(handle)) return true; } return false; } bool RemoteAtomDB::link_exists(const string& handle) { - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { if (peer->link_exists(handle)) return true; } return false; @@ -281,7 +281,7 @@ set RemoteAtomDB::atoms_exist(const vector& handles) { set result; set remaining(handles.begin(), handles.end()); - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { if (remaining.empty()) break; vector to_check(remaining.begin(), remaining.end()); auto found = peer->atoms_exist(to_check); @@ -297,7 +297,7 @@ set RemoteAtomDB::nodes_exist(const vector& handles) { set result; set remaining(handles.begin(), handles.end()); - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { if (remaining.empty()) break; vector to_check(remaining.begin(), remaining.end()); auto found = peer->nodes_exist(to_check); @@ -313,7 +313,7 @@ set RemoteAtomDB::links_exist(const vector& handles) { set result; set remaining(handles.begin(), handles.end()); - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { if (remaining.empty()) break; vector to_check(remaining.begin(), remaining.end()); auto found = peer->links_exist(to_check); @@ -325,69 +325,69 @@ set RemoteAtomDB::links_exist(const vector& handles) { return result; } -string RemoteAtomDB::add_atom(const atoms::Atom* atom, bool throw_if_exists) { +string RemoteAtomDB::add_atom(const atoms::Atom* atom, const atoms::Merger* merger) { string handle; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { LOG_DEBUG("add_atom(" << atom->handle() << ") to peer [" << uid << "]"); - handle = peer->add_atom(atom, throw_if_exists); + handle = peer->add_atom(atom, merger); } return handle; } -string RemoteAtomDB::add_node(const atoms::Node* node, bool throw_if_exists) { +string RemoteAtomDB::add_node(const atoms::Node* node, const atoms::Merger* merger) { string handle; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { LOG_DEBUG("add_node(" << node->handle() << ") to peer [" << uid << "]"); - handle = peer->add_node(node, throw_if_exists); + handle = peer->add_node(node, merger); } return handle; } -string RemoteAtomDB::add_link(const atoms::Link* link, bool throw_if_exists) { +string RemoteAtomDB::add_link(const atoms::Link* link, const atoms::Merger* merger) { string handle; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { LOG_DEBUG("add_link(" << link->handle() << ") to peer [" << uid << "]"); - handle = peer->add_link(link, throw_if_exists); + handle = peer->add_link(link, merger); } return handle; } -vector RemoteAtomDB::add_atoms(const vector& atoms, - bool throw_if_exists, - bool is_transactional) { +vector RemoteAtomDB::add_atoms(const vector& atom_list, + bool is_transactional, + const atoms::Merger* merger) { vector handles; - for (auto& [uid, peer] : remote_db_) { - LOG_DEBUG("add_atoms(" << atoms.size() << ") to peer [" << uid << "]"); - handles = peer->add_atoms(atoms, throw_if_exists, is_transactional); + for (auto& [uid, peer] : this->remote_db_) { + LOG_DEBUG("add_atoms(" << atom_list.size() << ") to peer [" << uid << "]"); + handles = peer->add_atoms(atom_list, is_transactional, merger); } return handles; } vector RemoteAtomDB::add_nodes(const vector& nodes, - bool throw_if_exists, - bool is_transactional) { + bool is_transactional, + const atoms::Merger* merger) { vector handles; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { LOG_DEBUG("add_nodes(" << nodes.size() << ") to peer [" << uid << "]"); - handles = peer->add_nodes(nodes, throw_if_exists, is_transactional); + handles = peer->add_nodes(nodes, is_transactional, merger); } return handles; } vector RemoteAtomDB::add_links(const vector& links, - bool throw_if_exists, - bool is_transactional) { + bool is_transactional, + const atoms::Merger* merger) { vector handles; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { LOG_DEBUG("add_links(" << links.size() << ") to peer [" << uid << "]"); - handles = peer->add_links(links, throw_if_exists, is_transactional); + handles = peer->add_links(links, is_transactional, merger); } return handles; } bool RemoteAtomDB::delete_atom(const string& handle, bool delete_link_targets) { bool ok = true; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { LOG_DEBUG("delete_atom(" << handle << ") from peer [" << uid << "]"); ok = peer->delete_atom(handle, delete_link_targets) && ok; } @@ -396,7 +396,7 @@ bool RemoteAtomDB::delete_atom(const string& handle, bool delete_link_targets) { bool RemoteAtomDB::delete_node(const string& handle, bool delete_link_targets) { bool ok = true; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { LOG_DEBUG("delete_node(" << handle << ") from peer [" << uid << "]"); ok = peer->delete_node(handle, delete_link_targets) && ok; } @@ -405,7 +405,7 @@ bool RemoteAtomDB::delete_node(const string& handle, bool delete_link_targets) { bool RemoteAtomDB::delete_link(const string& handle, bool delete_link_targets) { bool ok = true; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { LOG_DEBUG("delete_link(" << handle << ") from peer [" << uid << "]"); ok = peer->delete_link(handle, delete_link_targets) && ok; } @@ -414,7 +414,7 @@ bool RemoteAtomDB::delete_link(const string& handle, bool delete_link_targets) { uint RemoteAtomDB::delete_atoms(const vector& handles, bool delete_link_targets) { uint count = 0; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { LOG_DEBUG("delete_atoms(" << handles.size() << ") from peer [" << uid << "]"); count = peer->delete_atoms(handles, delete_link_targets); } @@ -423,7 +423,7 @@ uint RemoteAtomDB::delete_atoms(const vector& handles, bool delete_link_ uint RemoteAtomDB::delete_nodes(const vector& handles, bool delete_link_targets) { uint count = 0; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { LOG_DEBUG("delete_nodes(" << handles.size() << ") from peer [" << uid << "]"); count = peer->delete_nodes(handles, delete_link_targets); } @@ -432,7 +432,7 @@ uint RemoteAtomDB::delete_nodes(const vector& handles, bool delete_link_ uint RemoteAtomDB::delete_links(const vector& handles, bool delete_link_targets) { uint count = 0; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { LOG_DEBUG("delete_links(" << handles.size() << ") from peer [" << uid << "]"); count = peer->delete_links(handles, delete_link_targets); } @@ -440,7 +440,7 @@ uint RemoteAtomDB::delete_links(const vector& handles, bool delete_link_ } void RemoteAtomDB::re_index_patterns(bool flush_patterns) { - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { LOG_DEBUG("re_index_patterns(" << flush_patterns << ") from peer [" << uid << "]"); peer->re_index_patterns(flush_patterns); } @@ -448,7 +448,7 @@ void RemoteAtomDB::re_index_patterns(bool flush_patterns) { size_t RemoteAtomDB::node_count() const { size_t count = 0; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { count += peer->node_count(); } return count; @@ -456,7 +456,7 @@ size_t RemoteAtomDB::node_count() const { size_t RemoteAtomDB::link_count() const { size_t count = 0; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { count += peer->link_count(); } return count; @@ -464,7 +464,7 @@ size_t RemoteAtomDB::link_count() const { size_t RemoteAtomDB::atom_count() const { size_t count = 0; - for (auto& [uid, peer] : remote_db_) { + for (auto& [uid, peer] : this->remote_db_) { count += peer->atom_count(); } return count; diff --git a/src/atomdb/remotedb/RemoteAtomDB.h b/src/atomdb/remotedb/RemoteAtomDB.h index b63278a94..d1de2218f 100644 --- a/src/atomdb/remotedb/RemoteAtomDB.h +++ b/src/atomdb/remotedb/RemoteAtomDB.h @@ -48,19 +48,19 @@ class RemoteAtomDB : public AtomDB { set nodes_exist(const vector& handles) override; set links_exist(const vector& handles) override; - string add_atom(const atoms::Atom* atom, bool throw_if_exists = false) override; - string add_node(const atoms::Node* node, bool throw_if_exists = false) override; - string add_link(const atoms::Link* link, bool throw_if_exists = false) override; + string add_atom(const atoms::Atom* atom, const atoms::Merger* merger = NULL) override; + string add_node(const atoms::Node* node, const atoms::Merger* merger = NULL) override; + string add_link(const atoms::Link* link, const atoms::Merger* merger = NULL) override; - vector add_atoms(const vector& atoms, - bool throw_if_exists = false, - bool is_transactional = false) override; + vector add_atoms(const vector& atom_list, + bool is_transactional = false, + const atoms::Merger* merger = NULL) override; vector add_nodes(const vector& nodes, - bool throw_if_exists = false, - bool is_transactional = false) override; + bool is_transactional = false, + const atoms::Merger* merger = NULL) override; vector add_links(const vector& links, - bool throw_if_exists = false, - bool is_transactional = false) override; + bool is_transactional = false, + const atoms::Merger* merger = NULL) override; bool delete_atom(const string& handle, bool delete_link_targets = false) override; bool delete_node(const string& handle, bool delete_link_targets = false) override; diff --git a/src/atomdb/remotedb/RemoteAtomDBPeer.cc b/src/atomdb/remotedb/RemoteAtomDBPeer.cc index 4b68df11d..669d8f272 100644 --- a/src/atomdb/remotedb/RemoteAtomDBPeer.cc +++ b/src/atomdb/remotedb/RemoteAtomDBPeer.cc @@ -366,34 +366,34 @@ set RemoteAtomDBPeer::links_exist(const vector& handles) { return result; } -string RemoteAtomDBPeer::add_atom(const atoms::Atom* atom, bool throw_if_exists) { - return cache_.add_atom(atom, throw_if_exists); +string RemoteAtomDBPeer::add_atom(const atoms::Atom* atom, const atoms::Merger* merger) { + return cache_.add_atom(atom, merger); } -string RemoteAtomDBPeer::add_node(const atoms::Node* node, bool throw_if_exists) { - return cache_.add_node(node, throw_if_exists); +string RemoteAtomDBPeer::add_node(const atoms::Node* node, const atoms::Merger* merger) { + return cache_.add_node(node, merger); } -string RemoteAtomDBPeer::add_link(const atoms::Link* link, bool throw_if_exists) { - return cache_.add_link(link, throw_if_exists); +string RemoteAtomDBPeer::add_link(const atoms::Link* link, const atoms::Merger* merger) { + return cache_.add_link(link, merger); } -vector RemoteAtomDBPeer::add_atoms(const vector& atoms, - bool throw_if_exists, - bool is_transactional) { - return cache_.add_atoms(atoms, throw_if_exists, is_transactional); +vector RemoteAtomDBPeer::add_atoms(const vector& atom_list, + bool is_transactional, + const atoms::Merger* merger) { + return cache_.add_atoms(atom_list, is_transactional, merger); } vector RemoteAtomDBPeer::add_nodes(const vector& nodes, - bool throw_if_exists, - bool is_transactional) { - return cache_.add_nodes(nodes, throw_if_exists, is_transactional); + bool is_transactional, + const atoms::Merger* merger) { + return cache_.add_nodes(nodes, is_transactional, merger); } vector RemoteAtomDBPeer::add_links(const vector& links, - bool throw_if_exists, - bool is_transactional) { - return cache_.add_links(links, throw_if_exists, is_transactional); + bool is_transactional, + const atoms::Merger* merger) { + return cache_.add_links(links, is_transactional, merger); } bool RemoteAtomDBPeer::delete_atom(const string& handle, bool delete_link_targets) { @@ -510,7 +510,7 @@ void RemoteAtomDBPeer::release(const LinkSchema& link_schema) { string handle(handle_cstr); auto atom = cache_.get_atom(handle); if (atom) { - local_persistence_->add_atom(atom.get(), false); + local_persistence_->add_atom(atom.get()); cache_.delete_atom(handle, false); } } diff --git a/src/atomdb/remotedb/RemoteAtomDBPeer.h b/src/atomdb/remotedb/RemoteAtomDBPeer.h index 0c5916fea..713c498dc 100644 --- a/src/atomdb/remotedb/RemoteAtomDBPeer.h +++ b/src/atomdb/remotedb/RemoteAtomDBPeer.h @@ -56,19 +56,19 @@ class RemoteAtomDBPeer : public AtomDB, public processor::ThreadMethod { set nodes_exist(const vector& handles) override; set links_exist(const vector& handles) override; - string add_atom(const atoms::Atom* atom, bool throw_if_exists = false) override; - string add_node(const atoms::Node* node, bool throw_if_exists = false) override; - string add_link(const atoms::Link* link, bool throw_if_exists = false) override; + string add_atom(const atoms::Atom* atom, const atoms::Merger* merger = NULL) override; + string add_node(const atoms::Node* node, const atoms::Merger* merger = NULL) override; + string add_link(const atoms::Link* link, const atoms::Merger* merger = NULL) override; - vector add_atoms(const vector& atoms, - bool throw_if_exists = false, - bool is_transactional = false) override; + vector add_atoms(const vector& atom_list, + bool is_transactional = false, + const atoms::Merger* merger = NULL) override; vector add_nodes(const vector& nodes, - bool throw_if_exists = false, - bool is_transactional = false) override; + bool is_transactional = false, + const atoms::Merger* merger = NULL) override; vector add_links(const vector& links, - bool throw_if_exists = false, - bool is_transactional = false) override; + bool is_transactional = false, + const atoms::Merger* merger = NULL) override; bool delete_atom(const string& handle, bool delete_link_targets = false) override; bool delete_node(const string& handle, bool delete_link_targets = false) override; diff --git a/src/commons/atoms/BUILD b/src/commons/atoms/BUILD index 3cde5bc16..eecdec08d 100644 --- a/src/commons/atoms/BUILD +++ b/src/commons/atoms/BUILD @@ -18,6 +18,7 @@ cc_library( "HandleDecoder.h", "Link.h", "LinkSchema.h", + "Merger.h", "MettaParserActions.h", "Node.h", "UntypedVariable.h", diff --git a/src/commons/atoms/Merger.h b/src/commons/atoms/Merger.h new file mode 100644 index 000000000..6bd52f4b1 --- /dev/null +++ b/src/commons/atoms/Merger.h @@ -0,0 +1,81 @@ +#pragma once + +#include "Atom.h" +#include "Utils.h" + +using namespace commons; + +namespace atoms { + +/** + * Strategy for combining an existing stored Atom with an incoming one during add_*. + * + * When AtomDB add_* is called with merger == NULL, the incoming Atom replaces any + * existing one (upsert). When a Merger is provided and the Atom already exists, + * backends merge into a working copy and persist only when merge() returns true. + * + * Return contract: merge() returns true on success and false on failure. When + * false is returned, backends must not persist the working copy (existing stored + * state is left unchanged) and must report that add as "" (single add_* return + * value, or the corresponding index in a batch handles vector). Soft failures do + * not abort the rest of a batch. + * + * Exception contract: if merge() throws (e.g. ThrowIfExistsMerger), that exception + * propagates to the caller unchanged. Backends must not wrap or replace it, and + * must not persist the working copy. Earlier atoms in the same batch may already + * have been applied (no all-or-nothing precheck). + * + * Concurrency: merge-enabled adds are a read-modify-write with no cross-thread + * atomicity. Concurrent merge-enabled adds for the same handle can lose updates. + * Callers must serialize merge-enabled adds per handle (single-writer assumption). + */ +class Merger { + public: + virtual ~Merger() = default; + + /** + * @brief Merge incoming into existing (in place). + * @param existing Working copy of the Atom currently stored in the DB (mutated). + * @param incoming Atom being added (read-only; must not be mutated). + * @return true on success (backends may persist existing); false on failure + * (backends must leave stored state unchanged). + * @throws Propagated unchanged by AtomDB backends (e.g. ThrowIfExistsMerger). + */ + virtual bool merge(Atom* existing, const Atom* incoming) const = 0; +}; + +/** + * Merger that rejects any add when the Atom already exists by throwing + * (replaces throw_if_exists=true). Halts the add_* call via exception; earlier + * items in a batch may already have been applied. + */ +class ThrowIfExistsMerger : public Merger { + public: + bool merge(Atom* existing, const Atom* /*incoming*/) const override { + string atom_type = Atom::is_node(*existing) ? "Node" : "Link"; + RAISE_ERROR(atom_type + " already exists: " + existing->handle()); + return false; + } + + static const ThrowIfExistsMerger& instance() { + static ThrowIfExistsMerger inst; + return inst; + } +}; + +/** + * Merger that skips any add when the Atom already exists (soft reject). + * merge() returns false so backends leave stored state unchanged, report "", + * and continue with the rest of a batch. + */ +class SkipIfExistsMerger : public Merger { + public: + bool merge(Atom* /*existing*/, const Atom* /*incoming*/) const override { return false; } + + static const SkipIfExistsMerger& instance() { + static SkipIfExistsMerger inst; + return inst; + } +}; + +} // namespace atoms diff --git a/src/db_adapter/AtomPersister.cc b/src/db_adapter/AtomPersister.cc index 1accfa5f4..39d022bc0 100644 --- a/src/db_adapter/AtomPersister.cc +++ b/src/db_adapter/AtomPersister.cc @@ -169,7 +169,7 @@ void AtomPersister::send_batch(vector> atoms, for (const auto& atom : atoms) { atom_ptrs.push_back(atom.get()); } - this->atomdb->add_atoms(atom_ptrs, false, true); + this->atomdb->add_atoms(atom_ptrs, true); if (this->is_save_metta()) { for (auto& atom : atoms) { diff --git a/src/main/db_loader.cc b/src/main/db_loader.cc index 4473dddd6..761cc3135 100644 --- a/src/main/db_loader.cc +++ b/src/main/db_loader.cc @@ -175,7 +175,7 @@ int main(int argc, char* argv[]) { thread_atoms_count++; if (batch_atoms.size() >= static_cast(chunk_size)) { - thread_atomdb->add_atoms(batch_atoms, false, true); + thread_atomdb->add_atoms(batch_atoms, true); batch_atoms.clear(); if (parser_actions_list.size() > 10) { parser_actions_list.erase(parser_actions_list.begin(), @@ -190,7 +190,7 @@ int main(int argc, char* argv[]) { } if (!batch_atoms.empty()) { - thread_atomdb->add_atoms(batch_atoms, false, true); + thread_atomdb->add_atoms(batch_atoms, true); } total_atoms_processed += thread_atoms_count; @@ -280,8 +280,8 @@ int main(int argc, char* argv[]) { links.push_back(link_with_nested); if (j % chunk_size == 0) { - thread_db->add_nodes(nodes, false, true); - thread_db->add_links(links, false, true); + thread_db->add_nodes(nodes, true); + thread_db->add_links(links, true); nodes.clear(); links.clear(); } @@ -290,12 +290,12 @@ int main(int argc, char* argv[]) { if (!nodes.empty()) { LOG_INFO("[" + to_string(thread_id) + "] Final - Adding " + to_string(nodes.size()) + " nodes"); - thread_db->add_nodes(nodes, false, true); + thread_db->add_nodes(nodes, true); } if (!links.empty()) { LOG_INFO("[" + to_string(thread_id) + "] Final - Adding " + to_string(links.size()) + " links"); - thread_db->add_links(links, false, true); + thread_db->add_links(links, true); } // clang-format off diff --git a/src/tests/cpp/adapterdb_test.cc b/src/tests/cpp/adapterdb_test.cc index 54a0040fa..c1c1f97e1 100644 --- a/src/tests/cpp/adapterdb_test.cc +++ b/src/tests/cpp/adapterdb_test.cc @@ -12,6 +12,7 @@ #include "AtomDBSingleton.h" #include "Link.h" +#include "Merger.h" #include "MorkDB.h" #include "Node.h" #include "RedisMongoDB.h" @@ -402,8 +403,8 @@ TEST_P(AdapterDBTest, AddNodeWithThrowIfExists) { auto node = new Node("Symbol", "AdapterThrowIfExistsNode"); - EXPECT_EQ(db->add_node(node, true), node->handle()); - EXPECT_THROW({ db->add_node(node, true); }, runtime_error); + EXPECT_EQ(db->add_node(node, &ThrowIfExistsMerger::instance()), node->handle()); + EXPECT_THROW({ db->add_node(node, &ThrowIfExistsMerger::instance()); }, runtime_error); EXPECT_TRUE(db->delete_node(node->handle())); delete node; @@ -424,7 +425,7 @@ TEST_P(AdapterDBTest, AddLinkWithThrowIfExists) { auto link = new Link("Expression", {h1, h2, h3}); EXPECT_EQ(db->add_link(link), link->handle()); - EXPECT_THROW({ db->add_link(link, true); }, runtime_error); + EXPECT_THROW({ db->add_link(link, &ThrowIfExistsMerger::instance()); }, runtime_error); EXPECT_TRUE(db->delete_link(link->handle())); EXPECT_TRUE(db->delete_node(h1)); diff --git a/src/tests/cpp/chain_operator_test.cc b/src/tests/cpp/chain_operator_test.cc index 0d4ec55c4..1766a9358 100644 --- a/src/tests/cpp/chain_operator_test.cc +++ b/src/tests/cpp/chain_operator_test.cc @@ -53,25 +53,32 @@ class ChainOperatorTestEnvironment : public ::testing::Environment { public: void load_data() { auto db = AtomDBSingleton::get_instance(); - atoms::Node *node1, *node2; - atoms::Link* link; - node1 = new atoms::Node(NODE_TYPE, EVALUATION); - LOG_DEBUG("Add node: " + node1->handle() + " " + node1->to_string()); - db->add_node(node1, false); + + // Insert each node once (duplicate-tolerant upsert is the default). + auto evaluation = new atoms::Node(NODE_TYPE, EVALUATION); + EXPECT_EQ(db->add_node(evaluation), evaluation->handle()); + delete evaluation; + + vector nodes; + for (unsigned int i = 0; i <= (NODE_COUNT + 1); i++) { + auto node = new atoms::Node(NODE_TYPE, node_name(i)); + LOG_DEBUG("Add node: " + node->handle() + " " + node->to_string()); + EXPECT_EQ(db->add_node(node), node->handle()); + nodes.push_back(node); + } + for (unsigned int i = 0; i <= (NODE_COUNT + 1); i++) { - node1 = new atoms::Node(NODE_TYPE, node_name(i)); - db->add_node(node1, false); - LOG_DEBUG("Add node: " + node1->handle() + " " + node1->to_string()); for (unsigned int j = 0; j <= (NODE_COUNT + 1); j++) { - node2 = new atoms::Node(NODE_TYPE, node_name(j)); - LOG_DEBUG("Add node: " + node2->handle() + " " + node2->to_string()); - db->add_node(node2, false); - link = new atoms::Link( - LINK_TYPE, {EVALUATION_HANDLE, node1->handle(), node2->handle()}, true); + auto link = new atoms::Link( + LINK_TYPE, {EVALUATION_HANDLE, nodes[i]->handle(), nodes[j]->handle()}, true); LOG_DEBUG("Add link: " + link->handle() + " " + link->to_string()); - db->add_link(link, false); + db->add_link(link); + delete link; } } + for (auto* node : nodes) { + delete node; + } } void SetUp() override { diff --git a/src/tests/cpp/inmemorydb_test.cc b/src/tests/cpp/inmemorydb_test.cc index 234d0651c..99c865c8e 100644 --- a/src/tests/cpp/inmemorydb_test.cc +++ b/src/tests/cpp/inmemorydb_test.cc @@ -12,7 +12,9 @@ #include "InMemoryDBAPITypes.h" #include "Link.h" #include "LinkSchema.h" +#include "Merger.h" #include "Node.h" +#include "Properties.h" using namespace atomdb; using namespace atomdb::atomdb_api_types; @@ -37,12 +39,12 @@ TEST_F(InMemoryDBTest, AddNodesAndLinks) { auto similarity = new Node("Symbol", "Similarity"); auto inheritance = new Node("Symbol", "Inheritance"); - string human_handle = db->add_node(human, false); - string monkey_handle = db->add_node(monkey, false); - string chimp_handle = db->add_node(chimp, false); - string mammal_handle = db->add_node(mammal, false); - string similarity_handle = db->add_node(similarity, false); - string inheritance_handle = db->add_node(inheritance, false); + string human_handle = db->add_node(human); + string monkey_handle = db->add_node(monkey); + string chimp_handle = db->add_node(chimp); + string mammal_handle = db->add_node(mammal); + string similarity_handle = db->add_node(similarity); + string inheritance_handle = db->add_node(inheritance); // Verify nodes were added EXPECT_TRUE(db->node_exists(human_handle)); @@ -58,11 +60,11 @@ TEST_F(InMemoryDBTest, AddNodesAndLinks) { auto link4 = new Link("Expression", {inheritance_handle, monkey_handle, mammal_handle}); auto link5 = new Link("Expression", {inheritance_handle, chimp_handle, mammal_handle}); - string link1_handle = db->add_link(link1, false); - string link2_handle = db->add_link(link2, false); - string link3_handle = db->add_link(link3, false); - string link4_handle = db->add_link(link4, false); - string link5_handle = db->add_link(link5, false); + string link1_handle = db->add_link(link1); + string link2_handle = db->add_link(link2); + string link3_handle = db->add_link(link3); + string link4_handle = db->add_link(link4); + string link5_handle = db->add_link(link5); // Verify links were added EXPECT_TRUE(db->link_exists(link1_handle)); @@ -86,19 +88,19 @@ TEST_F(InMemoryDBTest, QueryForPattern) { auto mammal = new Node("Symbol", "\"mammal\""); auto inheritance = new Node("Symbol", "Inheritance"); - string human_handle = db->add_node(human, false); - string monkey_handle = db->add_node(monkey, false); - string chimp_handle = db->add_node(chimp, false); - string mammal_handle = db->add_node(mammal, false); - string inheritance_handle = db->add_node(inheritance, false); + string human_handle = db->add_node(human); + string monkey_handle = db->add_node(monkey); + string chimp_handle = db->add_node(chimp); + string mammal_handle = db->add_node(mammal); + string inheritance_handle = db->add_node(inheritance); auto link1 = new Link("Expression", {inheritance_handle, human_handle, mammal_handle}); auto link2 = new Link("Expression", {inheritance_handle, monkey_handle, mammal_handle}); auto link3 = new Link("Expression", {inheritance_handle, chimp_handle, mammal_handle}); - string link1_handle = db->add_link(link1, false); - string link2_handle = db->add_link(link2, false); - string link3_handle = db->add_link(link3, false); + string link1_handle = db->add_link(link1); + string link2_handle = db->add_link(link2); + string link3_handle = db->add_link(link3); // Re-index patterns to ensure re_index works db->re_index_patterns(true); @@ -137,13 +139,13 @@ TEST_F(InMemoryDBTest, QueryForPatternWithSpecificMatch) { auto monkey = new Node("Symbol", "\"monkey\""); auto similarity = new Node("Symbol", "Similarity"); - string human_handle = db->add_node(human, false); - string monkey_handle = db->add_node(monkey, false); - string similarity_handle = db->add_node(similarity, false); + string human_handle = db->add_node(human); + string monkey_handle = db->add_node(monkey); + string similarity_handle = db->add_node(similarity); auto link1 = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); - string link1_handle = db->add_link(link1, false); + string link1_handle = db->add_link(link1); LinkSchema link_schema({"LINK_TEMPLATE", "Expression", @@ -188,16 +190,16 @@ TEST_F(InMemoryDBTest, QueryForTargets) { auto node3 = new Node("Symbol", "Node3"); auto similarity = new Node("Symbol", "Similarity"); - string node1_handle = db->add_node(node1, false); - string node2_handle = db->add_node(node2, false); - string node3_handle = db->add_node(node3, false); - string similarity_handle = db->add_node(similarity, false); + string node1_handle = db->add_node(node1); + string node2_handle = db->add_node(node2); + string node3_handle = db->add_node(node3); + string similarity_handle = db->add_node(similarity); auto node_targets = db->query_for_targets(node1_handle); EXPECT_EQ(node_targets, nullptr); auto link1 = new Link("Expression", {similarity_handle, node1_handle, node2_handle, node3_handle}); - string link1_handle = db->add_link(link1, false); + string link1_handle = db->add_link(link1); auto link1_targets = db->query_for_targets(link1_handle); EXPECT_EQ(link1_targets->size(), 4); @@ -219,18 +221,18 @@ TEST_F(InMemoryDBTest, QueryForTargetsMultipleLinks) { auto chimp = new Node("Symbol", "\"chimp\""); auto similarity = new Node("Symbol", "Similarity"); - string human_handle = db->add_node(human, false); - string monkey_handle = db->add_node(monkey, false); - string chimp_handle = db->add_node(chimp, false); - string similarity_handle = db->add_node(similarity, false); + string human_handle = db->add_node(human); + string monkey_handle = db->add_node(monkey); + string chimp_handle = db->add_node(chimp); + string similarity_handle = db->add_node(similarity); auto link1 = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); auto link2 = new Link("Expression", {similarity_handle, human_handle, chimp_handle}); auto link3 = new Link("Expression", {similarity_handle, monkey_handle, chimp_handle}); - string link1_handle = db->add_link(link1, false); - string link2_handle = db->add_link(link2, false); - string link3_handle = db->add_link(link3, false); + string link1_handle = db->add_link(link1); + string link2_handle = db->add_link(link2); + string link3_handle = db->add_link(link3); auto link1_targets = db->query_for_targets(link1_handle); EXPECT_EQ(link1_targets->size(), 3); @@ -256,12 +258,12 @@ TEST_F(InMemoryDBTest, QueryForTargetsAfterDeletion) { auto node2 = new Node("Symbol", "Node2"); auto similarity = new Node("Symbol", "Similarity"); - string node1_handle = db->add_node(node1, false); - string node2_handle = db->add_node(node2, false); - string similarity_handle = db->add_node(similarity, false); + string node1_handle = db->add_node(node1); + string node2_handle = db->add_node(node2); + string similarity_handle = db->add_node(similarity); auto link1 = new Link("Expression", {similarity_handle, node1_handle, node2_handle}); - string link1_handle = db->add_link(link1, false); + string link1_handle = db->add_link(link1); auto targets = db->query_for_targets(link1_handle); EXPECT_EQ(targets->size(), 3); @@ -280,21 +282,21 @@ TEST_F(InMemoryDBTest, QueryForIncomingSet) { auto similarity = new Node("Symbol", "Similarity"); auto inheritance = new Node("Symbol", "Inheritance"); - string human_handle = db->add_node(human, false); - string monkey_handle = db->add_node(monkey, false); - string chimp_handle = db->add_node(chimp, false); - string mammal_handle = db->add_node(mammal, false); - string similarity_handle = db->add_node(similarity, false); - string inheritance_handle = db->add_node(inheritance, false); + string human_handle = db->add_node(human); + string monkey_handle = db->add_node(monkey); + string chimp_handle = db->add_node(chimp); + string mammal_handle = db->add_node(mammal); + string similarity_handle = db->add_node(similarity); + string inheritance_handle = db->add_node(inheritance); // Create links that reference human auto link1 = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); auto link2 = new Link("Expression", {similarity_handle, human_handle, chimp_handle}); auto link3 = new Link("Expression", {inheritance_handle, human_handle, mammal_handle}); - string link1_handle = db->add_link(link1, false); - string link2_handle = db->add_link(link2, false); - string link3_handle = db->add_link(link3, false); + string link1_handle = db->add_link(link1); + string link2_handle = db->add_link(link2); + string link3_handle = db->add_link(link3); // Query incoming set for human auto incoming_set = db->query_for_incoming_set(human_handle); @@ -331,12 +333,12 @@ TEST_F(InMemoryDBTest, QueryForIncomingSetAfterDeletion) { auto monkey = new Node("Symbol", "\"monkey\""); auto similarity = new Node("Symbol", "Similarity"); - string human_handle = db->add_node(human, false); - string monkey_handle = db->add_node(monkey, false); - string similarity_handle = db->add_node(similarity, false); + string human_handle = db->add_node(human); + string monkey_handle = db->add_node(monkey); + string similarity_handle = db->add_node(similarity); auto link1 = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); - string link1_handle = db->add_link(link1, false); + string link1_handle = db->add_link(link1); // Verify incoming set before deletion auto incoming_set = db->query_for_incoming_set(human_handle); @@ -355,13 +357,13 @@ TEST_F(InMemoryDBTest, DeleteAtom) { auto monkey = new Node("Symbol", "\"monkey\""); auto similarity = new Node("Symbol", "Similarity"); - string human_handle = db->add_node(human, false); - string monkey_handle = db->add_node(monkey, false); - string similarity_handle = db->add_node(similarity, false); + string human_handle = db->add_node(human); + string monkey_handle = db->add_node(monkey); + string similarity_handle = db->add_node(similarity); // Create a link that references human auto link1 = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); - string link1_handle = db->add_link(link1, false); + string link1_handle = db->add_link(link1); // Try to delete human atom with delete_link_targets=false (should fail) bool deleted = db->delete_atom(human_handle, false); @@ -373,7 +375,7 @@ TEST_F(InMemoryDBTest, DeleteAtom) { // Create a link that references human auto link2 = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); - string link2_handle = db->add_link(link2, false); + string link2_handle = db->add_link(link2); // Delete human atom with delete_link_targets=true (should succeed and delete the link) deleted = db->delete_atom(human_handle, true); @@ -391,13 +393,13 @@ TEST_F(InMemoryDBTest, DeleteNode) { auto monkey = new Node("Symbol", "\"monkey\""); auto similarity = new Node("Symbol", "Similarity"); - string human_handle = db->add_node(human, false); - string monkey_handle = db->add_node(monkey, false); - string similarity_handle = db->add_node(similarity, false); + string human_handle = db->add_node(human); + string monkey_handle = db->add_node(monkey); + string similarity_handle = db->add_node(similarity); // Create a link that references human auto link1 = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); - string link1_handle = db->add_link(link1, false); + string link1_handle = db->add_link(link1); // Try to delete human with delete_link_targets=false (should fail) bool deleted = db->delete_node(human_handle, false); @@ -413,7 +415,7 @@ TEST_F(InMemoryDBTest, DeleteNode) { // Create a link that references human auto link2 = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); - string link2_handle = db->add_link(link2, false); + string link2_handle = db->add_link(link2); // Delete human with delete_link_targets=true (should succeed and delete the link) deleted = db->delete_node(human_handle, true); @@ -435,13 +437,13 @@ TEST_F(InMemoryDBTest, DeleteLink) { auto monkey = new Node("Symbol", "\"monkey\""); auto similarity = new Node("Symbol", "Similarity"); - string human_handle = db->add_node(human, false); - string monkey_handle = db->add_node(monkey, false); - string similarity_handle = db->add_node(similarity, false); + string human_handle = db->add_node(human); + string monkey_handle = db->add_node(monkey); + string similarity_handle = db->add_node(similarity); // Create a link that references human and monkey auto link1 = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); - string link1_handle = db->add_link(link1, false); + string link1_handle = db->add_link(link1); // Delete link with delete_link_targets=false (should succeed, targets remain) bool deleted = db->delete_link(link1_handle, false); @@ -463,7 +465,7 @@ TEST_F(InMemoryDBTest, DeleteLink) { // Create a link that references human and monkey auto link2 = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); - string link2_handle = db->add_link(link2, false); + string link2_handle = db->add_link(link2); // Delete link with delete_link_targets=true (should delete targets if no other references) deleted = db->delete_link(link2_handle, true); @@ -484,16 +486,16 @@ TEST_F(InMemoryDBTest, DeleteLinkMultipleReferences) { auto chimp = new Node("Symbol", "\"chimp\""); auto similarity = new Node("Symbol", "Similarity"); - string human_handle = db->add_node(human, false); - string monkey_handle = db->add_node(monkey, false); - string chimp_handle = db->add_node(chimp, false); - string similarity_handle = db->add_node(similarity, false); + string human_handle = db->add_node(human); + string monkey_handle = db->add_node(monkey); + string chimp_handle = db->add_node(chimp); + string similarity_handle = db->add_node(similarity); // Create two links that both reference human auto link1 = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); auto link2 = new Link("Expression", {similarity_handle, human_handle, chimp_handle}); - string link1_handle = db->add_link(link1, false); - string link2_handle = db->add_link(link2, false); + string link1_handle = db->add_link(link1); + string link2_handle = db->add_link(link2); // Verify human has 2 incoming links auto human_incoming = db->query_for_incoming_set(human_handle); @@ -528,14 +530,14 @@ TEST_F(InMemoryDBTest, AtomsCount) { auto node2 = new Node("Symbol", "Node2"); auto similarity = new Node("Symbol", "Similarity"); - db->add_node(node1, false); - db->add_node(node2, false); - db->add_node(similarity, false); + db->add_node(node1); + db->add_node(node2); + db->add_node(similarity); EXPECT_EQ(db->atom_count(), 3); auto link1 = new Link("Expression", {similarity->handle(), node1->handle(), node2->handle()}); - db->add_link(link1, false); + db->add_link(link1); EXPECT_EQ(db->atom_count(), 4); EXPECT_EQ(db->empty(), false); @@ -651,6 +653,330 @@ TEST(HandleSetInMemoryTest, IteratorVisitsAllHandlesOnce) { EXPECT_NE(find(visited.begin(), visited.end(), handle_b), visited.end()); } +namespace { + +class SumStrengthMerger : public Merger { + public: + bool merge(Atom* existing, const Atom* incoming) const override { + double existing_strength = existing->custom_attributes.get_or("strength", 0.0); + double incoming_strength = incoming->custom_attributes.get_or("strength", 0.0); + existing->custom_attributes["strength"] = existing_strength + incoming_strength; + return true; + } +}; + +} // namespace + +TEST_F(InMemoryDBTest, AddNodeReplacesByDefault) { + Properties attrs1; + attrs1["strength"] = 0.1; + attrs1["obsolete"] = true; + auto node1 = new Node("Symbol", "\"replace_me\"", attrs1); + string handle = db->add_node(node1); + EXPECT_DOUBLE_EQ(db->get_node(handle)->custom_attributes.get_or("strength", -1.0), 0.1); + EXPECT_TRUE(db->get_node(handle)->custom_attributes.get_or("obsolete", false)); + + Properties attrs2; + attrs2["strength"] = 0.9; + auto node2 = new Node("Symbol", "\"replace_me\"", attrs2); + EXPECT_EQ(db->add_node(node2), handle); + EXPECT_DOUBLE_EQ(db->get_node(handle)->custom_attributes.get_or("strength", -1.0), 0.9); + EXPECT_FALSE(db->get_node(handle)->custom_attributes.get_or("obsolete", false)); + + delete node1; + delete node2; +} + +TEST_F(InMemoryDBTest, AddNodeThrowIfExistsMerger) { + auto node1 = new Node("Symbol", "\"throw_me\""); + auto node2 = new Node("Symbol", "\"throw_me\""); + EXPECT_EQ(db->add_node(node1, &ThrowIfExistsMerger::instance()), node1->handle()); + EXPECT_THROW(db->add_node(node2, &ThrowIfExistsMerger::instance()), runtime_error); + delete node1; + delete node2; +} + +TEST_F(InMemoryDBTest, AddNodeCustomMerger) { + Properties attrs1; + attrs1["strength"] = 0.2; + auto node1 = new Node("Symbol", "\"merge_me\"", attrs1); + string handle = db->add_node(node1); + + Properties attrs2; + attrs2["strength"] = 0.3; + auto node2 = new Node("Symbol", "\"merge_me\"", attrs2); + SumStrengthMerger merger; + EXPECT_EQ(db->add_node(node2, &merger), handle); + EXPECT_DOUBLE_EQ(db->get_node(handle)->custom_attributes.get_or("strength", -1.0), 0.5); + + delete node1; + delete node2; +} + +TEST_F(InMemoryDBTest, AddLinkReplacesByDefault) { + auto n1 = new Node("Symbol", "\"link_replace_a\""); + auto n2 = new Node("Symbol", "\"link_replace_b\""); + auto h1 = db->add_node(n1); + auto h2 = db->add_node(n2); + + Properties attrs1; + attrs1["strength"] = 0.1; + attrs1["obsolete"] = true; + auto link1 = new Link("Expression", {h1, h2}, attrs1); + string handle = db->add_link(link1); + EXPECT_DOUBLE_EQ(db->get_link(handle)->custom_attributes.get_or("strength", -1.0), 0.1); + EXPECT_TRUE(db->get_link(handle)->custom_attributes.get_or("obsolete", false)); + + Properties attrs2; + attrs2["strength"] = 0.9; + auto link2 = new Link("Expression", {h1, h2}, attrs2); + EXPECT_EQ(db->add_link(link2), handle); + EXPECT_DOUBLE_EQ(db->get_link(handle)->custom_attributes.get_or("strength", -1.0), 0.9); + EXPECT_FALSE(db->get_link(handle)->custom_attributes.get_or("obsolete", false)); + + auto incoming = db->query_for_incoming_set(h1); + EXPECT_EQ(incoming->size(), 1u); + + delete n1; + delete n2; + delete link1; + delete link2; +} + +TEST_F(InMemoryDBTest, AddLinkThrowIfExistsMerger) { + auto n1 = new Node("Symbol", "\"link_throw_a\""); + auto n2 = new Node("Symbol", "\"link_throw_b\""); + auto h1 = db->add_node(n1); + auto h2 = db->add_node(n2); + + auto link1 = new Link("Expression", {h1, h2}); + auto link2 = new Link("Expression", {h1, h2}); + EXPECT_EQ(db->add_link(link1, &ThrowIfExistsMerger::instance()), link1->handle()); + EXPECT_THROW(db->add_link(link2, &ThrowIfExistsMerger::instance()), runtime_error); + + delete n1; + delete n2; + delete link1; + delete link2; +} + +TEST_F(InMemoryDBTest, AddLinkCustomMerger) { + auto n1 = new Node("Symbol", "\"link_merge_a\""); + auto n2 = new Node("Symbol", "\"link_merge_b\""); + auto h1 = db->add_node(n1); + auto h2 = db->add_node(n2); + + Properties attrs1; + attrs1["strength"] = 0.2; + auto link1 = new Link("Expression", {h1, h2}, attrs1); + string handle = db->add_link(link1); + + Properties attrs2; + attrs2["strength"] = 0.3; + auto link2 = new Link("Expression", {h1, h2}, attrs2); + SumStrengthMerger merger; + EXPECT_EQ(db->add_link(link2, &merger), handle); + EXPECT_DOUBLE_EQ(db->get_link(handle)->custom_attributes.get_or("strength", -1.0), 0.5); + + delete n1; + delete n2; + delete link1; + delete link2; +} + +TEST_F(InMemoryDBTest, AddNodeSkipIfExistsMergerDoesNotPersist) { + Properties attrs1; + attrs1["strength"] = 0.2; + auto node1 = new Node("Symbol", "\"reject_me\"", attrs1); + string handle = db->add_node(node1); + + Properties attrs2; + attrs2["strength"] = 0.9; + auto node2 = new Node("Symbol", "\"reject_me\"", attrs2); + EXPECT_EQ(db->add_node(node2, &SkipIfExistsMerger::instance()), ""); + EXPECT_DOUBLE_EQ(db->get_node(handle)->custom_attributes.get_or("strength", -1.0), 0.2); + + delete node1; + delete node2; +} + +TEST_F(InMemoryDBTest, AddLinkSkipIfExistsMergerDoesNotPersistOrReindex) { + auto n1 = new Node("Symbol", "\"link_reject_a\""); + auto n2 = new Node("Symbol", "\"link_reject_b\""); + auto h1 = db->add_node(n1); + auto h2 = db->add_node(n2); + + Properties attrs1; + attrs1["strength"] = 0.2; + auto link1 = new Link("Expression", {h1, h2}, attrs1); + string handle = db->add_link(link1); + auto incoming_before = db->query_for_incoming_set(h1); + ASSERT_EQ(incoming_before->size(), 1u); + + Properties attrs2; + attrs2["strength"] = 0.9; + auto link2 = new Link("Expression", {h1, h2}, attrs2); + EXPECT_EQ(db->add_link(link2, &SkipIfExistsMerger::instance()), ""); + EXPECT_DOUBLE_EQ(db->get_link(handle)->custom_attributes.get_or("strength", -1.0), 0.2); + + auto incoming_after = db->query_for_incoming_set(h1); + EXPECT_EQ(incoming_after->size(), 1u); + + delete n1; + delete n2; + delete link1; + delete link2; +} + +TEST_F(InMemoryDBTest, AddNodesThrowIfExistsIsNotAllOrNothing) { + auto existing = new Node("Symbol", "\"batch_throw_existing\""); + db->add_node(existing); + + auto keep_absent = new Node("Symbol", "\"batch_throw_new_a\""); + auto collision = new Node("Symbol", "\"batch_throw_existing\""); + auto also_absent = new Node("Symbol", "\"batch_throw_new_b\""); + + // Earlier items may be applied; ThrowIfExists raises when the collision is hit. + EXPECT_THROW( + db->add_nodes({keep_absent, collision, also_absent}, false, &ThrowIfExistsMerger::instance()), + runtime_error); + EXPECT_TRUE(db->node_exists(keep_absent->handle())); + EXPECT_FALSE(db->node_exists(also_absent->handle())); + EXPECT_TRUE(db->node_exists(existing->handle())); + + // Duplicate only inside the batch: first insert succeeds, rematch raises. + auto batch_a = new Node("Symbol", "\"batch_throw_dup_only\""); + auto batch_a_copy = new Node("Symbol", "\"batch_throw_dup_only\""); + auto batch_b = new Node("Symbol", "\"batch_throw_sibling\""); + EXPECT_THROW( + db->add_nodes({batch_a, batch_a_copy, batch_b}, false, &ThrowIfExistsMerger::instance()), + runtime_error); + EXPECT_TRUE(db->node_exists(batch_a->handle())); + EXPECT_FALSE(db->node_exists(batch_b->handle())); + + delete existing; + delete keep_absent; + delete collision; + delete also_absent; + delete batch_a; + delete batch_a_copy; + delete batch_b; +} + +TEST_F(InMemoryDBTest, AddLinksThrowIfExistsIsNotAllOrNothing) { + auto n1 = new Node("Symbol", "\"batch_link_throw_a\""); + auto n2 = new Node("Symbol", "\"batch_link_throw_b\""); + auto n3 = new Node("Symbol", "\"batch_link_throw_c\""); + auto n4 = new Node("Symbol", "\"batch_link_throw_d\""); + auto n5 = new Node("Symbol", "\"batch_link_throw_e\""); + auto n6 = new Node("Symbol", "\"batch_link_throw_f\""); + auto h1 = db->add_node(n1); + auto h2 = db->add_node(n2); + auto h3 = db->add_node(n3); + auto h4 = db->add_node(n4); + auto h5 = db->add_node(n5); + auto h6 = db->add_node(n6); + + auto existing = new Link("Expression", {h1, h2}); + db->add_link(existing); + + auto keep_absent = new Link("Expression", {h3, h4}); + auto collision = new Link("Expression", {h1, h2}); + + EXPECT_THROW(db->add_links({keep_absent, collision}, false, &ThrowIfExistsMerger::instance()), + runtime_error); + EXPECT_TRUE(db->link_exists(keep_absent->handle())); + EXPECT_TRUE(db->link_exists(existing->handle())); + EXPECT_EQ(db->query_for_incoming_set(h3)->size(), 1u); + + EXPECT_TRUE(db->delete_link(keep_absent->handle())); + + auto batch_dup = new Link("Expression", {h5, h6}); + auto batch_dup_copy = new Link("Expression", {h5, h6}); + auto batch_other = new Link("Expression", {h3, h4}); + EXPECT_THROW( + db->add_links({batch_dup, batch_dup_copy, batch_other}, false, &ThrowIfExistsMerger::instance()), + runtime_error); + EXPECT_TRUE(db->link_exists(batch_dup->handle())); + EXPECT_FALSE(db->link_exists(batch_other->handle())); + EXPECT_EQ(db->query_for_incoming_set(h5)->size(), 1u); + + delete n1; + delete n2; + delete n3; + delete n4; + delete n5; + delete n6; + delete existing; + delete keep_absent; + delete collision; + delete batch_dup; + delete batch_dup_copy; + delete batch_other; +} + +TEST_F(InMemoryDBTest, AddNodesSkipIfExistsMergerReturnsEmptyHandleSlots) { + Properties attrs1; + attrs1["strength"] = 0.2; + auto existing = new Node("Symbol", "\"batch_reject_existing\"", attrs1); + string existing_handle = db->add_node(existing); + + auto fresh = new Node("Symbol", "\"batch_reject_new\""); + Properties attrs2; + attrs2["strength"] = 0.9; + auto collision = new Node("Symbol", "\"batch_reject_existing\"", attrs2); + + auto handles = db->add_nodes({fresh, collision}, false, &SkipIfExistsMerger::instance()); + ASSERT_EQ(handles.size(), 2u); + EXPECT_EQ(handles[0], fresh->handle()); + EXPECT_EQ(handles[1], ""); + EXPECT_TRUE(db->node_exists(fresh->handle())); + EXPECT_DOUBLE_EQ(db->get_node(existing_handle)->custom_attributes.get_or("strength", -1.0), + 0.2); + + delete existing; + delete fresh; + delete collision; +} + +TEST_F(InMemoryDBTest, AddLinksSkipIfExistsMergerReturnsEmptyHandleSlots) { + auto n1 = new Node("Symbol", "\"batch_link_reject_a\""); + auto n2 = new Node("Symbol", "\"batch_link_reject_b\""); + auto n3 = new Node("Symbol", "\"batch_link_reject_c\""); + auto n4 = new Node("Symbol", "\"batch_link_reject_d\""); + auto h1 = db->add_node(n1); + auto h2 = db->add_node(n2); + auto h3 = db->add_node(n3); + auto h4 = db->add_node(n4); + + Properties attrs1; + attrs1["strength"] = 0.2; + auto existing = new Link("Expression", {h1, h2}, attrs1); + string existing_handle = db->add_link(existing); + + auto fresh = new Link("Expression", {h3, h4}); + Properties attrs2; + attrs2["strength"] = 0.9; + auto collision = new Link("Expression", {h1, h2}, attrs2); + + auto handles = db->add_links({fresh, collision}, false, &SkipIfExistsMerger::instance()); + ASSERT_EQ(handles.size(), 2u); + EXPECT_EQ(handles[0], fresh->handle()); + EXPECT_EQ(handles[1], ""); + EXPECT_TRUE(db->link_exists(fresh->handle())); + EXPECT_DOUBLE_EQ(db->get_link(existing_handle)->custom_attributes.get_or("strength", -1.0), + 0.2); + EXPECT_EQ(db->query_for_incoming_set(h1)->size(), 1u); + + delete n1; + delete n2; + delete n3; + delete n4; + delete existing; + delete fresh; + delete collision; +} + int main(int argc, char** argv) { ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); diff --git a/src/tests/cpp/link_creation_agent_test.cc b/src/tests/cpp/link_creation_agent_test.cc index 55413e822..2da2550b2 100644 --- a/src/tests/cpp/link_creation_agent_test.cc +++ b/src/tests/cpp/link_creation_agent_test.cc @@ -724,7 +724,7 @@ TEST_F(LinkCreationAgentTest, TestMettaProcessorLinkCreationInnerCheck) { EXPECT_CALL(*mock_atomdb, add_node(testing::_, testing::_)) .Times(expected_nodes.size()) .WillRepeatedly(::testing::Invoke( - [&expected_nodes, &nodes_count](const atoms::Node* node, bool throw_if_exists) { + [&expected_nodes, &nodes_count](const atoms::Node* node, const atoms::Merger* merger) { EXPECT_EQ(node->type, "Symbol"); EXPECT_EQ(node->name, expected_nodes[nodes_count].c_str()); nodes_count++; @@ -749,7 +749,7 @@ TEST_F(LinkCreationAgentTest, TestMettaProcessorLinkCreationInnerCheck) { EXPECT_CALL(*mock_atomdb, add_link(testing::_, testing::_)) .Times(expected_links.size()) .WillRepeatedly(::testing::Invoke( - [&links_count, &expected_links](const atoms::Link* link, bool throw_if_exists) { + [&links_count, &expected_links](const atoms::Link* link, const atoms::Merger* merger) { EXPECT_EQ(link->type, "Expression"); EXPECT_EQ(link->targets.size(), expected_links[links_count].size()); for (size_t i = 0; i < expected_links[links_count].size(); i++) { @@ -793,7 +793,7 @@ TEST_F(LinkCreationAgentTest, TestMettaProcessorLinkCreationSimpleMeta) { EXPECT_CALL(*mock_atomdb, add_node(testing::_, testing::_)) .Times(expected_nodes.size()) .WillRepeatedly(::testing::Invoke( - [&expected_nodes, &nodes_count](const atoms::Node* node, bool throw_if_exists) { + [&expected_nodes, &nodes_count](const atoms::Node* node, const atoms::Merger* merger) { EXPECT_EQ(node->type, "Symbol"); EXPECT_EQ(node->name, expected_nodes[nodes_count].c_str()); nodes_count++; diff --git a/src/tests/cpp/morkdb_test.cc b/src/tests/cpp/morkdb_test.cc index 5ff1e41dd..330026268 100644 --- a/src/tests/cpp/morkdb_test.cc +++ b/src/tests/cpp/morkdb_test.cc @@ -11,6 +11,7 @@ #include "AtomDBSingleton.h" #include "LinkSchema.h" +#include "Merger.h" #include "MockAnimalsData.h" #include "TestAtomDBJsonConfig.h" using namespace atomdb; @@ -281,7 +282,7 @@ TEST_F(MorkDBTest, AddLinksWithDuplicateTargets) { nodes.push_back(new Node("Symbol", "DuplicateTargets1")); nodes.push_back(new Node("Symbol", "DuplicateTargets2")); nodes.push_back(new Node("Symbol", "DuplicateTargets3")); - EXPECT_EQ(db->add_nodes(nodes, true).size(), 3); + EXPECT_EQ(db->add_nodes(nodes, false, &ThrowIfExistsMerger::instance()).size(), 3); auto link = new Link("Expression", {nodes[0]->handle(), @@ -340,15 +341,15 @@ TEST_F(MorkDBTest, ConcurrentAddLinks) { links.push_back(link_with_nested); if (i % chunck_size == 0) { - db->add_nodes(nodes, false, true); - db->add_links(links, false, true); + db->add_nodes(nodes, true); + db->add_links(links, true); nodes.clear(); links.clear(); } } - if (!nodes.empty()) db->add_nodes(nodes, false, true); - if (!links.empty()) db->add_links(links, false, true); + if (!nodes.empty()) db->add_nodes(nodes, true); + if (!links.empty()) db->add_links(links, true); success_count++; } catch (const exception& e) { @@ -384,8 +385,8 @@ TEST_F(MorkDBTest, AddLinkWithoutMettaExpressionMustPopulateIt) { auto human = new Node("Symbol", "\"human\""); auto robot = new Node("Symbol", "\"robot\""); - db->add_node(similarity, false); - db->add_node(human, false); + db->add_node(similarity); + db->add_node(human); auto robot_handle = db->add_node(robot); auto link = new Link("Expression", {similarity->handle(), human->handle(), robot->handle()}); diff --git a/src/tests/cpp/redis_mongodb_test.cc b/src/tests/cpp/redis_mongodb_test.cc index e3a306961..dc6170bb3 100644 --- a/src/tests/cpp/redis_mongodb_test.cc +++ b/src/tests/cpp/redis_mongodb_test.cc @@ -11,6 +11,7 @@ #include "AtomDBSingleton.h" #include "Hasher.h" #include "Link.h" +#include "Merger.h" #include "MettaMapping.h" #include "MockAnimalsData.h" #include "Node.h" @@ -951,21 +952,21 @@ TEST_F(RedisMongoDBTest, AddSameAtomMustNotThrow) { TEST_F(RedisMongoDBTest, AddNodesWithThrowIfExists) { auto node1 = new Node("Symbol", "ThrowIfExists1"); - EXPECT_EQ(db->add_node(node1, true), node1->handle()); + EXPECT_EQ(db->add_node(node1, &ThrowIfExistsMerger::instance()), node1->handle()); vector nodes; nodes.push_back(new Node("Symbol", "ThrowIfExists2")); nodes.push_back(new Node("Symbol", "ThrowIfExists3")); - EXPECT_EQ(db->add_nodes(nodes, true).size(), 2); + EXPECT_EQ(db->add_nodes(nodes, false, &ThrowIfExistsMerger::instance()).size(), 2); auto link = new Link("Expression", {node1->handle(), nodes[0]->handle(), nodes[1]->handle()}); - EXPECT_EQ(db->add_link(link, true), link->handle()); + EXPECT_EQ(db->add_link(link, &ThrowIfExistsMerger::instance()), link->handle()); // Try to add the same node again - EXPECT_THROW(db->add_node(node1, true), runtime_error); - EXPECT_THROW(db->add_nodes(nodes, true), runtime_error); - EXPECT_THROW(db->add_link(link, true), runtime_error); + EXPECT_THROW(db->add_node(node1, &ThrowIfExistsMerger::instance()), runtime_error); + EXPECT_THROW(db->add_nodes(nodes, false, &ThrowIfExistsMerger::instance()), runtime_error); + EXPECT_THROW(db->add_link(link, &ThrowIfExistsMerger::instance()), runtime_error); EXPECT_EQ(db->delete_link(link->handle(), true), true); EXPECT_EQ(db->link_exists(link->handle()), false); @@ -974,12 +975,173 @@ TEST_F(RedisMongoDBTest, AddNodesWithThrowIfExists) { EXPECT_EQ(db->node_exists(nodes[1]->handle()), false); } +namespace { + +class SumStrengthMerger : public Merger { + public: + bool merge(Atom* existing, const Atom* incoming) const override { + double existing_strength = existing->custom_attributes.get_or("strength", 0.0); + double incoming_strength = incoming->custom_attributes.get_or("strength", 0.0); + existing->custom_attributes["strength"] = existing_strength + incoming_strength; + return true; + } +}; + +} // namespace + +TEST_F(RedisMongoDBTest, AddNodeReplacesByDefault) { + Properties attrs1; + attrs1["strength"] = 0.1; + auto node1 = new Node("Symbol", "ReplaceByDefault1", attrs1); + string handle = db->add_node(node1); + EXPECT_DOUBLE_EQ(db->get_node(handle)->custom_attributes.get_or("strength", -1.0), 0.1); + + Properties attrs2; + attrs2["strength"] = 0.9; + auto node2 = new Node("Symbol", "ReplaceByDefault1", attrs2); + EXPECT_EQ(db->add_node(node2), handle); + EXPECT_DOUBLE_EQ(db->get_node(handle)->custom_attributes.get_or("strength", -1.0), 0.9); + + EXPECT_TRUE(db->delete_node(handle)); + delete node1; + delete node2; +} + +TEST_F(RedisMongoDBTest, AddNodeCustomMerger) { + Properties attrs1; + attrs1["strength"] = 0.2; + auto node1 = new Node("Symbol", "CustomMerge1", attrs1); + string handle = db->add_node(node1); + + Properties attrs2; + attrs2["strength"] = 0.3; + auto node2 = new Node("Symbol", "CustomMerge1", attrs2); + SumStrengthMerger merger; + EXPECT_EQ(db->add_node(node2, &merger), handle); + EXPECT_DOUBLE_EQ(db->get_node(handle)->custom_attributes.get_or("strength", -1.0), 0.5); + + EXPECT_TRUE(db->delete_node(handle)); + delete node1; + delete node2; +} + +TEST_F(RedisMongoDBTest, AddLinkReplacesAndCustomMerger) { + auto n1 = new Node("Symbol", "LinkMergeA"); + auto n2 = new Node("Symbol", "LinkMergeB"); + auto h1 = db->add_node(n1); + auto h2 = db->add_node(n2); + + Properties attrs1; + attrs1["strength"] = 0.1; + auto link1 = new Link("Expression", {h1, h2}, attrs1); + string handle = db->add_link(link1); + EXPECT_DOUBLE_EQ(db->get_link(handle)->custom_attributes.get_or("strength", -1.0), 0.1); + + Properties attrs2; + attrs2["strength"] = 0.9; + auto link2 = new Link("Expression", {h1, h2}, attrs2); + EXPECT_EQ(db->add_link(link2), handle); + EXPECT_DOUBLE_EQ(db->get_link(handle)->custom_attributes.get_or("strength", -1.0), 0.9); + + Properties attrs3; + attrs3["strength"] = 0.25; + auto link3 = new Link("Expression", {h1, h2}, attrs3); + SumStrengthMerger merger; + EXPECT_EQ(db->add_link(link3, &merger), handle); + EXPECT_DOUBLE_EQ(db->get_link(handle)->custom_attributes.get_or("strength", -1.0), 1.15); + + EXPECT_TRUE(db->delete_link(handle, true)); + delete n1; + delete n2; + delete link1; + delete link2; + delete link3; +} + +TEST_F(RedisMongoDBTest, AddNodesBatchReplaceAndCustomMerger) { + Properties attrs1; + attrs1["strength"] = 0.1; + attrs1["obsolete"] = true; + Properties attrs2; + attrs2["strength"] = 0.4; + Properties attrs3; + attrs3["strength"] = 0.5; + + auto first_a = new Node("Symbol", "BatchNodeA", attrs1); + auto first_b = new Node("Symbol", "BatchNodeB", attrs1); + EXPECT_EQ(db->add_nodes({first_a, first_b}).size(), 2u); + EXPECT_TRUE(db->get_node(first_a->handle())->custom_attributes.get_or("obsolete", false)); + + auto replace_a = new Node("Symbol", "BatchNodeA", attrs2); + auto replace_b = new Node("Symbol", "BatchNodeB", attrs2); + EXPECT_EQ(db->add_nodes({replace_a, replace_b}).size(), 2u); + EXPECT_DOUBLE_EQ( + db->get_node(replace_a->handle())->custom_attributes.get_or("strength", -1.0), 0.4); + EXPECT_FALSE(db->get_node(replace_a->handle())->custom_attributes.get_or("obsolete", false)); + + auto merge_a1 = new Node("Symbol", "BatchNodeA", attrs3); + auto merge_a2 = new Node("Symbol", "BatchNodeA", attrs3); + SumStrengthMerger merger; + EXPECT_EQ(db->add_nodes({merge_a1, merge_a2}, false, &merger).size(), 2u); + // 0.4 + 0.5 + 0.5 from in-batch repeated handle + EXPECT_DOUBLE_EQ( + db->get_node(merge_a1->handle())->custom_attributes.get_or("strength", -1.0), 1.4); + + EXPECT_TRUE(db->delete_node(first_a->handle())); + EXPECT_TRUE(db->delete_node(first_b->handle())); + delete first_a; + delete first_b; + delete replace_a; + delete replace_b; + delete merge_a1; + delete merge_a2; +} + +TEST_F(RedisMongoDBTest, AddLinksBatchReplaceAndCustomMerger) { + auto n1 = new Node("Symbol", "BatchLinkN1"); + auto n2 = new Node("Symbol", "BatchLinkN2"); + auto h1 = db->add_node(n1); + auto h2 = db->add_node(n2); + + Properties attrs1; + attrs1["strength"] = 0.1; + attrs1["obsolete"] = true; + Properties attrs2; + attrs2["strength"] = 0.4; + Properties attrs3; + attrs3["strength"] = 0.5; + + auto link1 = new Link("Expression", {h1, h2}, attrs1); + EXPECT_EQ(db->add_links({link1}).size(), 1u); + + auto link2 = new Link("Expression", {h1, h2}, attrs2); + EXPECT_EQ(db->add_links({link2}).size(), 1u); + EXPECT_DOUBLE_EQ(db->get_link(link2->handle())->custom_attributes.get_or("strength", -1.0), + 0.4); + EXPECT_FALSE(db->get_link(link2->handle())->custom_attributes.get_or("obsolete", false)); + + auto link3a = new Link("Expression", {h1, h2}, attrs3); + auto link3b = new Link("Expression", {h1, h2}, attrs3); + SumStrengthMerger merger; + EXPECT_EQ(db->add_links({link3a, link3b}, false, &merger).size(), 2u); + EXPECT_DOUBLE_EQ(db->get_link(link3a->handle())->custom_attributes.get_or("strength", -1.0), + 1.4); + + EXPECT_TRUE(db->delete_link(link1->handle(), true)); + delete n1; + delete n2; + delete link1; + delete link2; + delete link3a; + delete link3b; +} + TEST_F(RedisMongoDBTest, AddLinksWithDuplicateTargets) { vector nodes; nodes.push_back(new Node("Symbol", "DuplicateTargets1")); nodes.push_back(new Node("Symbol", "DuplicateTargets2")); nodes.push_back(new Node("Symbol", "DuplicateTargets3")); - EXPECT_EQ(db->add_nodes(nodes, true).size(), 3); + EXPECT_EQ(db->add_nodes(nodes, false, &ThrowIfExistsMerger::instance()).size(), 3); auto link = new Link("Expression", {nodes[0]->handle(), @@ -1005,9 +1167,9 @@ TEST_F(RedisMongoDBTest, AtomsCount) { auto node2 = new Node("Symbol", "Node2"); auto similarity = new Node("Symbol", "Similarity"); - db->add_node(node1, false); - db->add_node(node2, false); - db->add_node(similarity, false); + db->add_node(node1); + db->add_node(node2); + db->add_node(similarity); EXPECT_EQ(db->node_count(), 3); EXPECT_EQ(db->link_count(), 0); @@ -1015,7 +1177,7 @@ TEST_F(RedisMongoDBTest, AtomsCount) { EXPECT_EQ(db->empty(), false); auto link1 = new Link("Expression", {similarity->handle(), node1->handle(), node2->handle()}); - db->add_link(link1, false); + db->add_link(link1); EXPECT_EQ(db->node_count(), 3); EXPECT_EQ(db->link_count(), 1); @@ -1075,8 +1237,8 @@ TEST_F(RedisMongoDBTest, CompositeTypeEnabledFlag) { {transactional_nodes[0]->handle(), transactional_nodes[1]->handle(), transactional_nodes[2]->handle()}); - ASSERT_EQ(db_disabled->add_nodes(transactional_nodes, false, true).size(), 3); - ASSERT_EQ(db_disabled->add_links({transactional_link}, false, true).size(), 1); + ASSERT_EQ(db_disabled->add_nodes(transactional_nodes, true).size(), 3); + ASSERT_EQ(db_disabled->add_links({transactional_link}, true).size(), 1); auto transactional_doc = db_disabled->get_atom_document(transactional_link->handle()); ASSERT_NE(transactional_doc, nullptr); @@ -1088,6 +1250,103 @@ TEST_F(RedisMongoDBTest, CompositeTypeEnabledFlag) { EXPECT_TRUE(db_disabled->delete_atom(transactional_link->handle(), true)); } +TEST_F(RedisMongoDBTest, TransactionalMergeUsesFinalLinkCompositeType) { + vector nodes = {new Node("Symbol", "TxMergeCT-A"), + new Node("Symbol", "TxMergeCT-B"), + new Node("Symbol", "TxMergeCT-C")}; + ASSERT_EQ(db->add_nodes(nodes, true).size(), 3u); + + Properties attrs1; + attrs1["strength"] = 0.2; + auto link1 = + new Link("Expression", {nodes[0]->handle(), nodes[1]->handle(), nodes[2]->handle()}, attrs1); + ASSERT_EQ(db->add_links({link1}, true).size(), 1u); + + string expected_hash = link1->composite_type_hash(*db); + auto doc1 = db->get_atom_document(link1->handle()); + ASSERT_NE(doc1, nullptr); + EXPECT_EQ(string(doc1->get("composite_type_hash")), expected_hash); + EXPECT_EQ(doc1->get_size("composite_type"), 4); + + Properties attrs2; + attrs2["strength"] = 0.3; + auto link2 = + new Link("Expression", {nodes[0]->handle(), nodes[1]->handle(), nodes[2]->handle()}, attrs2); + SumStrengthMerger merger; + ASSERT_EQ(db->add_nodes(nodes, true).size(), 3u); + ASSERT_EQ(db->add_links({link2}, true, &merger).size(), 1u); + + auto doc2 = db->get_atom_document(link2->handle()); + ASSERT_NE(doc2, nullptr); + EXPECT_EQ(string(doc2->get("composite_type_hash")), expected_hash); + EXPECT_EQ(doc2->get_size("composite_type"), 4); + EXPECT_DOUBLE_EQ(db->get_link(link2->handle())->custom_attributes.get_or("strength", -1.0), + 0.5); + + EXPECT_TRUE(db->delete_atom(link2->handle(), true)); + delete nodes[0]; + delete nodes[1]; + delete nodes[2]; + delete link1; + delete link2; +} + +TEST_F(RedisMongoDBTest, TransactionalRejectedMergeStillBooksCompositeType) { + vector nodes = {new Node("Symbol", "TxRejectCT-A"), + new Node("Symbol", "TxRejectCT-B"), + new Node("Symbol", "TxRejectCT-C"), + new Node("Symbol", "TxRejectCT-D"), + new Node("Symbol", "TxRejectCT-E")}; + ASSERT_EQ(db->add_nodes(nodes, true).size(), 5u); + + auto existing = new Link("Expression", {nodes[0]->handle(), nodes[1]->handle(), nodes[2]->handle()}); + ASSERT_EQ(db->add_links({existing}, true).size(), 1u); + string existing_hash = existing->composite_type_hash(*db); + + // duplicate is rejected by SkipIfExistsMerger; nested targets the rejected handle and must + // still resolve its composite-type bookkeeping in the same transactional batch. + auto duplicate = + new Link("Expression", {nodes[0]->handle(), nodes[1]->handle(), nodes[2]->handle()}); + auto nested = new Link("Expression", {existing->handle(), nodes[3]->handle(), nodes[4]->handle()}); + + ASSERT_EQ(db->add_nodes(nodes, true).size(), 5u); + auto handles = db->add_links({duplicate, nested}, true, &SkipIfExistsMerger::instance()); + ASSERT_EQ(handles.size(), 2u); + EXPECT_EQ(handles[0], ""); + EXPECT_EQ(handles[1], nested->handle()); + + auto existing_doc = db->get_atom_document(existing->handle()); + ASSERT_NE(existing_doc, nullptr); + EXPECT_EQ(string(existing_doc->get("composite_type_hash")), existing_hash); + + auto nested_doc = db->get_atom_document(nested->handle()); + ASSERT_NE(nested_doc, nullptr); + ASSERT_EQ(nested_doc->get_size("composite_type"), 4u); + // Transactional map stores named_type_hash for link targets (same as + // build_composite_type_entries_map). + EXPECT_EQ(string(nested_doc->get("composite_type", 0)), nested->named_type_hash()); + EXPECT_EQ(string(nested_doc->get("composite_type", 1)), existing->named_type_hash()); + EXPECT_FALSE(string(nested_doc->get("composite_type", 1)).empty()); + EXPECT_EQ(string(nested_doc->get("composite_type", 2)), nodes[3]->named_type_hash()); + EXPECT_EQ(string(nested_doc->get("composite_type", 3)), nodes[4]->named_type_hash()); + EXPECT_EQ(string(nested_doc->get("composite_type_hash")), + Hasher::composite_handle({nested->named_type_hash(), + existing->named_type_hash(), + nodes[3]->named_type_hash(), + nodes[4]->named_type_hash()})); + + EXPECT_TRUE(db->delete_atom(nested->handle(), false)); + EXPECT_TRUE(db->delete_atom(existing->handle(), true)); + EXPECT_TRUE(db->delete_node(nodes[3]->handle())); + EXPECT_TRUE(db->delete_node(nodes[4]->handle())); + for (auto* node : nodes) { + delete node; + } + delete existing; + delete duplicate; + delete nested; +} + int main(int argc, char** argv) { ::testing::InitGoogleTest(&argc, argv); ::testing::AddGlobalTestEnvironment(new RedisMongoDBTestEnvironment()); diff --git a/src/tests/cpp/redis_mongodb_test_2.cc b/src/tests/cpp/redis_mongodb_test_2.cc index 8127e2a37..75f5abee4 100644 --- a/src/tests/cpp/redis_mongodb_test_2.cc +++ b/src/tests/cpp/redis_mongodb_test_2.cc @@ -241,15 +241,15 @@ TEST_F(RedisMongoDBTest, ConcurrentAddLinks) { links.push_back(link_with_nested); if (i % chunck_size == 0) { - db2->add_nodes(nodes, false, true); - db2->add_links(links, false, true); + db2->add_nodes(nodes, true); + db2->add_links(links, true); nodes.clear(); links.clear(); } } - if (!nodes.empty()) db2->add_nodes(nodes, false, true); - if (!links.empty()) db2->add_links(links, false, true); + if (!nodes.empty()) db2->add_nodes(nodes, true); + if (!links.empty()) db2->add_links(links, true); success_count++; } catch (const exception& e) { diff --git a/src/tests/cpp/remote_atomdb_test.cc b/src/tests/cpp/remote_atomdb_test.cc index 76e3f2ce0..454ee8c15 100644 --- a/src/tests/cpp/remote_atomdb_test.cc +++ b/src/tests/cpp/remote_atomdb_test.cc @@ -47,8 +47,8 @@ TEST_F(RemoteAtomDBPeerTest, AddAndGetNodes) { auto human = new Node("Symbol", "\"human\""); auto monkey = new Node("Symbol", "\"monkey\""); - string human_handle = peer_->add_node(human, false); - string monkey_handle = peer_->add_node(monkey, false); + string human_handle = peer_->add_node(human); + string monkey_handle = peer_->add_node(monkey); EXPECT_FALSE(human_handle.empty()); EXPECT_FALSE(monkey_handle.empty()); @@ -65,12 +65,12 @@ TEST_F(RemoteAtomDBPeerTest, AddAndGetLinks) { auto monkey = new Node("Symbol", "\"monkey\""); auto similarity = new Node("Symbol", "Similarity"); - string human_handle = peer_->add_node(human, false); - string monkey_handle = peer_->add_node(monkey, false); - string similarity_handle = peer_->add_node(similarity, false); + string human_handle = peer_->add_node(human); + string monkey_handle = peer_->add_node(monkey); + string similarity_handle = peer_->add_node(similarity); auto link = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); - string link_handle = peer_->add_link(link, false); + string link_handle = peer_->add_link(link); EXPECT_FALSE(link_handle.empty()); EXPECT_TRUE(peer_->link_exists(link_handle)); @@ -83,7 +83,7 @@ TEST_F(RemoteAtomDBPeerTest, AddAndGetLinks) { TEST_F(RemoteAtomDBPeerTest, GetFromCacheThenRemote) { // Add to remote only (bypass peer) auto human = new Node("Symbol", "\"human\""); - string human_handle = remote_->add_node(human, false); + string human_handle = remote_->add_node(human); // Peer should find it via remote auto retrieved = peer_->get_node(human_handle); @@ -98,7 +98,7 @@ TEST_F(RemoteAtomDBPeerTest, GetFromCacheThenRemote) { TEST_F(RemoteAtomDBPeerTest, PersistsToLocal) { auto human = new Node("Symbol", "\"human\""); - string human_handle = peer_->add_node(human, false); + string human_handle = peer_->add_node(human); // cache should have it EXPECT_TRUE(peer_->node_exists(human_handle)); @@ -115,15 +115,15 @@ TEST_F(RemoteAtomDBPeerTest, QueryForPattern) { auto mammal = new Node("Symbol", "\"mammal\""); auto inheritance = new Node("Symbol", "Inheritance"); - string human_handle = remote_->add_node(human, false); - string monkey_handle = remote_->add_node(monkey, false); - string mammal_handle = remote_->add_node(mammal, false); - string inheritance_handle = remote_->add_node(inheritance, false); + string human_handle = remote_->add_node(human); + string monkey_handle = remote_->add_node(monkey); + string mammal_handle = remote_->add_node(mammal); + string inheritance_handle = remote_->add_node(inheritance); auto link1 = new Link("Expression", {inheritance_handle, human_handle, mammal_handle}); auto link2 = new Link("Expression", {inheritance_handle, monkey_handle, mammal_handle}); - string link1_handle = remote_->add_link(link1, false); - string link2_handle = remote_->add_link(link2, false); + string link1_handle = remote_->add_link(link1); + string link2_handle = remote_->add_link(link2); remote_->re_index_patterns(true); @@ -158,12 +158,12 @@ TEST_F(RemoteAtomDBPeerTest, QueryForTargets) { auto node2 = new Node("Symbol", "Node2"); auto similarity = new Node("Symbol", "Similarity"); - string node1_handle = peer_->add_node(node1, false); - string node2_handle = peer_->add_node(node2, false); - string similarity_handle = peer_->add_node(similarity, false); + string node1_handle = peer_->add_node(node1); + string node2_handle = peer_->add_node(node2); + string similarity_handle = peer_->add_node(similarity); auto link = new Link("Expression", {similarity_handle, node1_handle, node2_handle}); - string link_handle = peer_->add_link(link, false); + string link_handle = peer_->add_link(link); auto targets = peer_->query_for_targets(link_handle); ASSERT_NE(targets, nullptr); @@ -178,12 +178,12 @@ TEST_F(RemoteAtomDBPeerTest, QueryForIncomingSet) { auto monkey = new Node("Symbol", "\"monkey\""); auto similarity = new Node("Symbol", "Similarity"); - string human_handle = peer_->add_node(human, false); - string monkey_handle = peer_->add_node(monkey, false); - string similarity_handle = peer_->add_node(similarity, false); + string human_handle = peer_->add_node(human); + string monkey_handle = peer_->add_node(monkey); + string similarity_handle = peer_->add_node(similarity); auto link = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); - string link_handle = peer_->add_link(link, false); + string link_handle = peer_->add_link(link); auto incoming = peer_->query_for_incoming_set(human_handle); ASSERT_NE(incoming, nullptr); @@ -200,12 +200,12 @@ TEST_F(RemoteAtomDBPeerTest, DeleteLink) { auto monkey = new Node("Symbol", "\"monkey\""); auto similarity = new Node("Symbol", "Similarity"); - string human_handle = peer_->add_node(human, false); - string monkey_handle = peer_->add_node(monkey, false); - string similarity_handle = peer_->add_node(similarity, false); + string human_handle = peer_->add_node(human); + string monkey_handle = peer_->add_node(monkey); + string similarity_handle = peer_->add_node(similarity); auto link = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); - string link_handle = peer_->add_link(link, false); + string link_handle = peer_->add_link(link); bool deleted = peer_->delete_link(link_handle, false); EXPECT_TRUE(deleted); @@ -223,15 +223,15 @@ TEST_F(RemoteAtomDBPeerTest, FetchAndRelease) { auto inheritance = new Node("Symbol", "Inheritance"); // Add nodes and links to the remote DB directly (bypass peer) - string human_handle = remote_->add_node(human, false); - string monkey_handle = remote_->add_node(monkey, false); - string mammal_handle = remote_->add_node(mammal, false); - string inheritance_handle = remote_->add_node(inheritance, false); + string human_handle = remote_->add_node(human); + string monkey_handle = remote_->add_node(monkey); + string mammal_handle = remote_->add_node(mammal); + string inheritance_handle = remote_->add_node(inheritance); auto link1 = new Link("Expression", {inheritance_handle, human_handle, mammal_handle}); auto link2 = new Link("Expression", {inheritance_handle, monkey_handle, mammal_handle}); - string link1_handle = remote_->add_link(link1, false); - string link2_handle = remote_->add_link(link2, false); + string link1_handle = remote_->add_link(link1); + string link2_handle = remote_->add_link(link2); remote_->re_index_patterns(true); LinkSchema link_schema({"LINK_TEMPLATE", @@ -281,12 +281,12 @@ TEST_F(RemoteAtomDBPeerTest, ReleaseWithoutLocalPersistence) { auto mammal = new Node("Symbol", "\"mammal\""); auto inheritance = new Node("Symbol", "Inheritance"); - string human_handle = remote->add_node(human, false); - string mammal_handle = remote->add_node(mammal, false); - string inheritance_handle = remote->add_node(inheritance, false); + string human_handle = remote->add_node(human); + string mammal_handle = remote->add_node(mammal); + string inheritance_handle = remote->add_node(inheritance); auto link = new Link("Expression", {inheritance_handle, human_handle, mammal_handle}); - string link_handle = remote->add_link(link, false); + string link_handle = remote->add_link(link); remote->re_index_patterns(true); LinkSchema link_schema({"LINK_TEMPLATE", @@ -322,14 +322,14 @@ TEST_F(RemoteAtomDBPeerTest, AtomsCount) { auto node2 = new Node("Symbol", "Node2"); auto similarity = new Node("Symbol", "Similarity"); - peer_->add_node(node1, false); - peer_->add_node(node2, false); - peer_->add_node(similarity, false); + peer_->add_node(node1); + peer_->add_node(node2); + peer_->add_node(similarity); EXPECT_EQ(peer_->atom_count(), 3); auto link1 = new Link("Expression", {similarity->handle(), node1->handle(), node2->handle()}); - peer_->add_link(link1, false); + peer_->add_link(link1); EXPECT_EQ(peer_->atom_count(), 4); EXPECT_EQ(peer_->empty(), false); @@ -411,7 +411,7 @@ TEST_F(RemoteAtomDBTest, GetPeer) { TEST_F(RemoteAtomDBTest, AddAndGetAcrossPeers) { auto human = new Node("Symbol", "\"human\""); - string human_handle = db_->add_node(human, false); + string human_handle = db_->add_node(human); EXPECT_FALSE(human_handle.empty()); EXPECT_TRUE(db_->node_exists(human_handle)); @@ -427,15 +427,15 @@ TEST_F(RemoteAtomDBTest, AddLinksAndRetrieve) { auto mammal = new Node("Symbol", "\"mammal\""); auto inheritance = new Node("Symbol", "Inheritance"); - string human_handle = db_->add_node(human, false); - string monkey_handle = db_->add_node(monkey, false); - string mammal_handle = db_->add_node(mammal, false); - string inheritance_handle = db_->add_node(inheritance, false); + string human_handle = db_->add_node(human); + string monkey_handle = db_->add_node(monkey); + string mammal_handle = db_->add_node(mammal); + string inheritance_handle = db_->add_node(inheritance); auto link1 = new Link("Expression", {inheritance_handle, human_handle, mammal_handle}); auto link2 = new Link("Expression", {inheritance_handle, monkey_handle, mammal_handle}); - string link1_handle = db_->add_link(link1, false); - string link2_handle = db_->add_link(link2, false); + string link1_handle = db_->add_link(link1); + string link2_handle = db_->add_link(link2); EXPECT_TRUE(db_->link_exists(link1_handle)); EXPECT_TRUE(db_->link_exists(link2_handle)); @@ -454,12 +454,12 @@ TEST_F(RemoteAtomDBTest, DeleteOperations) { auto monkey = new Node("Symbol", "\"monkey\""); auto similarity = new Node("Symbol", "Similarity"); - string human_handle = db_->add_node(human, false); - string monkey_handle = db_->add_node(monkey, false); - string similarity_handle = db_->add_node(similarity, false); + string human_handle = db_->add_node(human); + string monkey_handle = db_->add_node(monkey); + string similarity_handle = db_->add_node(similarity); auto link = new Link("Expression", {similarity_handle, human_handle, monkey_handle}); - string link_handle = db_->add_link(link, false); + string link_handle = db_->add_link(link); bool deleted = db_->delete_link(link_handle, false); EXPECT_TRUE(deleted); @@ -493,7 +493,7 @@ TEST_F(RemoteAtomDBConfigTest, SingleConfigWorks) { EXPECT_NE(peers.find("single_peer"), peers.end()); auto human = new Node("Symbol", "\"human\""); - string human_handle = db_->add_node(human, false); + string human_handle = db_->add_node(human); EXPECT_TRUE(db_->node_exists(human_handle)); } @@ -562,15 +562,15 @@ static vector populate_inheritance_links(shared_ptr backend) auto mammal = new Node("Symbol", "\"mammal\""); auto inheritance = new Node("Symbol", "Inheritance"); - string human_handle = backend->add_node(human, false); - string monkey_handle = backend->add_node(monkey, false); - string mammal_handle = backend->add_node(mammal, false); - string inheritance_handle = backend->add_node(inheritance, false); + string human_handle = backend->add_node(human); + string monkey_handle = backend->add_node(monkey); + string mammal_handle = backend->add_node(mammal); + string inheritance_handle = backend->add_node(inheritance); auto link1 = new Link("Expression", {inheritance_handle, human_handle, mammal_handle}); auto link2 = new Link("Expression", {inheritance_handle, monkey_handle, mammal_handle}); - string link1_handle = backend->add_link(link1, false); - string link2_handle = backend->add_link(link2, false); + string link1_handle = backend->add_link(link1); + string link2_handle = backend->add_link(link2); backend->re_index_patterns(true); return {link1_handle, link2_handle}; } @@ -684,7 +684,7 @@ TEST(RemoteAtomDBFederationTest, CacheFirstProbingAcrossPeers) { auto backend2 = make_shared("fed_cache_peer2_"); auto only_in_peer2 = new Node("Symbol", "\"only_in_peer2\""); - string handle = backend2->add_node(only_in_peer2, false); + string handle = backend2->add_node(only_in_peer2); map> peers; peers["peer1"] = make_shared(backend1, nullptr, "peer1"); diff --git a/src/tests/cpp/test_commons/mocks/MockAtomDB.h b/src/tests/cpp/test_commons/mocks/MockAtomDB.h index bb9c8c720..85c82d1c8 100644 --- a/src/tests/cpp/test_commons/mocks/MockAtomDB.h +++ b/src/tests/cpp/test_commons/mocks/MockAtomDB.h @@ -52,21 +52,21 @@ class AtomDBMock : public AtomDB { MOCK_METHOD(set, nodes_exist, (const vector& handles), (override)); MOCK_METHOD(set, links_exist, (const vector& link_handles), (override)); - MOCK_METHOD(string, add_node, (const Node* node, bool throw_if_exists), (override)); - MOCK_METHOD(string, add_link, (const Link* link, bool throw_if_exists), (override)); - MOCK_METHOD(string, add_atom, (const Atom* atom, bool throw_if_exists), (override)); + MOCK_METHOD(string, add_node, (const Node* node, const Merger* merger), (override)); + MOCK_METHOD(string, add_link, (const Link* link, const Merger* merger), (override)); + MOCK_METHOD(string, add_atom, (const Atom* atom, const Merger* merger), (override)); MOCK_METHOD(vector, add_atoms, - (const vector& atoms, bool throw_if_exists, bool is_transactional), + (const vector& atom_list, bool is_transactional, const Merger* merger), (override)); MOCK_METHOD(vector, add_nodes, - (const vector& nodes, bool throw_if_exists, bool is_transactional), + (const vector& nodes, bool is_transactional, const Merger* merger), (override)); MOCK_METHOD(vector, add_links, - (const vector& links, bool throw_if_exists, bool is_transactional), + (const vector& links, bool is_transactional, const Merger* merger), (override)); MOCK_METHOD(bool, delete_atom, (const string& handle, bool delete_link_targets), (override)); diff --git a/src/tests/main/evaluation_evolution.cc b/src/tests/main/evaluation_evolution.cc index 2cea63d75..1b7851265 100644 --- a/src/tests/main/evaluation_evolution.cc +++ b/src/tests/main/evaluation_evolution.cc @@ -399,7 +399,7 @@ static shared_ptr add_or_update_link(const string& type_handle, if (strength != old_link->custom_attributes.get_or(STRENGTH_TAG, 1)) { if (WRITE_CREATED_LINKS_TO_DB) { LOG_DEBUG("Updating Link in AtomDB"); - db->delete_link(handle, false); + // Default merger (NULL) upserts/replaces the existing atom. db->add_link(new_link.get()); } if (WRITE_CREATED_LINKS_TO_FILE) { @@ -992,7 +992,7 @@ static void add_preset_links(const vector& implication_to_target_predica auto link = std::dynamic_pointer_cast(parser_handler->element_stack.top()); link->custom_attributes["strength"] = (double) Utils::string_to_float(line[0]); LOG_DEBUG("Adding Link: [" + line[0] + "] " + line[1]); - db->add_link(link.get(), false); + db->add_link(link.get()); count++; line.clear(); buffer_determiners.push_back({link->handle(), link->targets[1], link->targets[2]}); @@ -1390,7 +1390,7 @@ static void insert_type_symbols() { Node* node; for (string node_name : to_insert) { node = new Node(SYMBOL, node_name); - db->add_node(node, false); + db->add_node(node); delete (node); } }