Fixed babies creation in cluster with undefined keys.

arangod/Cluster/ClusterInfo.cpp

@@ -2418,7 +2418,8 @@ std::shared_ptr<std::vector<ShardID>> ClusterInfo::getShardList(
 int ClusterInfo::getResponsibleShard(CollectionID const& collectionID,
                                      VPackSlice slice, bool docComplete,
                                      ShardID& shardID,
-                                     bool& usesDefaultShardingAttributes) {
+                                     bool& usesDefaultShardingAttributes,
+                                     std::string const& key) {
   // Note that currently we take the number of shards and the shardKeys
   // from Plan, since they are immutable. Later we will have to switch
   // this to Current, when we allow to add and remove shards.
@@ -2464,7 +2465,7 @@ int ClusterInfo::getResponsibleShard(CollectionID const& collectionID,
 
   int error = TRI_ERROR_NO_ERROR;
   uint64_t hash = arangodb::basics::VelocyPackHelper::hashByAttributes(
-      slice, *shardKeysPtr, docComplete, error);
+      slice, *shardKeysPtr, docComplete, error, key);
   static char const* magicPhrase =
       "Foxx you have stolen the goose, give she back again!";
   static size_t const len = 52;

arangod/Cluster/ClusterInfo.h

@@ -800,7 +800,8 @@ class ClusterInfo {
 
   int getResponsibleShard(CollectionID const&, arangodb::velocypack::Slice,
                           bool docComplete, ShardID& shardID,
-                          bool& usesDefaultShardingAttributes);
+                          bool& usesDefaultShardingAttributes,
+                          std::string const& key = "");
 
 
   //////////////////////////////////////////////////////////////////////////////

arangod/Cluster/ClusterMethods.cpp

@@ -568,7 +568,7 @@ int createDocumentOnCoordinator(
    }
  
    std::string const collid = StringUtils::itoa(collinfo->id());
-   std::unordered_map<ShardID, std::vector<VPackValueLength>> shardMap;
+   std::unordered_map<ShardID, std::vector<std::pair<VPackValueLength, std::string>>> shardMap;
    std::vector<std::pair<ShardID, VPackValueLength>> reverseMapping;
    bool useMultiple = slice.isArray();
 
@@ -597,8 +597,15 @@ int createDocumentOnCoordinator(
      // Now find the responsible shard:
      bool usesDefaultShardingAttributes;
      ShardID shardID;
-     int error = ci->getResponsibleShard(collid, node, true, shardID,
+     int error TRI_ERROR_NO_ERROR;
+     if (userSpecifiedKey) {
+       error = ci->getResponsibleShard(collid, node, true, shardID,
                                            usesDefaultShardingAttributes);
+     } else {
+       error = ci->getResponsibleShard(collid, node, true, shardID,
+                                           usesDefaultShardingAttributes,
+                                           _key);
+     }
      if (error == TRI_ERROR_ARANGO_COLLECTION_NOT_FOUND) {
        return TRI_ERROR_CLUSTER_SHARD_GONE;
      }
@@ -610,10 +617,19 @@ int createDocumentOnCoordinator(
      // We found the responsible shard. Add it to the list.
      auto it = shardMap.find(shardID);
      if (it == shardMap.end()) {
-       std::vector<VPackValueLength> counter({index});
+       if (userSpecifiedKey) {
+         std::vector<std::pair<VPackValueLength, std::string>> counter({{index, ""}});
          shardMap.emplace(shardID, counter);
        } else {
-       it->second.emplace_back(index);
+         std::vector<std::pair<VPackValueLength, std::string>> counter({{index, _key}});
+         shardMap.emplace(shardID, counter);
+       }
+     } else {
+       if (userSpecifiedKey) {
+         it->second.emplace_back(index, "");
+       } else {
+         it->second.emplace_back(index, _key);
+       }
      }
      reverseMapping.emplace_back(shardID, index);
      return TRI_ERROR_NO_ERROR;
@@ -649,12 +665,30 @@ int createDocumentOnCoordinator(
    auto body = std::make_shared<std::string>();
    for (auto const& it : shardMap) {
      if (!useMultiple) {
+       TRI_ASSERT(it.second.size() == 1);
+       auto idx = it.second.front();
+       if (idx.second.empty()) {
          body->assign(slice.toJson());
+       } else {
+         reqBuilder.clear();
+         reqBuilder.openObject();
+         TRI_SanitizeObject(slice, reqBuilder);
+         reqBuilder.add(TRI_VOC_ATTRIBUTE_KEY, VPackValue(idx.second));
+         reqBuilder.close();
+         body->assign(reqBuilder.slice().toJson());
+       }
      } else {
        reqBuilder.clear();
        reqBuilder.openArray();
-       for (auto idx : it.second) {
+       for (auto const& idx : it.second) {
-         reqBuilder.add(slice.at(idx));
+         if (idx.second.empty()) {
+           reqBuilder.add(slice.at(idx.first));
+         } else {
+           reqBuilder.openObject();
+           TRI_SanitizeObject(slice.at(idx.first), reqBuilder);
+           reqBuilder.add(TRI_VOC_ATTRIBUTE_KEY, VPackValue(idx.second));
+           reqBuilder.close();
+         }
        }
        reqBuilder.close();
        body->assign(reqBuilder.slice().toJson());

lib/Basics/VelocyPackHelper.cpp

@@ -700,15 +700,23 @@ double VelocyPackHelper::toDouble(VPackSlice const& slice, bool& failed) {
 
 uint64_t VelocyPackHelper::hashByAttributes(
     VPackSlice slice, std::vector<std::string> const& attributes,
-    bool docComplete, int& error) {
+    bool docComplete, int& error, std::string const& key) {
   error = TRI_ERROR_NO_ERROR;
   uint64_t hash = TRI_FnvHashBlockInitial();
   if (slice.isObject()) {
     for (auto const& attr: attributes) {
       VPackSlice sub = slice.get(attr);
-      if (sub.isNone() && !docComplete) {
+      if (sub.isNone()) {
+        if (attr == "_key" && !key.empty()) {
+          VPackBuilder temporaryBuilder;
+          temporaryBuilder.add(VPackValue(key));
+          temporaryBuilder.slice().normalizedHash(hash);
+          continue;
+        }
+        if (!docComplete) {
           error = TRI_ERROR_CLUSTER_NOT_ALL_SHARDING_ATTRIBUTES_GIVEN;
         }
+      }
       sub.normalizedHash(hash);
     }
   }

lib/Basics/VelocyPackHelper.h

@@ -225,7 +225,7 @@ class VelocyPackHelper {
   static double toDouble(VPackSlice const&, bool&);
 
   static uint64_t hashByAttributes(VPackSlice, std::vector<std::string> const&,
-                                   bool, int&);
+                                   bool, int&, std::string const& key = "");
 
   static arangodb::velocypack::Slice NullValue();
   static arangodb::velocypack::Slice TrueValue();