Make collection to shard map a map of string -> vector<string>, requi…

…red for SmartGraphs
arangodb · mchacki · Jul 6, 2021 · Apr 20, 2021 · Apr 22, 2021 · Apr 22, 2021
commit c01bec86608867e7aa011091d39fc08b8894e52c
diff --git a/arangod/Aql/GraphNode.cpp b/arangod/Aql/GraphNode.cpp
@@ -90,7 +90,7 @@ TRI_edge_direction_e parseDirection(AstNode const* node) {
   return uint64ToDirection(dirNode->getIntValue());
 }
 
-}
+}  // namespace
 
 GraphNode::GraphNode(ExecutionPlan* plan, ExecutionNodeId id,
                      TRI_vocbase_t* vocbase, AstNode const* direction,
@@ -357,7 +357,8 @@ GraphNode::GraphNode(ExecutionPlan* plan, arangodb::velocypack::Slice const& bas
   auto getAqlCollectionFromName = [&](std::string const& name) -> aql::Collection& {
     // if the collection was already existent in the query, addCollection will
     // just return it.
-    return *query.collections().add(name, AccessMode::Type::READ, aql::Collection::Hint::Collection);
+    return *query.collections().add(name, AccessMode::Type::READ,
+                                    aql::Collection::Hint::Collection);
   };
 
   auto vPackDirListIter = VPackArrayIterator(dirList);
@@ -397,7 +398,8 @@ GraphNode::GraphNode(ExecutionPlan* plan, arangodb::velocypack::Slice const& bas
        "graph needs a translation from collection to shard names");
   }
   for (auto const& item : VPackObjectIterator(collectionToShard)) {
-    _collectionToShard.insert({item.key.copyString(), item.value.copyString()});
+    _collectionToShard.insert({item.key.copyString(),
+                               std::vector<std::string>{item.value.copyString()}});
   }
 
   // Out variables
@@ -510,9 +512,11 @@ std::string const& GraphNode::collectionToShardName(std::string const& collName)
 
   auto found = _collectionToShard.find(collName);
   if (found == _collectionToShard.end()) {
-    THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_INTERNAL_AQL, "unable to find shard '" + collName + "' in query shard map");
+    THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_INTERNAL_AQL,
+                                   "unable to find shard '" + collName +
+                                       "' in query shard map");
   }
