Make metadata collection setable with flag

There might be a performance impect of updating the metadata store on
bulk operations. Hence this flag which is disabling the collection by
default. If the queries to obtain the information are called with this
flag disabled, the database will throw an exception.

config/flags.yaml

@@ -111,6 +111,10 @@ modifications:
     value: "false"
     override: true
 
+  - name: "storage_enable_schema_metadata"
+    value: "false"
+    override: true
+
   - name: "query_callable_mappings_path"
     value: "/etc/memgraph/apoc_compatibility_mappings.json"
     override: true

src/flags/general.cpp

@@ -114,6 +114,10 @@ DEFINE_uint64(storage_recovery_thread_count,
                        memgraph::storage::Config::Durability().recovery_thread_count),
               "The number of threads used to recover persisted data from disk.");
 
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DEFINE_bool(storage_enable_schema_metadata, false,
+            "Controls whether metadata should be collected about the resident labels and edge types.");
+
 #ifdef MG_ENTERPRISE
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
 DEFINE_bool(storage_delete_on_drop, true,

src/flags/general.hpp

@@ -77,6 +77,8 @@ DECLARE_uint64(storage_items_per_batch);
 DECLARE_bool(storage_parallel_index_recovery);
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
 DECLARE_uint64(storage_recovery_thread_count);
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DECLARE_bool(storage_enable_schema_metadata);
 #ifdef MG_ENTERPRISE
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
 DECLARE_bool(storage_delete_on_drop);

src/memgraph.cpp

@@ -291,7 +291,8 @@ int main(int argc, char **argv) {
   memgraph::storage::Config db_config{
       .gc = {.type = memgraph::storage::Config::Gc::Type::PERIODIC,
              .interval = std::chrono::seconds(FLAGS_storage_gc_cycle_sec)},
-      .items = {.properties_on_edges = FLAGS_storage_properties_on_edges},
+      .items = {.properties_on_edges = FLAGS_storage_properties_on_edges,
+                .enable_schema_metadata = FLAGS_storage_enable_schema_metadata},
       .durability = {.storage_directory = FLAGS_data_directory,
                      .recover_on_startup = FLAGS_storage_recover_on_startup || FLAGS_data_recovery_on_startup,
                      .snapshot_retention_count = FLAGS_storage_snapshot_retention_count,

src/query/interpreter.cpp

@@ -3049,6 +3049,11 @@ PreparedQuery PrepareDatabaseInfoQuery(ParsedQuery parsed_query, bool in_explici
     case DatabaseInfoQuery::InfoType::EDGE_TYPES: {
       header = {"edge types"};
       handler = [storage = current_db.db_acc_->get()->storage(), dba] {
+        if (!storage->config_.items.enable_schema_metadata) {
+          throw QueryRuntimeException(
+              "The metadata collection for edge-types is disabled. To enable it, restart your instance and set the "
+              "storage-enable-schema-metadata flag to True.");
+        }
         auto edge_types = dba->ListAllPossiblyPresentEdgeTypes();
         std::vector<std::vector<TypedValue>> results;
         results.reserve(edge_types.size());
@@ -3064,6 +3069,11 @@ PreparedQuery PrepareDatabaseInfoQuery(ParsedQuery parsed_query, bool in_explici
     case DatabaseInfoQuery::InfoType::NODE_LABELS: {
       header = {"node labels"};
       handler = [storage = current_db.db_acc_->get()->storage(), dba] {
+        if (!storage->config_.items.enable_schema_metadata) {
+          throw QueryRuntimeException(
+              "The metadata collection for node-labels is disabled. To enable it, restart your instance and set the "
+              "storage-enable-schema-metadata flag to True.");
+        }
         auto node_labels = dba->ListAllPossiblyPresentVertexLabels();
         std::vector<std::vector<TypedValue>> results;
         results.reserve(node_labels.size());

src/storage/v2/config.hpp

@@ -40,6 +40,7 @@ struct Config {
 
   struct Items {
     bool properties_on_edges{true};
+    bool enable_schema_metadata{false};
     friend bool operator==(const Items &lrh, const Items &rhs) = default;
   } items;
 

src/storage/v2/disk/storage.cpp

@@ -944,7 +944,9 @@ Result<EdgeAccessor> DiskStorage::DiskAccessor::CreateEdge(VertexAccessor *from,
   transaction_.manyDeltasCache.Invalidate(from_vertex, edge_type, EdgeDirection::OUT);
   transaction_.manyDeltasCache.Invalidate(to_vertex, edge_type, EdgeDirection::IN);
 
-  storage_->stored_edge_types_.try_insert(edge_type);
+  if (storage_->config_.items.enable_schema_metadata) {
+    storage_->stored_edge_types_.try_insert(edge_type);
+  }
   storage_->edge_count_.fetch_add(1, std::memory_order_acq_rel);
 
   return EdgeAccessor(edge, edge_type, from_vertex, to_vertex, storage_, &transaction_);

src/storage/v2/inmemory/storage.cpp

@@ -278,7 +278,9 @@ Result<EdgeAccessor> InMemoryStorage::InMemoryAccessor::CreateEdge(VertexAccesso
     if (to_vertex->deleted) return Error::DELETED_OBJECT;
   }
 
-  storage_->stored_edge_types_.try_insert(edge_type);
+  if (storage_->config_.items.enable_schema_metadata) {
+    storage_->stored_edge_types_.try_insert(edge_type);
+  }
   auto *mem_storage = static_cast<InMemoryStorage *>(storage_);
   auto gid = storage::Gid::FromUint(mem_storage->edge_id_.fetch_add(1, std::memory_order_acq_rel));
   EdgeRef edge(gid);
@@ -343,7 +345,9 @@ Result<EdgeAccessor> InMemoryStorage::InMemoryAccessor::CreateEdgeEx(VertexAcces
     if (to_vertex->deleted) return Error::DELETED_OBJECT;
   }
 
-  storage_->stored_edge_types_.try_insert(edge_type);
+  if (storage_->config_.items.enable_schema_metadata) {
+    storage_->stored_edge_types_.try_insert(edge_type);
+  }
 
   // NOTE: When we update the next `edge_id_` here we perform a RMW
   // (read-modify-write) operation that ISN'T atomic! But, that isn't an issue

src/storage/v2/vertex_accessor.cpp

@@ -109,7 +109,10 @@ Result<bool> VertexAccessor::AddLabel(LabelId label) {
 
   CreateAndLinkDelta(transaction_, vertex_, Delta::RemoveLabelTag(), label);
   vertex_->labels.push_back(label);
-  storage_->stored_node_labels_.try_insert(label);
+
+  if (storage_->config_.items.enable_schema_metadata) {
+    storage_->stored_node_labels_.try_insert(label);
+  }
 
   /// TODO: some by pointers, some by reference => not good, make it better
   storage_->constraints_.unique_constraints_->UpdateOnAddLabel(label, *vertex_, transaction_->start_timestamp);

tests/e2e/configuration/default_config.py

@@ -115,6 +115,11 @@ startup_config_dict = {
         "false",
         "Controls whether the index creation can be done in a multithreaded fashion.",
     ),
+    "storage_enable_schema_metadata": (
+        "false",
+        "false",
+        "Controls whether metadata should be collected about the resident labels and edge types.",
+    ),
     "password_encryption_algorithm": ("bcrypt", "bcrypt", "The password encryption algorithm used for authentication."),
     "pulsar_service_url": ("", "", "Default URL used while connecting to Pulsar brokers."),
     "query_execution_timeout_sec": (