New API and optimizations added to MutableSubscriptionSet (#7008)

* New API and optimizations added to MutableSubscriptionSet

* Changes after code review

* Update changelog

Author: danieltabacaru

CHANGELOG.md

@@ -3,6 +3,12 @@
 ### Enhancements
 * <New feature description> (PR [#????](https://github.com/realm/realm-core/pull/????))
 * Allow collections of non-embedded links in asymmetric objects. ([PR #7003](https://github.com/realm/realm-core/pull/7003))
+* Flexible sync API improvements:
+  - Erase Subscriptions by class type for C API. 
+  - `MutableSubscriptionSet::erase(iterator)` now runs in constant time.
+  - Introduce `MutableSubscriptionSet::erase_by_id()`.
+  - Introduce `MutableSubscriptionSet::erase_by_class_name()`.
+  ([PR #7008](https://github.com/realm/realm-core/pull/7008))
 
 ### Fixed
 * <How do the end-user experience this issue? what was the impact?> ([#????](https://github.com/realm/realm-core/issues/????), since v?.?.?)

src/realm.h

@@ -3691,6 +3691,12 @@ RLM_API bool realm_sync_subscription_set_erase_by_query(realm_flx_sync_mutable_s
  */
 RLM_API bool realm_sync_subscription_set_erase_by_results(realm_flx_sync_mutable_subscription_set_t*,
                                                           realm_results_t*, bool* erased);
+/**
+ *  Remove all subscriptions for a given class type. If operation completes successfully set the bool out param.
+ *  @return true if no error occurred, false otherwise (use realm_get_last_error for fetching the error).
+ */
+RLM_API bool realm_sync_subscription_set_erase_by_class_name(realm_flx_sync_mutable_subscription_set_t*, const char*,
+                                                             bool* erased);
 /**
  *  Commit the subscription_set passed as parameter (in order that all the changes made will take effect)
  *  @return pointer to a valid immutable subscription if commit was successful

src/realm/object-store/c_api/sync.cpp

@@ -609,13 +609,7 @@ RLM_API bool realm_sync_subscription_set_erase_by_id(realm_flx_sync_mutable_subs
     REALM_ASSERT(subscription_set != nullptr && id != nullptr);
     *erased = false;
     return wrap_err([&] {
-        auto it = std::find_if(subscription_set->begin(), subscription_set->end(), [id](const Subscription& sub) {
-            return from_capi(*id) == sub.id;
-        });
-        if (it != subscription_set->end()) {
-            subscription_set->erase(it);
-            *erased = true;
-        }
+        *erased = subscription_set->erase_by_id(from_capi(*id));
         return true;
     });
 }
@@ -655,6 +649,18 @@ RLM_API bool realm_sync_subscription_set_erase_by_results(realm_flx_sync_mutable
     });
 }
 
+RLM_API bool
+realm_sync_subscription_set_erase_by_class_name(realm_flx_sync_mutable_subscription_set_t* subscription_set,
+                                                const char* object_class_name, bool* erased)
+{
+    REALM_ASSERT(subscription_set != nullptr && object_class_name != nullptr);
+    *erased = false;
+    return wrap_err([&]() {
+        *erased = subscription_set->erase_by_class_name(object_class_name);
+        return true;
+    });
+}
+
 RLM_API realm_flx_sync_subscription_set_t*
 realm_sync_subscription_set_commit(realm_flx_sync_mutable_subscription_set_t* subscription_set)
 {

src/realm/sync/subscriptions.cpp

@@ -33,6 +33,8 @@
 #include <initializer_list>
 #include <stdexcept>
 
+#include <algorithm>
+
 namespace realm::sync {
 namespace {
 // Schema version history:
@@ -273,11 +275,24 @@ void MutableSubscriptionSet::check_is_mutable() const
     }
 }
 
+// This uses the 'swap and pop' idiom to run in constant time.
+// The iterator returned is:
+//  1. end(), if the last subscription is removed
+//  2. same iterator it is passed (but pointing to the last subscription in set), otherwise
 MutableSubscriptionSet::iterator MutableSubscriptionSet::erase(const_iterator it)
 {
     check_is_mutable();
     REALM_ASSERT(it != end());
-    return m_subs.erase(it);
+    if (it == std::prev(m_subs.end())) {
+        m_subs.pop_back();
+        return end();
+    }
+    auto back = std::prev(m_subs.end());
+    // const_iterator to iterator in constant time (See https://stackoverflow.com/a/10669041)
+    auto iterator = m_subs.erase(it, it);
+    std::swap(*iterator, *back);
+    m_subs.pop_back();
+    return iterator;
 }
 
 bool MutableSubscriptionSet::erase(StringData name)
@@ -286,7 +301,8 @@ bool MutableSubscriptionSet::erase(StringData name)
     auto ptr = find(name);
     if (!ptr)
         return false;
-    m_subs.erase(m_subs.begin() + (ptr - &m_subs.front()));
+    auto it = m_subs.begin() + (ptr - &m_subs.front());
+    erase(it);
     return true;
 }
 
@@ -296,7 +312,31 @@ bool MutableSubscriptionSet::erase(const Query& query)
     auto ptr = find(query);
     if (!ptr)
         return false;
-    m_subs.erase(m_subs.begin() + (ptr - &m_subs.front()));
+    auto it = m_subs.begin() + (ptr - &m_subs.front());
+    erase(it);
+    return true;
+}
+
+bool MutableSubscriptionSet::erase_by_class_name(StringData object_class_name)
+{
+    // TODO: Use std::erase_if when switching to C++20.
+    auto it = std::remove_if(m_subs.begin(), m_subs.end(), [&object_class_name](const Subscription& sub) {
+        return sub.object_class_name == object_class_name;
+    });
+    auto erased = end() - it;
+    m_subs.erase(it, end());
+    return erased > 0;
+}
+
+bool MutableSubscriptionSet::erase_by_id(ObjectId id)
+{
+    auto it = std::find_if(m_subs.begin(), m_subs.end(), [&id](const Subscription& sub) -> bool {
+        return sub.id == id;
+    });
+    if (it == end()) {
+        return false;
+    }
+    erase(it);
     return true;
 }
 

src/realm/sync/subscriptions.hpp

@@ -249,6 +249,9 @@ class MutableSubscriptionSet : public SubscriptionSet {
     bool erase(StringData name);
     bool erase(const Query& query);
 
+    bool erase_by_class_name(StringData object_class_name);
+    bool erase_by_id(ObjectId id);
+
     // Updates the state of the transaction and optionally updates its error information.
     //
     // You may only set an error_str when the State is State::Error.

test/object-store/c_api/c_api.cpp

@@ -5991,6 +5991,27 @@ TEST_CASE("app: flx-sync basic tests", "[sync][flx][c_api][baas]") {
             realm_release(sub_c_1);
         }
 
+        {
+            auto sub = realm_sync_get_latest_subscription_set(&c_wrap_realm);
+            auto mut_sub = realm_sync_make_subscription_set_mutable(sub);
+            std::size_t index = -1;
+            bool inserted = false;
+            CHECK(realm_sync_subscription_set_insert_or_assign_query(mut_sub, c_wrap_query_bar, nullptr, &index,
+                                                                     &inserted));
+            CHECK(inserted);
+            CHECK(realm_sync_subscription_set_insert_or_assign_query(mut_sub, c_wrap_query_foo, nullptr, &index,
+                                                                     &inserted));
+            CHECK(inserted);
+            bool erased = false;
+            CHECK(realm_sync_subscription_set_erase_by_class_name(mut_sub, "Obj", &erased));
+            CHECK(erased);
+            // Nothing to remove when trying again.
+            CHECK(realm_sync_subscription_set_erase_by_class_name(mut_sub, "Obj", &erased));
+            CHECK_FALSE(erased);
+            realm_release(sub);
+            realm_release(mut_sub);
+        }
+
         realm_release(c_wrap_query_foo);
         realm_release(c_wrap_query_bar);
     });

test/test_sync_subscriptions.cpp

@@ -844,4 +844,115 @@ TEST(Sync_SyncMetadataSchemaVersions_LegacyTable)
     }
 }
 
+TEST(Sync_MutableSubscriptionSetOperations)
+{
+    SHARED_GROUP_TEST_PATH(sub_store_path);
+    SubscriptionStoreFixture fixture(sub_store_path);
+    auto store = SubscriptionStore::create(fixture.db, [](int64_t) {});
+
+    auto read_tr = fixture.db->start_read();
+    Query query_a(read_tr->get_table("class_a"));
+    query_a.greater_equal(fixture.bar_col, int64_t(1));
+    Query query_b(read_tr->get_table(fixture.a_table_key));
+    query_b.equal(fixture.foo_col, "Realm");
+    Query query_c(read_tr->get_table(fixture.a_table_key));
+
+    // insert_or_assign
+    {
+        auto out = store->get_latest().make_mutable_copy();
+        auto [it, inserted] = out.insert_or_assign("a sub", query_a);
+        CHECK(inserted);
+        auto named_id = it->id;
+        out.insert_or_assign("b sub", query_b);
+        CHECK_EQUAL(out.size(), 2);
+        std::tie(it, inserted) = out.insert_or_assign("a sub", query_a);
+        CHECK_NOT(inserted);
+        CHECK_EQUAL(it->id, named_id);
+        CHECK_EQUAL(out.size(), 2);
+    }
+
+    // find
+    {
+        auto out = store->get_latest().make_mutable_copy();
+        auto [it, inserted] = out.insert_or_assign("a sub", query_a);
+        std::tie(it, inserted) = out.insert_or_assign("b sub", query_b);
+        CHECK(out.find(query_b));
+        CHECK(out.find("a sub"));
+    }
+
+    // erase
+    {
+        auto out = store->get_latest().make_mutable_copy();
+        out.insert_or_assign("a sub", query_a);
+        out.insert_or_assign("b sub", query_b);
+        out.insert_or_assign("c sub", query_c);
+        CHECK_EQUAL(out.size(), 3);
+        auto it = out.erase(out.begin());
+        // Iterator points to last query inserted due do "swap and pop" idiom.
+        CHECK_EQUAL(it->query_string, query_c.get_description());
+        CHECK_EQUAL(out.size(), 2);
+        CHECK_NOT(out.erase("a sub"));
+        CHECK_EQUAL(out.size(), 2);
+        CHECK(out.erase(query_b));
+        CHECK_EQUAL(out.size(), 1);
+        CHECK(out.erase("c sub"));
+        CHECK_EQUAL(out.size(), 0);
+    }
+
+    // erase_by_class_name
+    {
+        auto out = store->get_latest().make_mutable_copy();
+        out.insert_or_assign("a sub", query_a);
+        out.insert_or_assign("b sub", query_b);
+        out.insert_or_assign("c sub", query_c);
+        // Nothing to erase.
+        CHECK_NOT(out.erase_by_class_name("foo"));
+        // Erase all queries for the class type of the first query.
+        CHECK(out.erase_by_class_name(out.begin()->object_class_name));
+        // No queries left.
+        CHECK_EQUAL(out.size(), 0);
+    }
+
+    // erase_by_id
+    {
+        auto out = store->get_latest().make_mutable_copy();
+        out.insert_or_assign("a sub", query_a);
+        out.insert_or_assign("b sub", query_b);
+        // Nothing to erase.
+        CHECK_NOT(out.erase_by_id(ObjectId::gen()));
+        // Erase first query.
+        CHECK(out.erase_by_id(out.begin()->id));
+        CHECK_EQUAL(out.size(), 1);
+    }
+
+    // clear
+    {
+        auto out = store->get_latest().make_mutable_copy();
+        out.insert_or_assign("a sub", query_a);
+        out.insert_or_assign("b sub", query_b);
+        out.insert_or_assign("c sub", query_c);
+        CHECK_EQUAL(out.size(), 3);
+        out.clear();
+        CHECK_EQUAL(out.size(), 0);
+    }
+
+    // import
+    {
+        auto out = store->get_latest().make_mutable_copy();
+        out.insert_or_assign("a sub", query_a);
+        out.insert_or_assign("b sub", query_b);
+        auto subs = out.commit();
+
+        // This is an empty subscription set.
+        auto out2 = store->get_active().make_mutable_copy();
+        out2.insert_or_assign("c sub", query_c);
+        out2.import(subs);
+        // "c sub" is erased when 'import' is used.
+        CHECK_EQUAL(out2.size(), 2);
+        // insert "c sub" again.
+        out2.insert_or_assign("c sub", query_c);
+        CHECK_EQUAL(out2.size(), 3);
+    }
+}
+
 } // namespace realm::sync