Feature/hybrid smart graph test setup helper (#14455)

arangodb · mchacki · Sep 10, 2021 · Apr 22, 2021 · Apr 22, 2021 · Apr 22, 2021
commit fa37f394eca575444308c230da5936c281fbfa9f
diff --git a/arangod/Aql/EngineInfoContainerDBServerServerBased.cpp b/arangod/Aql/EngineInfoContainerDBServerServerBased.cpp
@@ -36,8 +36,8 @@
 #include "StorageEngine/TransactionState.h"
 #include "Utils/CollectionNameResolver.h"
 
-#include <set>
 #include <velocypack/Collection.h>
+#include <set>
 
 using namespace arangodb;
 using namespace arangodb::aql;
@@ -80,10 +80,13 @@ EngineInfoContainerDBServerServerBased::TraverserEngineShardLists::TraverserEngi
   auto const& edges = _node->edgeColls();
   TRI_ASSERT(!edges.empty());
   auto const& restrictToShards = query.queryOptions().restrictToShards;
+#ifdef USE_ENTERPRISE
+  transaction::Methods trx{query.newTrxContext()};
+#endif
   // Extract the local shards for edge collections.
   for (auto const& col : edges) {
 #ifdef USE_ENTERPRISE
-    if (query.trxForOptimization().isInaccessibleCollection(col->id())) {
+    if (trx.isInaccessibleCollection(col->id())) {
       _inaccessible.insert(col->name());
       _inaccessible.insert(std::to_string(col->id().id()));
     }
@@ -100,7 +103,7 @@ EngineInfoContainerDBServerServerBased::TraverserEngineShardLists::TraverserEngi
   // Or if we guarantee to never read vertex data.
   for (auto const& col : vertices) {
 #ifdef USE_ENTERPRISE
-    if (query.trxForOptimization().isInaccessibleCollection(col->id())) {
+    if (trx.isInaccessibleCollection(col->id())) {
       _inaccessible.insert(col->name());
       _inaccessible.insert(std::to_string(col->id().id()));
     }
@@ -112,8 +115,8 @@ EngineInfoContainerDBServerServerBased::TraverserEngineShardLists::TraverserEngi
 }
 
 std::vector<ShardID> EngineInfoContainerDBServerServerBased::TraverserEngineShardLists::getAllLocalShards(
-    std::unordered_map<ShardID, ServerID> const& shardMapping,
-    ServerID const& server, std::shared_ptr<std::vector<std::string>> shardIds, bool colIsSatellite) {
+    std::unordered_map<ShardID, ServerID> const& shardMapping, ServerID const& server,
+    std::shared_ptr<std::vector<std::string>> shardIds, bool colIsSatellite) {
   std::vector<ShardID> localShards;
   for (auto const& shard : *shardIds) {
     auto const& it = shardMapping.find(shard);
@@ -255,8 +258,7 @@ void EngineInfoContainerDBServerServerBased::closeSnippet(QueryId inputSnippet)
 }
 
 std::vector<bool> EngineInfoContainerDBServerServerBased::buildEngineInfo(
-    QueryId clusterQueryId,
-    VPackBuilder& infoBuilder, ServerID const& server,
+    QueryId clusterQueryId, VPackBuilder& infoBuilder, ServerID const& server,
     std::unordered_map<ExecutionNodeId, ExecutionNode*> const& nodesById,
     std::map<ExecutionNodeId, ExecutionNodeId>& nodeAliases) {
   LOG_TOPIC("4bbe6", DEBUG, arangodb::Logger::AQL)
@@ -322,19 +324,19 @@ arangodb::futures::Future<Result> EngineInfoContainerDBServerServerBased::buildS
 
   auto buildCallback =
       [this, server, serverDest, didCreateEngine = std::move(didCreateEngine),
-       &serverToQueryId, &serverToQueryIdLock, &snippetIds, globalId](
-          arangodb::futures::Try<arangodb::network::Response> const& response) -> Result {
+       &serverToQueryId, &serverToQueryIdLock, &snippetIds,
+       globalId](arangodb::futures::Try<arangodb::network::Response> const& response) -> Result {
     auto const& resolvedResponse = response.get();
     auto queryId = globalId;
-    
+
     std::unique_lock<std::mutex> guard{serverToQueryIdLock};
 
     if (resolvedResponse.fail()) {
       Result res = resolvedResponse.combinedResult();
       LOG_TOPIC("f9a77", DEBUG, Logger::AQL)
           << server << " responded with " << res.errorNumber() << ": "
           << res.errorMessage();
-    
+
       serverToQueryId.emplace_back(serverDest, globalId);
       return res;
     }
@@ -351,8 +353,8 @@ arangodb::futures::Future<Result> EngineInfoContainerDBServerServerBased::buildS
   };
 
   return network::sendRequestRetry(pool, serverDest, fuerte::RestVerb::Post,
-                              "/_api/aql/setup", std::move(buffer), options,
-                              std::move(headers))
+                                   "/_api/aql/setup", std::move(buffer),
+                                   options, std::move(headers))
       .then([buildCallback = std::move(buildCallback)](futures::Try<network::Response>&& resp) mutable {
         return buildCallback(resp);
       });
@@ -373,7 +375,8 @@ bool EngineInfoContainerDBServerServerBased::isNotSatelliteLeader(VPackSlice inf
     return true;
   }
 
-  TRI_ASSERT((infoSlice.get("snippets").isObject() && !infoSlice.get("snippets").isEmptyObject()) ||
+  TRI_ASSERT((infoSlice.get("snippets").isObject() &&
+              !infoSlice.get("snippets").isEmptyObject()) ||
              infoSlice.hasKey("traverserEngines"));
 
   return false;
@@ -406,7 +409,7 @@ Result EngineInfoContainerDBServerServerBased::buildEngines(
   TRI_ASSERT(!_closedSnippets.empty() || !_graphNodes.empty());
 
   ErrorCode cleanupReason = TRI_ERROR_CLUSTER_TIMEOUT;
-  
+
   auto cleanupGuard = scopeGuard([this, &serverToQueryId, &cleanupReason]() {
     // Fire and forget
     std::ignore = cleanupEngines(cleanupReason, _query.vocbase().name(), serverToQueryId);
@@ -427,19 +430,21 @@ Result EngineInfoContainerDBServerServerBased::buildEngines(
   options.timeout = network::Timeout(SETUP_TIMEOUT);
   options.skipScheduler = true;  // hack to speed up future.get()
   options.param("ttl", std::to_string(_query.queryOptions().ttl));
-  
+
   TRI_IF_FAILURE("Query::setupTimeout") {
-    options.timeout = network::Timeout(0.01 + (double) RandomGenerator::interval(uint32_t(10)));
+    options.timeout =
+        network::Timeout(0.01 + (double)RandomGenerator::interval(uint32_t(10)));
   }
-  
+
   TRI_IF_FAILURE("Query::setupTimeoutFailSequence") {
     options.timeout = network::Timeout(0.5);
   }
-  
+
   /// cluster global query id, under which the query will be registered
   /// on DB servers from 3.8 onwards.
-  QueryId clusterQueryId = _query.vocbase().server().getFeature<ClusterFeature>().clusterInfo().uniqid();
-
+  QueryId clusterQueryId =
+      _query.vocbase().server().getFeature<ClusterFeature>().clusterInfo().uniqid();
+
   // decreases lock timeout manually for fast path
   auto oldLockTimeout = _query.getLockTimeout();
   _query.setLockTimeout(FAST_PATH_LOCK_TIMEOUT);
@@ -449,7 +454,8 @@ Result EngineInfoContainerDBServerServerBased::buildEngines(
   for (ServerID const& server : dbServers) {
     // Build Lookup Infos
     VPackBuilder infoBuilder;
-    auto didCreateEngine = buildEngineInfo(clusterQueryId, infoBuilder, server, nodesById, nodeAliases);
+    auto didCreateEngine =
+        buildEngineInfo(clusterQueryId, infoBuilder, server, nodesById, nodeAliases);
     VPackSlice infoSlice = infoBuilder.slice();
 
     if (isNotSatelliteLeader(infoSlice)) {
@@ -468,7 +474,7 @@ Result EngineInfoContainerDBServerServerBased::buildEngines(
           .thenValue([](std::vector<arangodb::futures::Try<Result>>&& responses) -> Result {
             // We can directly report a non TRI_ERROR_LOCK_TIMEOUT
             // error as we need to abort after.
-            // Otherwise we need to report 
+            // Otherwise we need to report
             Result res{TRI_ERROR_NO_ERROR};
             for (auto const& tryRes : responses) {
               auto response = tryRes.get();
@@ -498,16 +504,18 @@ Result EngineInfoContainerDBServerServerBased::buildEngines(
 
     {
       // in case of fast path failure, we need to cleanup engines
-      auto requests = cleanupEngines(fastPathResult.get().errorNumber(), _query.vocbase().name(), serverToQueryId);
+      auto requests = cleanupEngines(fastPathResult.get().errorNumber(),
+                                     _query.vocbase().name(), serverToQueryId);
       // Wait for all requests to complete.
       // So we know that all Transactions are aborted.
       // We do NOT care for the actual result.
       futures::collectAll(requests).wait();
       snippetIds.clear();
     }
-  
+
     // we must generate a new query id, because the fast path setup has failed
-    clusterQueryId = _query.vocbase().server().getFeature<ClusterFeature>().clusterInfo().uniqid();
+    clusterQueryId =
+        _query.vocbase().server().getFeature<ClusterFeature>().clusterInfo().uniqid();
 
     // set back to default lock timeout for slow path fallback
     _query.setLockTimeout(oldLockTimeout);
@@ -521,7 +529,8 @@ Result EngineInfoContainerDBServerServerBased::buildEngines(
     std::sort(engineInformation.begin(), engineInformation.end(),
               [](auto const& lhs, auto const& rhs) {
                 // Entry <0> is the Server
-                return TransactionState::ServerIdLessThan(std::get<0>(lhs), std::get<0>(rhs));
+                return TransactionState::ServerIdLessThan(std::get<0>(lhs),
+                                                          std::get<0>(rhs));
               });
 #ifdef ARANGODB_ENABLE_MAINTAINER_MODE
     // Make sure we always maintain the correct ordering of servers
@@ -547,7 +556,9 @@ Result EngineInfoContainerDBServerServerBased::buildEngines(
       overwrittenOptions.add("clusterQueryId", VPackValue(clusterQueryId));
       addOptionsPart(overwrittenOptions, server);
       overwrittenOptions.close();
-      auto newRequest = arangodb::velocypack::Collection::merge(infoSlice, overwrittenOptions.slice(), false);
+      auto newRequest =
+          arangodb::velocypack::Collection::merge(infoSlice,
+                                                  overwrittenOptions.slice(), false);
 
       auto request = buildSetupRequest(trx, std::move(server), newRequest.slice(),
                                        std::move(didCreateEngine), snippetIds, serverToQueryId,
@@ -680,7 +691,7 @@ std::vector<arangodb::network::FutureRes> EngineInfoContainerDBServerServerBased
   options.database = dbname;
   options.timeout = network::Timeout(10.0);  // Picked arbitrarily
   options.skipScheduler = true;              // hack to speed up future.get()
-    
+
   // Shutdown query snippets
   std::string url("/_api/aql/finish/");
   VPackBuffer<uint8_t> body;
@@ -691,8 +702,8 @@ std::vector<arangodb::network::FutureRes> EngineInfoContainerDBServerServerBased
   requests.reserve(serverQueryIds.size());
   for (auto const& [server, queryId] : serverQueryIds) {
     requests.emplace_back(network::sendRequestRetry(pool, server, fuerte::RestVerb::Delete,
-                                                          url + std::to_string(queryId),
-                                                          /*copy*/ body, options));
+                                                    url + std::to_string(queryId),
+                                                    /*copy*/ body, options));
   }
   _query.incHttpRequests(static_cast<unsigned>(serverQueryIds.size()));
 
@@ -703,8 +714,9 @@ std::vector<arangodb::network::FutureRes> EngineInfoContainerDBServerServerBased
   for (auto& gn : _graphNodes) {
     auto allEngines = gn->engines();
     for (auto const& engine : *allEngines) {
-      requests.emplace_back(network::sendRequestRetry(pool, "server:" + engine.first, fuerte::RestVerb::Delete,
-                           url + basics::StringUtils::itoa(engine.second), noBody, options));
+      requests.emplace_back(network::sendRequestRetry(
+          pool, "server:" + engine.first, fuerte::RestVerb::Delete,
+          url + basics::StringUtils::itoa(engine.second), noBody, options));
     }
     _query.incHttpRequests(static_cast<unsigned>(allEngines->size()));
     gn->clearEngines();

diff --git a/arangod/Aql/EngineInfoContainerDBServerServerBased.h b/arangod/Aql/EngineInfoContainerDBServerServerBased.h
@@ -40,7 +40,7 @@ namespace network {
 class ConnectionPool;
 struct RequestOptions;
 struct Response;
-}
+}  // namespace network
 
 namespace velocypack {
 class Builder;
@@ -56,6 +56,9 @@ class QuerySnippet;
 
 class EngineInfoContainerDBServerServerBased {
  private:
+  // TODO Temporary. Do not check in!!
+  // We need to access the TraverserEngineShardList nothing else.
+ public:
   // @brief Local struct to create the
   // information required to build traverser engines
   // on DB servers.
@@ -123,7 +126,6 @@ class EngineInfoContainerDBServerServerBased {
   // the given queryid of the coordinator as data provider.
   void closeSnippet(QueryId inputSnippet);
 
-
   // Build the Engines for the DBServer
   //   * Creates one Query-Entry for each Snippet per Shard (multiple on the
   //   same DB)
@@ -140,28 +142,28 @@ class EngineInfoContainerDBServerServerBased {
                       MapRemoteToSnippet& snippetIds, aql::ServerQueryIdList& serverQueryIds,
                       std::map<ExecutionNodeId, ExecutionNodeId>& nodeAliases);
 
-
   // Insert a GraphNode that needs to generate TraverserEngines on
   // the DBServers. The GraphNode itself will retain on the coordinator.
   void addGraphNode(GraphNode* node, bool pushToSingleServer);
 
  private:
   /**
-  * @brief Helper method to generate the Request to be send to a specific database server.
-  * this request contains all the necessary information to create a transaction with correct shard
-  * locking, as well as QuerySnippets and GraphEngines on the receiver side.
-  *
-  * @param clusterQueryId cluster-wide query id (used from 3.8 onwards)
-  * @param infoBuilder (mutable) the request body will be written into this builder. 
-  * @param server The DatabaseServer we suppose to send the request to, used to identify the shards on this server
-  * @param nodesById A vector to get Nodes by their id.
-  * @param nodeAliases (mutable) A map of node-aliases, if a server is responsible for more then one shard we need to duplicate some nodes in the query (e.g. an IndexNode can only access one shard at a time) this list can map cloned node -> original node ids.
-  *
-  * @return A vector with one entry per GraphNode in the query (in order) it indicates if this Server has created a GraphEngine for this Node and needs to participate in the GraphOperation or not.
-  */
-  std::vector<bool> buildEngineInfo(QueryId clusterQueryId, VPackBuilder& infoBuilder, ServerID const& server,
-                                    std::unordered_map<ExecutionNodeId, ExecutionNode*> const& nodesById,
-                                    std::map<ExecutionNodeId, ExecutionNodeId>& nodeAliases);
+   * @brief Helper method to generate the Request to be send to a specific database server.
+   * this request contains all the necessary information to create a transaction with correct shard
+   * locking, as well as QuerySnippets and GraphEngines on the receiver side.
+   *
+   * @param clusterQueryId cluster-wide query id (used from 3.8 onwards)
+   * @param infoBuilder (mutable) the request body will be written into this builder.
+   * @param server The DatabaseServer we suppose to send the request to, used to identify the shards on this server
+   * @param nodesById A vector to get Nodes by their id.
+   * @param nodeAliases (mutable) A map of node-aliases, if a server is responsible for more then one shard we need to duplicate some nodes in the query (e.g. an IndexNode can only access one shard at a time) this list can map cloned node -> original node ids.
+   *
+   * @return A vector with one entry per GraphNode in the query (in order) it indicates if this Server has created a GraphEngine for this Node and needs to participate in the GraphOperation or not.
+   */
+  std::vector<bool> buildEngineInfo(
+      QueryId clusterQueryId, VPackBuilder& infoBuilder, ServerID const& server,
+      std::unordered_map<ExecutionNodeId, ExecutionNode*> const& nodesById,
+      std::map<ExecutionNodeId, ExecutionNodeId>& nodeAliases);
 
   arangodb::futures::Future<Result> buildSetupRequest(
       transaction::Methods& trx, ServerID const& server, VPackSlice infoSlice,
@@ -171,7 +173,6 @@ class EngineInfoContainerDBServerServerBased {
 
   [[nodiscard]] bool isNotSatelliteLeader(VPackSlice infoSlice) const;
 
-
   /**
    * @brief Will send a shutdown to all engines registered in the list of
    * queryIds.
@@ -186,8 +187,8 @@ class EngineInfoContainerDBServerServerBased {
    * -> queryid.
    */
   std::vector<futures::Future<network::Response>> cleanupEngines(
-      ErrorCode errorCode, std::string const& dbname, aql::ServerQueryIdList& serverQueryIds) const;
-
+      ErrorCode errorCode, std::string const& dbname,
+      aql::ServerQueryIdList& serverQueryIds) const;
 
   // Insert the Locking information into the message to be send to DBServers
   void addLockingPart(arangodb::velocypack::Builder& builder, ServerID const& server) const;