Merge branch 'master' into T595-notification-when-sync-replica-down

src/query/interpreter.cpp

@@ -1,4 +1,4 @@
-// Copyright 2022 Memgraph Ltd.
+// Copyright 2023 Memgraph Ltd.
 //
 // Use of this software is governed by the Business Source License
 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -2496,13 +2496,13 @@ void RunTriggersIndividually(const utils::SkipList<Trigger> &triggers, Interpret
 }
 }  // namespace
 
-void Interpreter::Commit() {
+Interpreter::TransactionCommitResponse Interpreter::Commit() {
   // It's possible that some queries did not finish because the user did
   // not pull all of the results from the query.
   // For now, we will not check if there are some unfinished queries.
   // We should document clearly that all results should be pulled to complete
   // a query.
-  if (!db_accessor_) return;
+  if (!db_accessor_) return Interpreter::TransactionCommitResponse{.success = true};
 
   std::optional<TriggerContext> trigger_context = std::nullopt;
   if (trigger_context_collector_) {
@@ -2590,8 +2590,15 @@ void Interpreter::Commit() {
 
   SPDLOG_DEBUG("Finished committing the transaction");
   if (!commit_confirmed_by_all_sync_repplicas) {
-    throw ReplicationException("At least one SYNC replica has not confirmed committing last transaction.");
+    std::vector<Notification> notifications;
+    notifications.reserve(1);
+    notifications.emplace_back(SeverityLevel::WARNING, NotificationCode::REPLICA_PORT_WARNING,
+                               "At least one SYNC replica has not confirmed committing last transaction.");
+
+    return TransactionCommitResponse{.success = true, .notifications = std::move(notifications)};
   }
+
+  return Interpreter::TransactionCommitResponse{.success = true};
 }
 
 void Interpreter::AdvanceCommand() {

src/query/interpreter.hpp

@@ -1,4 +1,4 @@
-// Copyright 2022 Memgraph Ltd.
+// Copyright 2023 Memgraph Ltd.
 //
 // Use of this software is governed by the Business Source License
 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -326,6 +326,11 @@ class Interpreter final {
     }
   };
 
+  struct TransactionCommitResponse {
+    bool success;
+    std::vector<Notification> notifications;
+  };
+
   // Interpreter supports multiple prepared queries at the same time.
   // The client can reference a specific query for pull using an arbitrary qid
   // which is in our case the index of the query in the vector.
@@ -354,7 +359,7 @@ class Interpreter final {
   std::optional<storage::IsolationLevel> next_transaction_isolation_level;
 
   PreparedQuery PrepareTransactionQuery(std::string_view query_upper);
-  void Commit();
+  TransactionCommitResponse Commit();
   void AdvanceCommand();
   void AbortCommand(std::unique_ptr<QueryExecution> *query_execution);
   std::optional<storage::IsolationLevel> GetIsolationLevelOverride();
@@ -402,28 +407,34 @@ std::map<std::string, TypedValue> Interpreter::Pull(TStream *result_stream, std:
     if (maybe_res) {
       // Save its summary
       maybe_summary.emplace(std::move(query_execution->summary));
-      if (!query_execution->notifications.empty()) {
-        std::vector<TypedValue> notifications;
-        notifications.reserve(query_execution->notifications.size());
-        for (const auto &notification : query_execution->notifications) {
-          notifications.emplace_back(notification.ConvertToMap());
-        }
-        maybe_summary->insert_or_assign("notifications", std::move(notifications));
+      std::vector<TypedValue> notifications;
+
+      notifications.reserve(query_execution->notifications.size());
+      for (const auto &notification : query_execution->notifications) {
+        notifications.emplace_back(notification.ConvertToMap());
       }
+
       if (!in_explicit_transaction_) {
         switch (*maybe_res) {
-          case QueryHandlerResult::COMMIT:
-            Commit();
+          case QueryHandlerResult::COMMIT: {
+            auto commit_response = Commit();
+            for (const auto &notification : commit_response.notifications) {
+              notifications.emplace_back(notification.ConvertToMap());
+            }
+
             break;
-          case QueryHandlerResult::ABORT:
+          }
+          case QueryHandlerResult::ABORT: {
             Abort();
             break;
-          case QueryHandlerResult::NOTHING:
+          }
+          case QueryHandlerResult::NOTHING: {
             // The only cases in which we have nothing to do are those where
             // we're either in an explicit transaction or the query is such that
             // a transaction wasn't started on a call to `Prepare()`.
             MG_ASSERT(in_explicit_transaction_ || !db_accessor_);
             break;
+          }
         }
         // As the transaction is done we can clear all the executions
         // NOTE: we cannot clear query_execution inside the Abort and Commit
@@ -435,6 +446,10 @@ std::map<std::string, TypedValue> Interpreter::Pull(TStream *result_stream, std:
         // in the transaction can be in unfinished state
         query_execution.reset(nullptr);
       }
+
+      if (!notifications.empty()) {
+        maybe_summary->insert_or_assign("notifications", std::move(notifications));
+      }
     }
   } catch (const ExplicitTransactionUsageException &) {
     query_execution.reset(nullptr);

src/query/metadata.cpp

@@ -1,4 +1,4 @@
-// Copyright 2022 Memgraph Ltd.
+// Copyright 2023 Memgraph Ltd.
 //
 // Use of this software is governed by the Business Source License
 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -42,6 +42,8 @@ constexpr std::string_view GetCodeString(const NotificationCode code) {
       return "CheckStream"sv;
     case NotificationCode::CREATE_TRIGGER:
       return "CreateTrigger"sv;
+    case NotificationCode::COMMIT_TO_REPLICAS:
+      return "CommitToReplicas"sv;
     case NotificationCode::DROP_CONSTRAINT:
       return "DropConstraint"sv;
     case NotificationCode::DROP_REPLICA:

src/query/metadata.hpp

@@ -1,4 +1,4 @@
-// Copyright 2022 Memgraph Ltd.
+// Copyright 2023 Memgraph Ltd.
 //
 // Use of this software is governed by the Business Source License
 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -24,9 +24,10 @@ namespace memgraph::query {
 enum class SeverityLevel : uint8_t { INFO, WARNING };
 
 enum class NotificationCode : uint8_t {
+  CHECK_STREAM,
+  COMMIT_TO_REPLICAS,
   CREATE_CONSTRAINT,
   CREATE_INDEX,
-  CHECK_STREAM,
   CREATE_STREAM,
   CREATE_TRIGGER,
   DROP_CONSTRAINT,

tests/e2e/replication/show_while_creating_invalid_state.py

@@ -9,17 +9,16 @@
 # by the Apache License, Version 2.0, included in the file
 # licenses/APL.txt.
 
-import sys
-
 import os
-import pytest
 import random
+import sys
+import tempfile
 
-from common import execute_and_fetch_all
-from mg_utils import mg_sleep_and_assert
 import interactive_mg_runner
 import mgclient
-import tempfile
+import pytest
+from common import execute_and_fetch_all
+from mg_utils import mg_sleep_and_assert
 
 interactive_mg_runner.SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
 interactive_mg_runner.PROJECT_DIR = os.path.normpath(
@@ -286,10 +285,7 @@ def test_basic_recovery(connection):
     assert actual_data == expected_data
 
     # 12/
-    with pytest.raises(mgclient.DatabaseError):
-        interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(
-            "CREATE (p1:Number {name:'Magic_again_again', value:44})"
-        )
+    interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query("CREATE (p1:Number {name:'Magic_again_again', value:44})")
     expected_data = {
         ("replica_1", "127.0.0.1:10001", "sync", 0, 0, "invalid"),
         ("replica_2", "127.0.0.1:10002", "sync", 9, 0, "ready"),
@@ -709,8 +705,7 @@ def test_attempt_to_write_data_on_main_when_sync_replica_is_down():
     assert actual_data == expected_data
 
     # 4/
-    with pytest.raises(mgclient.DatabaseError):
-        interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query("CREATE (p:Number {name:2});")
+    interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query("CREATE (p:Number {name:2});")
 
     res_from_main = interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_TO_CHECK)
     assert len(res_from_main) == 2
@@ -994,8 +989,7 @@ def test_trigger_on_create_before_commit_with_offline_sync_replica():
     assert actual_data == expected_data
 
     # 6/
-    with pytest.raises(mgclient.DatabaseError):
-        interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_CREATE_NODE)
+    interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_CREATE_NODE)
 
     # 7/
     res_from_main = interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_TO_CHECK)
@@ -1098,8 +1092,7 @@ def test_trigger_on_update_before_commit_with_offline_sync_replica():
     assert actual_data == expected_data
 
     # 7/
-    with pytest.raises(mgclient.DatabaseError):
-        interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_TO_UPDATE)
+    interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_TO_UPDATE)
 
     # 8/
     res_from_main = interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_TO_CHECK)
@@ -1206,8 +1199,7 @@ def test_trigger_on_delete_before_commit_with_offline_sync_replica():
     assert actual_data == expected_data
 
     # 7/
-    with pytest.raises(mgclient.DatabaseError):
-        interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_TO_DELETE)
+    interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_TO_DELETE)
 
     # 8/
     res_from_main = interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_TO_CHECK)
@@ -1313,8 +1305,7 @@ def test_trigger_on_create_before_and_after_commit_with_offline_sync_replica():
     assert actual_data == expected_data
 
     # 6/
-    with pytest.raises(mgclient.DatabaseError):
-        interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_CREATE_NODE)
+    interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_CREATE_NODE)
 
     # 7/
     res_from_main = interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_TO_CHECK)
@@ -1418,8 +1409,7 @@ def test_triggers_on_create_before_commit_with_offline_sync_replica():
     assert actual_data == expected_data
 
     # 6/
-    with pytest.raises(mgclient.DatabaseError):
-        interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_CREATE_NODE)
+    interactive_mg_runner.MEMGRAPH_INSTANCES["main"].query(QUERY_CREATE_NODE)
 
     # 7/
     def get_number_of_nodes():