Revert "Fix locking of shards in TraverserEngines."

This reverts commit 467088b8af.

arangod/Aql/ExecutionEngine.cpp

@@ -1007,14 +1007,7 @@ struct CoordinatorInstanciator : public WalkerWorker<ExecutionNode> {
       if (!shardSet.empty()) {
         arangodb::CoordTransactionID coordTransactionID = TRI_NewTickServer();
         std::unordered_map<std::string, std::string> headers;
-        std::string shardList;
-        for (auto const& shard : shardSet) {
-          if (!shardList.empty()) {
-            shardList += ",";
-          }
-          shardList += shard;
-        }
-        headers["X-Arango-Nolock"] = shardList;  // Prevent locking
+
         auto res = cc->syncRequest("", coordTransactionID, "server:" + list.first,
                                    RequestType::POST, url, engineInfo.toJson(),
                                    headers, 30.0);
@@ -1025,7 +1018,7 @@ struct CoordinatorInstanciator : public WalkerWorker<ExecutionNode> {
             message += std::string(" : ") + res->errorMessage;
           }
           THROW_ARANGO_EXCEPTION_MESSAGE(
-              TRI_ERROR_CLUSTER_BACKEND_UNAVAILABLE, message);
+              TRI_ERROR_QUERY_COLLECTION_LOCK_FAILED, message);
         } else {
           // Only if the result was successful we will get here
           arangodb::basics::StringBuffer& body = res->result->getBody();
@@ -1035,7 +1028,7 @@ struct CoordinatorInstanciator : public WalkerWorker<ExecutionNode> {
           VPackSlice resultSlice = builder->slice();
           if (!resultSlice.isNumber()) {
             THROW_ARANGO_EXCEPTION_MESSAGE(
-                TRI_ERROR_INTERNAL, "got unexpected response from engine build request");
+                TRI_ERROR_INTERNAL, "got unexpected response from engine lock request");
           }
           auto engineId = resultSlice.getNumericValue<traverser::TraverserEngineID>();
           TRI_ASSERT(engineId != 0);
@@ -1216,9 +1209,6 @@ ExecutionEngine* ExecutionEngine::instantiateFromPlan(
           engine->_lockedShards = new std::unordered_set<std::string>();
           engine->_previouslyLockedShards = nullptr;
         }
-        // Note that it is crucial that this is a map and not an unordered_map,
-        // because we need to guarantee the order of locking by using
-        // alphabetical order on the shard names!
         std::map<std::string, std::pair<std::string, bool>> forLocking;
         for (auto& q : inst.get()->queryIds) {
           std::string theId = q.first;

arangod/Cluster/TraverserEngine.cpp

@@ -94,12 +94,9 @@ BaseTraverserEngine::BaseTraverserEngine(TRI_vocbase_t* vocbase,
   auto params = std::make_shared<VPackBuilder>();
   auto opts = std::make_shared<VPackBuilder>();
 
-  _trx = new arangodb::AqlTransaction(
-      arangodb::StandaloneTransactionContext::Create(vocbase),
-      _collections.collections(), true);
-  // true here as last argument is crucial: it leads to the fact that the
-  // created transaction is considered a "MAIN" part and will not switch
-  // off collection locking completely!
+  _trx = new aql::AqlTransaction(
+      arangodb::transaction::StandaloneContext::Create(vocbase),
+      _collections.collections(), false);
   _query = new aql::Query(true, vocbase, "", 0, params, opts, aql::PART_DEPENDENT);
   _query->injectTransaction(_trx);
 
@@ -115,6 +112,7 @@ BaseTraverserEngine::BaseTraverserEngine(TRI_vocbase_t* vocbase,
     }
   }
 
+
   _trx->begin(); // We begin the transaction before we lock.
                  // We also setup indexes before we lock.
 }

arangod/RestHandler/RestVocbaseBaseHandler.cpp

@@ -627,16 +627,8 @@ void RestVocbaseBaseHandler::prepareExecute() {
   std::string const& shardId = _request->header("x-arango-nolock", found);
 
   if (found) {
-    _nolockHeaderSet = new std::unordered_set<std::string>();
-    // Split value at commas, if there are any, otherwise take full value:
-    size_t pos = shardId.find(',');
-    size_t oldpos = 0;
-    while (pos != std::string::npos) {
-      _nolockHeaderSet->emplace(shardId.substr(oldpos, pos - oldpos));
-      oldpos = pos + 1;
-      pos = shardId.find(',', oldpos);
-    }
-    _nolockHeaderSet->emplace(shardId.substr(oldpos));
+    _nolockHeaderSet =
+        new std::unordered_set<std::string>{std::string(shardId)};
     CollectionLockState::_noLockHeaders = _nolockHeaderSet;
   }
 }