Fixed error in recovery for duplicate collection name

arangod/Cluster/ClusterInfo.cpp

@@ -466,7 +466,7 @@ void ClusterInfo::loadPlan() {
               std::shared_ptr<LogicalCollection> newCollection;
 #ifndef USE_ENTERPRISE
               newCollection = std::make_shared<LogicalCollection>(
-                  vocbase, collectionSlice, false);
+                  vocbase, collectionSlice);
 #else
               VPackSlice isSmart = collectionSlice.get("isSmart");
               if (isSmart.isTrue()) {
@@ -480,7 +480,7 @@ void ClusterInfo::loadPlan() {
                 }
               } else {
                 newCollection = std::make_shared<LogicalCollection>(
-                    vocbase, collectionSlice, false);
+                    vocbase, collectionSlice);
               }
 #endif
               std::string const collectionName = newCollection->name();

arangod/MMFiles/MMFilesEngine.cpp

@@ -1282,7 +1282,7 @@ TRI_vocbase_t* MMFilesEngine::openExistingDatabase(TRI_voc_tick_t id, std::strin
     for (auto const& it : VPackArrayIterator(slice)) {
       // we found a collection that is still active
       TRI_ASSERT(!it.get("id").isNone() || !it.get("cid").isNone());
-      auto uniqCol = std::make_unique<arangodb::LogicalCollection>(vocbase.get(), it, true);
+      auto uniqCol = std::make_unique<arangodb::LogicalCollection>(vocbase.get(), it);
       auto collection = uniqCol.get();
       TRI_ASSERT(collection != nullptr);
       StorageEngine::registerCollection(vocbase.get(), uniqCol.get());

arangod/VocBase/LogicalCollection.cpp

@@ -168,6 +168,7 @@ LogicalCollection::LogicalCollection(LogicalCollection const& other)
       _cleanupIndexes(0),
       _persistentIndexes(0),
       _physical(other.getPhysical()->clone(this, other.getPhysical())) {
+  TRI_ASSERT(_physical != nullptr);
   if (ServerState::instance()->isDBServer() ||
       !ServerState::instance()->isRunningInCluster()) {
     _followers.reset(new FollowerInfo(this));
@@ -185,7 +186,7 @@ LogicalCollection::LogicalCollection(LogicalCollection const& other)
 // The Slice contains the part of the plan that
 // is relevant for this collection.
 LogicalCollection::LogicalCollection(TRI_vocbase_t* vocbase,
-                                     VPackSlice const& info, bool isPhysical)
+                                     VPackSlice const& info)
     : _internalVersion(0),
       _cid(ReadCid(info)),
       _planId(ReadPlanId(info, _cid)),
@@ -215,7 +216,7 @@ LogicalCollection::LogicalCollection(TRI_vocbase_t* vocbase,
       _persistentIndexes(0),
       _physical(
           EngineSelectorFeature::ENGINE->createPhysicalCollection(this, info)) {
-  getPhysical()->setPath(ReadStringValue(info, "path", ""));
+  TRI_ASSERT(_physical != nullptr);
   if (!IsAllowedName(info)) {
     THROW_ARANGO_EXCEPTION(TRI_ERROR_ARANGO_ILLEGAL_NAME);
   }
@@ -398,21 +399,6 @@ LogicalCollection::LogicalCollection(TRI_vocbase_t* vocbase,
   }
 #endif
 
-  if (!ServerState::instance()->isCoordinator() && isPhysical) {
-    // If we are not in the coordinator we need a path
-    // to the physical data.
-    StorageEngine* engine = EngineSelectorFeature::ENGINE;
-    if (getPhysical()->path().empty()) {
-      std::string path = engine->createCollection(_vocbase, _cid, this);
-      getPhysical()->setPath(path);
-    }
-  }
-
-  int64_t count = Helper::readNumericValue<int64_t>(info, "count", -1);
-  if (count != -1) {
-    _physical->updateCount(count);
-  }
-
   if (ServerState::instance()->isDBServer() ||
       !ServerState::instance()->isRunningInCluster()) {
     _followers.reset(new FollowerInfo(this));
@@ -420,7 +406,6 @@ LogicalCollection::LogicalCollection(TRI_vocbase_t* vocbase,
 
   // update server's tick value
   TRI_UpdateTickServer(static_cast<TRI_voc_tick_t>(_cid));
-
 }
 
 LogicalCollection::~LogicalCollection() {}
@@ -1128,6 +1113,21 @@ bool LogicalCollection::dropIndex(TRI_idx_iid_t iid) {
   return _physical->dropIndex(iid);
 }
 
+
+/// @brief Persist the connected physical collection.
+///        This should be called AFTER the collection is successfully
+///        created and only on Sinlge/DBServer
+void LogicalCollection::persistPhysicalCollection() {
+  // Coordinators are not allowed to have local collections!
+  TRI_ASSERT(!ServerState::instance()->isCoordinator());
+
+  // We have not yet persisted this collection!
+  TRI_ASSERT(getPhysical()->path().empty());
+  StorageEngine* engine = EngineSelectorFeature::ENGINE;
+  std::string path = engine->createCollection(_vocbase, _cid, this);
+  getPhysical()->setPath(path);
+}
+
 /// @brief creates the initial indexes for the collection
 void LogicalCollection::createInitialIndexes() {
   if (!_indexes.empty()) {

arangod/VocBase/LogicalCollection.h

@@ -92,8 +92,7 @@ class LogicalCollection {
   friend struct ::TRI_vocbase_t;
 
  public:
-  LogicalCollection(TRI_vocbase_t*, velocypack::Slice const&,
+  LogicalCollection(TRI_vocbase_t*, velocypack::Slice const&);
-                    bool isPhysical);
 
   virtual ~LogicalCollection();
 
@@ -318,6 +317,11 @@ class LogicalCollection {
                                TRI_voc_tick_t maxTick,
                                ManagedDocumentResult& result);
 
+  /// @brief Persist the connected physical collection.
+  ///        This should be called AFTER the collection is successfully
+  ///        created and only on Sinlge/DBServer
+  void persistPhysicalCollection();
+
  private:
   // SECTION: Index creation
 

arangod/VocBase/vocbase.cpp

@@ -339,7 +339,7 @@ arangodb::LogicalCollection* TRI_vocbase_t::createCollectionWorker(
   // Try to create a new collection. This is not registered yet
 
   std::unique_ptr<arangodb::LogicalCollection> collection =
-      std::make_unique<arangodb::LogicalCollection>(this, parameters, true);
+      std::make_unique<arangodb::LogicalCollection>(this, parameters);
   TRI_ASSERT(collection != nullptr);
 
   WRITE_LOCKER(writeLocker, _collectionsLock);
@@ -364,6 +364,10 @@ arangodb::LogicalCollection* TRI_vocbase_t::createCollectionWorker(
     collection->setStatus(TRI_VOC_COL_STATUS_LOADED);
     // set collection version to 3.1, as the collection is just created
     collection->setVersion(LogicalCollection::VERSION_31);
+
+    // Let's try to persist it.
+    collection->persistPhysicalCollection();
+ 
     events::CreateCollection(name, TRI_ERROR_NO_ERROR);
     return collection.release();
   } catch (...) {