diff --git a/CHANGELOG b/CHANGELOG
index 3df2f4b02f..1ca28a41f8 100644
--- a/CHANGELOG
+++ b/CHANGELOG
@@ -1,7 +1,14 @@
+v3.3.3 (XXXX-XX-XX)
+-------------------
+
+* fixed issue #4255: AQL SORT consuming too much memory
+
+
 v3.3.2 (2018-01-04)
 -------------------
 
-* fixed issue #4199: Internal failure: JavaScript exception in file 'arangosh.js' at 98,7: ArangoError 4: Expecting type String
+* fixed issue #4199: Internal failure: JavaScript exception in file 'arangosh.js' 
+  at 98,7: ArangoError 4: Expecting type String 
 
 * fixed issue in agency supervision with a good server being left in
   failedServers
diff --git a/arangod/Aql/ClusterBlocks.cpp b/arangod/Aql/ClusterBlocks.cpp
index 168e3beea4..c57e312ed2 100644
--- a/arangod/Aql/ClusterBlocks.cpp
+++ b/arangod/Aql/ClusterBlocks.cpp
@@ -306,7 +306,8 @@ AqlItemBlock* GatherBlock::getSome(size_t atLeast, size_t atMost) {
   size_t toSend = (std::min)(available, atMost);  // nr rows in outgoing block
 
   // the following is similar to AqlItemBlock's slice method . . .
-  std::unordered_set<AqlValue> cache;
+  std::vector<std::unordered_map<AqlValue, AqlValue>> cache;
+  cache.resize(_gatherBlockBuffer.size());
 
   // comparison function
   OurLessThan ourLessThan(_trx, _gatherBlockBuffer, _sortRegisters);
@@ -341,9 +342,9 @@ AqlItemBlock* GatherBlock::getSome(size_t atLeast, size_t atMost) {
       TRI_ASSERT(!_gatherBlockBuffer[val.first].empty());
       AqlValue const& x(_gatherBlockBuffer[val.first].front()->getValueReference(val.second, col));
       if (!x.isEmpty()) {
-        auto it = cache.find(x);
+        auto it = cache[val.first].find(x);
 
-        if (it == cache.end()) {
+        if (it == cache[val.first].end()) {
           AqlValue y = x.clone();
           try {
             res->setValue(i, col, y);
@@ -351,9 +352,9 @@ AqlItemBlock* GatherBlock::getSome(size_t atLeast, size_t atMost) {
             y.destroy();
             throw;
           }
-          cache.emplace(y);
+          cache[val.first].emplace(x, y);
         } else {
-          res->setValue(i, col, (*it));
+          res->setValue(i, col, (*it).second);
         }
       }
     }
@@ -382,6 +383,7 @@ AqlItemBlock* GatherBlock::getSome(size_t atLeast, size_t atMost) {
         // more data for the shard for which we have no more local
         // values. 
         getBlock(val.first, atLeast, atMost);
+        cache[val.first].clear();
         // note that if getBlock() returns false here, this is not
         // a problem, because the sort function used takes care of
         // this
diff --git a/arangod/Aql/SortBlock.cpp b/arangod/Aql/SortBlock.cpp
index 5138564222..f7cf5980f2 100644
--- a/arangod/Aql/SortBlock.cpp
+++ b/arangod/Aql/SortBlock.cpp
@@ -31,9 +31,11 @@ using namespace arangodb::aql;
 
 SortBlock::SortBlock(ExecutionEngine* engine, SortNode const* en)
     : ExecutionBlock(engine, en), _sortRegisters(), _stable(en->_stable), _mustFetchAll(true) {
+  
+  auto regPlan = en->getRegisterPlan();
   for (auto const& p : en->_elements) {
-    auto it = en->getRegisterPlan()->varInfo.find(p.var->id);
-    TRI_ASSERT(it != en->getRegisterPlan()->varInfo.end());
+    auto it = regPlan->varInfo.find(p.var->id);
+    TRI_ASSERT(it != regPlan->varInfo.end());
     TRI_ASSERT(it->second.registerId < ExecutionNode::MaxRegisterId);
     _sortRegisters.emplace_back(
         std::make_pair(it->second.registerId, p.ascending));
@@ -126,7 +128,7 @@ void SortBlock::doSorting() {
     count = 0;
     RegisterId const nrRegs = _buffer.front()->getNrRegs();
 
-    std::unordered_set<AqlValue> cache;
+    std::unordered_map<AqlValue, AqlValue> cache;
 
     // install the rearranged values from _buffer into newbuffer
 
@@ -144,7 +146,6 @@ void SortBlock::doSorting() {
         throw;
       }
 
-      cache.clear();
       // only copy as much as needed!
       for (size_t i = 0; i < sizeNext; i++) {
         for (RegisterId j = 0; j < nrRegs; j++) {
@@ -160,7 +161,7 @@ void SortBlock::doSorting() {
               // the new block already has either a copy or stolen
               // the AqlValue:
               _buffer[coords[count].first]->eraseValue(coords[count].second, j);
-              next->setValue(i, j, (*it));
+              next->setValue(i, j, (*it).second);
             } else {
               // We need to copy a, if it has already been stolen from
               // its original buffer, which we know by looking at the
@@ -174,7 +175,7 @@ void SortBlock::doSorting() {
                   TRI_IF_FAILURE("SortBlock::doSortingCache") {
                     THROW_ARANGO_EXCEPTION(TRI_ERROR_DEBUG);
                   }
-                  cache.emplace(b);
+                  cache.emplace(a, b);
                 } catch (...) {
                   b.destroy();
                   throw;
@@ -186,15 +187,14 @@ void SortBlock::doSorting() {
                   }
                   next->setValue(i, j, b);
                 } catch (...) {
-                  cache.erase(b);
+                  cache.erase(a);
                   b.destroy();
                   throw;
                 }
                 // It does not matter whether the following works or not,
                 // since the original block keeps its responsibility
                 // for a:
-                _buffer[coords[count].first]->eraseValue(coords[count].second,
-                                                         j);
+                _buffer[coords[count].first]->eraseValue(coords[count].second, j);
               } else {
                 TRI_IF_FAILURE("SortBlock::doSortingNext2") {
                   THROW_ARANGO_EXCEPTION(TRI_ERROR_DEBUG);
@@ -203,13 +203,12 @@ void SortBlock::doSorting() {
                 // steal it:
                 next->setValue(i, j, a);
                 _buffer[coords[count].first]->steal(a);
-                _buffer[coords[count].first]->eraseValue(coords[count].second,
-                                                         j);
-                // If this has worked, responsibility is now with the
-                // new block or indeed with us!
+                _buffer[coords[count].first]->eraseValue(coords[count].second, j);
+                // This might throw as well, however, the responsibility
+                // is already with the new block.
                 // If the following does not work, we will create a
                 // few unnecessary copies, but this does not matter:
-                cache.emplace(a);
+                cache.emplace(a, a);
               }
             }
           }
diff --git a/js/server/tests/aql/aql-failures-noncluster.js b/js/server/tests/aql/aql-failures-noncluster.js
index 36b8ef4dca..4474cc5fd3 100644
--- a/js/server/tests/aql/aql-failures-noncluster.js
+++ b/js/server/tests/aql/aql-failures-noncluster.js
@@ -259,9 +259,9 @@ function ahuacatlFailureSuite () {
     testSortBlock5 : function () {
       internal.debugSetFailAt("SortBlock::doSortingNext2");
       // we need values that are >= 16 bytes long
-      assertFailingQuery("FOR i IN " + c.name() + " COLLECT key = i._key SORT CONCAT('xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx', key) RETURN key");
-      assertFailingQuery("FOR i IN " + c.name() + " COLLECT key = i.value SORT CONCAT('xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx', key) RETURN key");
-      assertFailingQuery("FOR i IN " + c.name() + " COLLECT key = i.value2 SORT CONCAT('xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx', key) RETURN key");
+      assertFailingQuery("LET x = NOOPT('xxxxxxxxxxxxxxxxxxxx') FOR i IN " + c.name() + " COLLECT key = i._key SORT CONCAT('xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx', key) RETURN { key, x }");
+      assertFailingQuery("LET x = NOOPT('xxxxxxxxxxxxxxxxxxxx') FOR i IN " + c.name() + " COLLECT key = i.value SORT CONCAT('xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx', key) RETURN { key, x }");
+      assertFailingQuery("LET x = NOOPT('xxxxxxxxxxxxxxxxxxxx') FOR i IN " + c.name() + " COLLECT key = i.value2 SORT CONCAT('xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx', key) RETURN { key, x }");
     },
 
 ////////////////////////////////////////////////////////////////////////////////