ManagedDocumentResult now manages data when using setManaged

arangod/MMFiles/MMFilesCollection.cpp

@@ -398,12 +398,12 @@ bool MMFilesCollection::OpenIterator(MMFilesMarker const* marker, MMFilesCollect
     }
     
     if (++data->_operations % 1024 == 0) {
-      data->_mmdr.clear();
+      data->_mmdr.reset();
     }
   } else if (type == TRI_DF_MARKER_VPACK_REMOVE) {
     res = OpenIteratorHandleDeletionMarker(marker, datafile, data);
     if (++data->_operations % 1024 == 0) {
-      data->_mmdr.clear();
+      data->_mmdr.reset();
     }
   } else {
     if (type == TRI_DF_MARKER_HEADER) {
@@ -1837,7 +1837,7 @@ bool MMFilesCollection::readDocument(transaction::Methods* trx,
   TRI_voc_rid_t revisionId = tkn->revisionId();
   uint8_t const* vpack = lookupRevisionVPack(revisionId);
   if (vpack != nullptr) {
-    result.addExisting(vpack, revisionId);
+    result.setUnmanaged(vpack, revisionId);
     return true;
   } 
   return false;
@@ -1852,7 +1852,7 @@ bool MMFilesCollection::readDocumentConditional(
   TRI_ASSERT(revisionId != 0);
   uint8_t const* vpack  = lookupRevisionVPackConditional(revisionId, maxTick, true);
   if (vpack != nullptr) {
-    result.addExisting(vpack, revisionId);
+    result.setUnmanaged(vpack, revisionId);
     return true;
   } 
   return false;
@@ -2705,7 +2705,7 @@ int MMFilesCollection::insert(transaction::Methods* trx,
   if (res == TRI_ERROR_NO_ERROR) {
     uint8_t const* vpack = lookupRevisionVPack(revisionId);
     if (vpack != nullptr) {
-      result.addExisting(vpack, revisionId);
+      result.setUnmanaged(vpack, revisionId);
     }
 
     // store the tick that was used for writing the document
@@ -3030,7 +3030,7 @@ int MMFilesCollection::update(arangodb::transaction::Methods* trx,
 
   if (newSlice.length() <= 1) {
     // no need to do anything
-    result = previous;
+    result = std::move(previous);
     if (_logicalCollection->waitForSync()) {
       options.waitForSync = true;
     }
@@ -3080,7 +3080,7 @@ int MMFilesCollection::update(arangodb::transaction::Methods* trx,
     
     if (oldRevisionId == revisionId) {
       // update with same revision id => can happen if isRestore = true
-      result.clear();
+      result.reset();
     }
 
     res = updateDocument(trx, oldRevisionId, oldDoc, revisionId, newDoc,
@@ -3098,7 +3098,7 @@ int MMFilesCollection::update(arangodb::transaction::Methods* trx,
   } else {
     uint8_t const* vpack = lookupRevisionVPack(revisionId);
     if (vpack != nullptr) {
-      result.addExisting(vpack, revisionId);
+      result.setUnmanaged(vpack, revisionId);
     }
     if (options.waitForSync) {
       // store the tick that was used for writing the new document
@@ -3203,7 +3203,7 @@ int MMFilesCollection::replace(
     
     if (oldRevisionId == revisionId) {
       // update with same revision id => can happen if isRestore = true
-      result.clear();
+      result.reset();
     }
 
     res = updateDocument(trx, oldRevisionId, oldDoc, revisionId, newDoc,
@@ -3221,11 +3221,11 @@ int MMFilesCollection::replace(
   } else {
     if (oldRevisionId == revisionId) {
       // update with same revision id => can happen if isRestore = true
-      result.clear();
+      result.reset();
     }
     uint8_t const* vpack = lookupRevisionVPack(revisionId);
     if (vpack != nullptr) {
-      result.addExisting(vpack, revisionId);
+      result.setUnmanaged(vpack, revisionId);
     }
 
     if (options.waitForSync) {
@@ -3542,7 +3542,7 @@ int MMFilesCollection::lookupDocument(transaction::Methods* trx,
     TRI_voc_rid_t revisionId = element.revisionId();
     uint8_t const* vpack = lookupRevisionVPack(revisionId);
     if (vpack != nullptr) {
-      result.addExisting(vpack, revisionId);
+      result.setUnmanaged(vpack, revisionId);
     }
     return TRI_ERROR_NO_ERROR;
   }

arangod/RocksDBEngine/RocksDBCollection.cpp

@@ -134,7 +134,7 @@ size_t RocksDBCollection::memory() const {
 }
 
 void RocksDBCollection::open(bool ignoreErrors) {
-  THROW_ARANGO_NOT_YET_IMPLEMENTED();
+  //THROW_ARANGO_NOT_YET_IMPLEMENTED();
 }
 
 /// @brief iterate all markers of a collection on load

arangod/VocBase/ManagedDocumentResult.h

@@ -24,35 +24,86 @@
 #ifndef ARANGOD_VOC_BASE_MANAGED_DOCUMENT_RESULT_H
 #define ARANGOD_VOC_BASE_MANAGED_DOCUMENT_RESULT_H 1
 
+#include "velocypack/Slice.h"
+#include "velocypack/Buffer.h"
+#include "velocypack/velocypack-aliases.h"
 #include "Basics/Common.h"
 
 namespace arangodb {
 
 class ManagedDocumentResult {
  public:
-  ManagedDocumentResult() : _vpack(nullptr), _lastRevisionId(0) {}
-  ~ManagedDocumentResult() = default;
+  ManagedDocumentResult() : _length(0), _lastRevisionId(0), _vpack(nullptr),
+                             _managed(false) {}
+  ~ManagedDocumentResult() { reset(); }
+  ManagedDocumentResult(ManagedDocumentResult const& other) = delete;
+  ManagedDocumentResult& operator=(ManagedDocumentResult const& other) = delete;
 
-  inline uint8_t const* vpack() const { 
-    TRI_ASSERT(_vpack != nullptr); 
-    return _vpack; 
+  ManagedDocumentResult& operator=(ManagedDocumentResult&& other){
+    if (other._managed){
+      reset();
+      _vpack = other._vpack;
+      _length = other._length;
+      _lastRevisionId = other._lastRevisionId;
+      _managed = true;
+      other._managed = false;
+      other.reset();
+    } else {
+      setUnmanaged(other._vpack, other._lastRevisionId);
+    }
+    return *this;
   }
-  
-  inline void addExisting(uint8_t const* vpack, TRI_voc_rid_t revisionId) {
-    _vpack = vpack;
+
+  ManagedDocumentResult(ManagedDocumentResult&& other) = delete;
+
+  inline uint8_t const* vpack() const {
+    TRI_ASSERT(_vpack != nullptr);
+    return _vpack;
+  }
+
+  //add unmanaged vpack 
+  inline void setUnmanaged(uint8_t const* vpack, TRI_voc_rid_t revisionId) {
+    if(_managed) {
+      reset();
+    }
+    TRI_ASSERT(_length == 0);
+    _vpack = const_cast<uint8_t*>(vpack);
     _lastRevisionId = revisionId;
   }
 
+  inline void setManaged(uint8_t const* vpack, TRI_voc_rid_t revisionId) {
+    VPackSlice slice(vpack);
+    auto newLen = slice.byteSize();
+    if (_length >= newLen && _managed){
+      std::memcpy(_vpack, vpack, newLen);
+    } else {
+      reset();
+      _vpack = new uint8_t[newLen];
+      std::memcpy(_vpack, vpack, newLen);
+      _length=newLen;
+    }
+    _lastRevisionId = revisionId;
+    _managed = true;
+  }
+
   inline TRI_voc_rid_t lastRevisionId() const { return _lastRevisionId; }
 
-  void clear() {
+  void reset() noexcept {
+    if(_managed) {
+      delete _vpack;
+    }
     _vpack = nullptr;
     _lastRevisionId = 0;
+    _managed = false;
+    _length = 0;
   }
 
  private:
-  uint8_t const* _vpack;
+  uint64_t _length;
   TRI_voc_rid_t _lastRevisionId;
+  uint8_t* _vpack;
+  bool _managed;
+
 };
 
 }