-  return found->second;
+  return found->second.front();
 }
 
 void GraphNode::toVelocyPackHelper(VPackBuilder& nodes, unsigned flags,
@@ -581,7 +585,8 @@ void GraphNode::toVelocyPackHelper(VPackBuilder& nodes, unsigned flags,
   {
     VPackObjectBuilder guard(&nodes);
     for (auto const& item : _collectionToShard) {
-      nodes.add(item.first, VPackValue(item.second));
+      TRI_ASSERT(item.second.size() == 1);
+      nodes.add(item.first, VPackValue(item.second.front()));
     }
   }
 
@@ -714,8 +719,8 @@ void GraphNode::enhanceEngineInfo(VPackBuilder& builder) const {
 }
 #endif
 
-void GraphNode::addEdgeCollection(Collections const& collections, std::string const& name,
-                                  TRI_edge_direction_e dir) {
+void GraphNode::addEdgeCollection(Collections const& collections,
+                                  std::string const& name, TRI_edge_direction_e dir) {
   auto aqlCollection = collections.get(name);
   if (aqlCollection != nullptr) {
     addEdgeCollection(*aqlCollection, dir);

diff --git a/arangod/Aql/GraphNode.h b/arangod/Aql/GraphNode.h
@@ -23,12 +23,12 @@
 
 #pragma once
 
-#include "Aql/types.h"
 #include "Aql/Condition.h"
 #include "Aql/ExecutionNode.h"
 #include "Aql/ExecutionNodeId.h"
 #include "Aql/GraphNode.h"
 #include "Aql/Graphs.h"
+#include "Aql/types.h"
 #include "Cluster/ClusterTypes.h"
 #include "VocBase/LogicalCollection.h"
 #include "VocBase/voc-types.h"
@@ -192,20 +192,21 @@ class GraphNode : public ExecutionNode {
   void injectVertexCollection(aql::Collection& other);
 
   std::vector<aql::Collection const*> collections() const;
-  void setCollectionToShard(std::map<std::string, std::string> const& map) {
+  void setCollectionToShard(std::unordered_map<std::string, std::vector<std::string>> const& map) {
     _collectionToShard = map;
   }
   void addCollectionToShard(std::string const& coll, std::string const& shard) {
     // NOTE: Do not replace this by emplace or insert.
     // This is also used to overwrite the existing entry.
-    _collectionToShard[coll] = shard;
+    _collectionToShard[coll] = std::vector<std::string>{shard};
   }
 
  public:
   graph::Graph const* graph() const noexcept;
 
  private:
-  void addEdgeCollection(aql::Collections const& collections, std::string const& name, TRI_edge_direction_e dir);
+  void addEdgeCollection(aql::Collections const& collections,
+                         std::string const& name, TRI_edge_direction_e dir);
   void addEdgeCollection(aql::Collection& collection, TRI_edge_direction_e dir);
   void addVertexCollection(aql::Collections const& collections, std::string const& name);
   void addVertexCollection(aql::Collection& collection);
@@ -268,7 +269,7 @@ class GraphNode : public ExecutionNode {
   std::unordered_map<ServerID, aql::EngineId> _engines;
 
   /// @brief list of shards involved, required for one-shard-databases
-  std::map<std::string, std::string> _collectionToShard;
+  std::unordered_map<std::string, std::vector<std::string>> _collectionToShard;
 };
 
 }  // namespace aql

diff --git a/arangod/Graph/BaseOptions.cpp b/arangod/Graph/BaseOptions.cpp
@@ -365,7 +365,8 @@ void BaseOptions::serializeVariables(VPackBuilder& builder) const {
   _expressionCtx.serializeAllVariables(_query.vpackOptions(), builder);
 }
 
-void BaseOptions::setCollectionToShard(std::map<std::string, std::string> const& in) {
+void BaseOptions::setCollectionToShard(
+    std::unordered_map<std::string, std::vector<std::string>> const& in) {
   _collectionToShard = std::move(in);
 }
 

diff --git a/arangod/Graph/BaseOptions.h b/arangod/Graph/BaseOptions.h
@@ -117,7 +117,7 @@ struct BaseOptions {
 
   void serializeVariables(arangodb::velocypack::Builder&) const;
 
-  void setCollectionToShard(std::map<std::string, std::string> const&);
+  void setCollectionToShard(std::unordered_map<std::string, std::vector<std::string>> const&);
 
   bool produceVertices() const { return _produceVertices; }
 
@@ -157,7 +157,7 @@ struct BaseOptions {
   void activateCache(bool enableDocumentCache,
                      std::unordered_map<ServerID, aql::EngineId> const* engines);
 
-  std::map<std::string, std::string> const& collectionToShard() const {
+  std::unordered_map<std::string, std::vector<std::string>> const& collectionToShard() const {
     return _collectionToShard;
   }
 
@@ -215,7 +215,7 @@ struct BaseOptions {
   std::unique_ptr<TraverserCache> _cache;
 
   // @brief - translations for one-shard-databases
-  std::map<std::string, std::string> _collectionToShard;
+  std::unordered_map<std::string, std::vector<std::string>> _collectionToShard;
 
   /// @brief a value of 1 (which is the default) means "no parallelism"
   size_t _parallelism;

diff --git a/arangod/Graph/Cache/RefactoredTraverserCache.cpp b/arangod/Graph/Cache/RefactoredTraverserCache.cpp
@@ -53,8 +53,7 @@ using namespace arangodb::graph;
 
 namespace {
 bool isWithClauseMissing(arangodb::basics::Exception const& ex) {
-  if (ServerState::instance()->isDBServer() &&
-      ex.code() == TRI_ERROR_ARANGO_DATA_SOURCE_NOT_FOUND) {
+  if (ServerState::instance()->isDBServer() && ex.code() == TRI_ERROR_ARANGO_DATA_SOURCE_NOT_FOUND) {
     // on a DB server, we could have got here only in the OneShard case.
     // in this case turn the rather misleading "collection or view not found"
     // error into a nicer "collection not known to traversal, please add WITH"
@@ -68,19 +67,20 @@ bool isWithClauseMissing(arangodb::basics::Exception const& ex) {
 
   return false;
 }
-} // namespace
+}  // namespace
 
 RefactoredTraverserCache::RefactoredTraverserCache(
     arangodb::transaction::Methods* trx, aql::QueryContext* query,
     arangodb::ResourceMonitor& resourceMonitor, arangodb::aql::TraversalStats& stats,
-    std::map<std::string, std::string> const& collectionToShardMap)
+    std::unordered_map<std::string, std::vector<std::string>> const& collectionToShardMap)
     : _query(query),
       _trx(trx),
       _stringHeap(resourceMonitor, 4096), /* arbitrary block-size may be adjusted for performance */
       _collectionToShardMap(collectionToShardMap),
       _resourceMonitor(resourceMonitor),
-      _allowImplicitCollections(ServerState::instance()->isSingleServer() &&
-                                !_query->vocbase().server().getFeature<QueryRegistryFeature>().requireWith()) {
+      _allowImplicitCollections(
+          ServerState::instance()->isSingleServer() &&
+          !_query->vocbase().server().getFeature<QueryRegistryFeature>().requireWith()) {
   TRI_ASSERT(!ServerState::instance()->isCoordinator());
 }
 
@@ -106,17 +106,20 @@ bool RefactoredTraverserCache::appendEdge(EdgeDocumentToken const& idToken,
     return false;
   }
 
-  auto res = col->getPhysical()->read(
-      _trx, idToken.localDocumentId(), [&](LocalDocumentId const&, VPackSlice edge) -> bool {
-        // NOTE: Do not count this as Primary Index Scan, we counted it in the
-        // edge Index before copying...
-        if constexpr (std::is_same_v<ResultType, aql::AqlValue>) {
-          result = aql::AqlValue(edge);
-        } else if constexpr (std::is_same_v<ResultType, velocypack::Builder>) {
-          result.add(edge);
-        }
-        return true;
-      }).ok();
+  auto res =
+      col->getPhysical()
+          ->read(_trx, idToken.localDocumentId(),
+                 [&](LocalDocumentId const&, VPackSlice edge) -> bool {
+                   // NOTE: Do not count this as Primary Index Scan, we counted
+                   // it in the edge Index before copying...
+                   if constexpr (std::is_same_v<ResultType, aql::AqlValue>) {
+                     result = aql::AqlValue(edge);
+                   } else if constexpr (std::is_same_v<ResultType, velocypack::Builder>) {
+                     result.add(edge);
+                   }
+                   return true;
+                 })
+          .ok();
   if (ADB_UNLIKELY(!res)) {
     // We already had this token, inconsistent state. Return NULL in Production
     LOG_TOPIC("daac5", ERR, arangodb::Logger::GRAPHS)
@@ -139,20 +142,7 @@ ResultT<std::pair<std::string, size_t>> RefactoredTraverserCache::extractCollect
   }
 
   std::string colName = idHashed.substr(0, pos).toString();
-  if (_collectionToShardMap.empty()) {
-    TRI_ASSERT(!ServerState::instance()->isDBServer());
-    return std::make_pair(colName, pos);
-  }
-  auto it = _collectionToShardMap.find(colName);
-  if (it == _collectionToShardMap.end()) {
-    // Connected to a vertex where we do not know the Shard to.
-    return Result{TRI_ERROR_QUERY_COLLECTION_LOCK_FAILED,
-                  "collection not known to traversal: '" + colName +
-                      "'. please add 'WITH " + colName +
-                      "' as the first line in your AQL"};
-  }
-  // We have translated to a Shard
-  return std::make_pair(it->second, pos);
+  return std::make_pair(colName, pos);
 }
 
 template <typename ResultType>
@@ -163,43 +153,68 @@ bool RefactoredTraverserCache::appendVertex(aql::TraversalStats& stats,
   if (collectionNameResult.fail()) {
     THROW_ARANGO_EXCEPTION(collectionNameResult.result());
   }
-
-  std::string const& collectionName = collectionNameResult.get().first;
 
-  try {
-    transaction::AllowImplicitCollectionsSwitcher disallower(_trx->state()->options(), _allowImplicitCollections);
-
-    Result res = _trx->documentFastPathLocal(
-        collectionName,
-        id.substr(collectionNameResult.get().second + 1).stringRef(),
-        [&](LocalDocumentId const&, VPackSlice doc) -> bool {
-          stats.addScannedIndex(1);
-          // copying...
-          if constexpr (std::is_same_v<ResultType, aql::AqlValue>) {
-            result = aql::AqlValue(doc);
-          } else if constexpr (std::is_same_v<ResultType, velocypack::Builder>) {
-            result.add(doc);
-          }
-          return true;
-        });
-    if (res.ok()) {
-      return true;
+  auto findDocumentInShard = [&](std::string const& collectionName) -> bool {
+    try {
+      transaction::AllowImplicitCollectionsSwitcher disallower(_trx->state()->options(),
+                                                               _allowImplicitCollections);
+
+      Result res = _trx->documentFastPathLocal(
+          collectionName, id.substr(collectionNameResult.get().second + 1).stringRef(),
+          [&](LocalDocumentId const&, VPackSlice doc) -> bool {
+            stats.addScannedIndex(1);
+            // copying...
+            if constexpr (std::is_same_v<ResultType, aql::AqlValue>) {
+              result = aql::AqlValue(doc);
+            } else if constexpr (std::is_same_v<ResultType, velocypack::Builder>) {
+              result.add(doc);
+            }
+            return true;
+          });
+      if (res.ok()) {
+        return true;
+      }
+
+      if (!res.is(TRI_ERROR_ARANGO_DOCUMENT_NOT_FOUND)) {
+        // ok we are in a rather bad state. Better throw and abort.
+        THROW_ARANGO_EXCEPTION(res);
+      }
+    } catch (basics::Exception const& ex) {
+      if (isWithClauseMissing(ex)) {
+        // turn
3D16
 the error into a different error
+        THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_QUERY_COLLECTION_LOCK_FAILED,
+                                       "collection not known to traversal: '" + collectionName +
+                                           "'. please add 'WITH " + collectionName +
+                                           "' as the first line in your AQL");
+      }
+      // rethrow original error
+      throw;
     }
+    return false;
+  };
 
-    if (!res.is(TRI_ERROR_ARANGO_DOCUMENT_NOT_FOUND)) {
-      // ok we are in a rather bad state. Better throw and abort.
-      THROW_ARANGO_EXCEPTION(res);
+  std::string const& collectionName = collectionNameResult.get().first;
+  if (_collectionToShardMap.empty()) {
+    TRI_ASSERT(!ServerState::instance()->isDBServer());
+    if (findDocumentInShard(collectionName)) {
+      return true;
+    }
+  } else {
+    auto it = _collectionToShardMap.find(collectionName);
+    if (it == _collectionToShardMap.end()) {
+      // Connected to a vertex where we do not know the Shard to.
+      throw Result{TRI_ERROR_QUERY_COLLECTION_LOCK_FAILED,
+                   "collection not known to traversal: '" + collectionName +
+                       "'. please add 'WITH " + collectionNam
CEB2
e +
+                       "' as the first line in your AQL"};
     }
-  } catch (basics::Exception const& ex) {
-    if (isWithClauseMissing(ex)) {
-      // turn the error into a different error 
-      THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_QUERY_COLLECTION_LOCK_FAILED,
-                                     "collection not known to traversal: '" +
-                                     collectionName + "'. please add 'WITH " + collectionName +
-                                     "' as the first line in your AQL");
+    for (auto const& shard : it->second) {
+      if (findDocumentInShard(shard)) {
+        // Short circuit, as soon as one shard contains this document
+        // we can return it.
+        return true;
+      }
     }
-    // rethrow original error
-    throw;
   }
 
   // Register a warning. It is okay though but helps the user

diff --git a/arangod/Graph/Cache/RefactoredTraverserCache.h b/arangod/Graph/Cache/RefactoredTraverserCache.h
@@ -62,11 +62,10 @@ struct EdgeDocumentToken;
 
 class RefactoredTraverserCache {
  public:
-  explicit RefactoredTraverserCache(arangodb::transaction::Methods* trx,
-                                    aql::QueryContext* query,
-                                    arangodb::ResourceMonitor& resourceMonitor,
-                                    arangodb::aql::TraversalStats& stats,
-                                    std::map<std::string, std::string> const& collectionToShardMap);
+  explicit RefactoredTraverserCache(
+      arangodb::transaction::Methods* trx, aql::QueryContext* query,
+      arangodb::ResourceMonitor& resourceMonitor, arangodb::aql::TraversalStats& stats,
+      std::unordered_map<std::string, std::vector<std::string>> const& collectionToShardMap);
   ~RefactoredTraverserCache();
 
   RefactoredTraverserCache(RefactoredTraverserCache const&) = delete;
@@ -81,8 +80,7 @@ class RefactoredTraverserCache {
   /// @brief Inserts the real document stored within the token
   ///        into the given builder.
   //////////////////////////////////////////////////////////////////////////////
-  void insertEdgeIntoResult(graph::EdgeDocumentToken const& etkn,
-                            velocypack::Builder& builder);
+  void insertEdgeIntoResult(graph::EdgeDocumentToken const& etkn, velocypack::Builder& builder);
 
   //////////////////////////////////////////////////////////////////////////////
   /// @brief Inserts the real document identified by the _id string
@@ -148,9 +146,9 @@ class RefactoredTraverserCache {
   std::unordered_set<arangodb::velocypack::HashedStringRef> _persistedStrings;
 
  private:
-  std::map<std::string, std::string> const& _collectionToShardMap;
+  std::unordered_map<std::string, std::vector<std::string>> const& _collectionToShardMap;
   arangodb::ResourceMonitor& _resourceMonitor;
-  
+
   /// @brief whether or not to allow adding of previously unknown collections
   /// during the traversal
   bool const _allowImplicitCollections;

diff --git a/arangod/Graph/Enumerators/OneSidedEnumerator.cpp b/arangod/Graph/Enumerators/OneSidedEnumerator.cpp
@@ -96,7 +96,13 @@ auto OneSidedEnumerator<QueueType, PathStoreType, ProviderType, PathValidator>::
   auto posPrevious = _interior.append(step);
 
   ValidationResult res = _validator.validatePath(step);
-  if ((step.getDepth() >= _options.getMinDepth()) && !res.isFiltered()) {
+
+  LOG_TOPIC("78156", TRACE, Logger::GRAPHS)
+      << std::boolalpha << "<Traverser> Validated Vertex: " << step.getVertex().getID()
+      << " filtered " << res.isFiltered() << " pruned " << res.isPruned()
+      << " depth " << _options.getMinDepth() << " <= " << step.getDepth()
+      << "<= " << _options.getMaxDepth();
+  if (step.getDepth() >= _options.getMinDepth() && !res.isFiltered()) {
     // Include it in results.
     _results.push_back(step);
   }