From 0669e517edaba8a133cc2d802b69364670ab0481 Mon Sep 17 00:00:00 2001
From: Jan Christoph Uhde <Jan@UhdeJC.com>
Date: Thu, 1 Jun 2017 11:27:08 +0200
Subject: [PATCH] Fix creation of millions of VPackBuilders in edge import -
 part 1 - #194

---
 .../include/velocypack/Collection.h           |  1 +
 3rdParty/velocypack/src/Collection.cpp        | 65 +++++++++++++++++++
 arangod/RestHandler/RestImportHandler.cpp     |  7 +-
 3 files changed, 69 insertions(+), 4 deletions(-)

diff --git a/3rdParty/velocypack/include/velocypack/Collection.h b/3rdParty/velocypack/include/velocypack/Collection.h
index 9afea06d46..6997eb8d50 100644
--- a/3rdParty/velocypack/include/velocypack/Collection.h
+++ b/3rdParty/velocypack/include/velocypack/Collection.h
@@ -199,6 +199,7 @@ class Collection {
                        bool mergeValues, bool nullMeansRemove = false) {
     return merge(*left, *right, mergeValues, nullMeansRemove);
   }
+  static Builder& merge(Builder& builder, Slice const& left, Slice const& right, bool mergeValues, bool nullMeansRemove = false);
 
   static void visitRecursive(
       Slice const& slice, VisitationOrder order,
diff --git a/3rdParty/velocypack/src/Collection.cpp b/3rdParty/velocypack/src/Collection.cpp
index 51c39c469a..e498a38454 100644
--- a/3rdParty/velocypack/src/Collection.cpp
+++ b/3rdParty/velocypack/src/Collection.cpp
@@ -403,6 +403,71 @@ Builder Collection::merge(Slice const& left, Slice const& right,
   return b;
 }
 
+Builder& Collection::merge(Builder& builder, Slice const& left, Slice const& right,
+                          bool mergeValues, bool nullMeansRemove) {
+  if (!left.isObject() || !right.isObject()) {
+    throw Exception(Exception::InvalidValueType, "Expecting type Object");
+  }
+
+  builder.add(Value(ValueType::Object));
+
+  std::unordered_map<std::string, Slice> rightValues;
+  {
+    ObjectIterator it(right);
+    while (it.valid()) {
+      rightValues.emplace(it.key(true).copyString(), it.value());
+      it.next();
+    }
+  }
+
+  {
+    ObjectIterator it(left);
+
+    while (it.valid()) {
+      auto key = it.key(true).copyString();
+      auto found = rightValues.find(key);
+
+      if (found == rightValues.end()) {
+        // use left value
+        builder.add(key, it.value());
+      } else if (mergeValues && it.value().isObject() &&
+                 (*found).second.isObject()) {
+        // merge both values
+        auto& value = (*found).second;
+        if (!nullMeansRemove || (!value.isNone() && !value.isNull())) {
+          Collection::merge(builder, it.value(), value, true, nullMeansRemove);
+        }
+        // clear the value in the map so its not added again
+        (*found).second = Slice();
+      } else {
+        // use right value
+        auto& value = (*found).second;
+        if (!nullMeansRemove || (!value.isNone() && !value.isNull())) {
+          builder.add(key, value);
+        }
+        // clear the value in the map so its not added again
+        (*found).second = Slice();
+      }
+      it.next();
+    }
+  }
+
+  // add remaining values that were only in right
+  for (auto& it : rightValues) {
+    auto& s = it.second;
+    if (s.isNone()) {
+      continue;
+    }
+    if (nullMeansRemove && s.isNull()) {
+      continue;
+    }
+    builder.add(std::move(it.first), s);
+  }
+
+  builder.close();
+  return builder;
+}
+
 template <Collection::VisitationOrder order>
 static bool doVisit(
     Slice const& slice,
diff --git a/arangod/RestHandler/RestImportHandler.cpp b/arangod/RestHandler/RestImportHandler.cpp
index a8f3886caf..961aafcd89 100644
--- a/arangod/RestHandler/RestImportHandler.cpp
+++ b/arangod/RestHandler/RestImportHandler.cpp
@@ -201,7 +201,7 @@ int RestImportHandler::handleSingleDocument(SingleCollectionTransaction& trx,
 
 
   // document ok, now import it
-  VPackBuilder newBuilder;
+  transaction::BuilderLeaser newBuilder(&trx);
 
   // add prefixes to _from and _to
   if (!_fromPrefix.empty() || !_toPrefix.empty()) {
@@ -239,9 +239,8 @@ int RestImportHandler::handleSingleDocument(SingleCollectionTransaction& trx,
     tempBuilder->close();
 
     if (tempBuilder->slice().length() > 0) {
-      newBuilder =
-          VPackCollection::merge(slice, tempBuilder->slice(), false, false);
-      slice = newBuilder.slice();
+      VPackCollection::merge(*(newBuilder.builder()), slice, tempBuilder->slice(), false, false);
+      slice = newBuilder->slice();
     }
   }