diff --git a/arangod/Aql/OutputAqlItemRow.cpp b/arangod/Aql/OutputAqlItemRow.cpp
index a753bf6adf..37f87fe0b4 100644
--- a/arangod/Aql/OutputAqlItemRow.cpp
+++ b/arangod/Aql/OutputAqlItemRow.cpp
@@ -48,7 +48,8 @@ OutputAqlItemRow::OutputAqlItemRow(
       _doNotCopyInputRow(copyRowBehaviour == CopyRowBehaviour::DoNotCopyInputRows),
       _outputRegisters(std::move(outputRegisters)),
       _registersToKeep(std::move(registersToKeep)),
-      _registersToClear(std::move(registersToClear))
+      _registersToClear(std::move(registersToClear)),
+      _allowSourceRowUninitialized(false)
 #ifdef ARANGODB_ENABLE_MAINTAINER_MODE
       ,
       _setBaseIndexNotUsed(true)
diff --git a/arangod/Aql/OutputAqlItemRow.h b/arangod/Aql/OutputAqlItemRow.h
index 4bcc45fe96..a0d05f06b3 100644
--- a/arangod/Aql/OutputAqlItemRow.h
+++ b/arangod/Aql/OutputAqlItemRow.h
@@ -229,6 +229,10 @@ class OutputAqlItemRow {
 #endif
     _baseIndex = index;
   }
+  // Use this function with caution! We need it only for the SortedCollectExecutor
+  void setAllowSourceRowUninitialized() {
+    _allowSourceRowUninitialized = true;
+  }
 
   // This function can be used to restore the row's invariant.
   // After setting this value numRowsWritten() rather returns
@@ -301,6 +305,8 @@ class OutputAqlItemRow {
 #ifdef ARANGODB_ENABLE_MAINTAINER_MODE
   bool _setBaseIndexNotUsed;
 #endif
+  // need this special bool for allowing an empty AqlValue inside the SortedCollectExecutor
+  bool _allowSourceRowUninitialized;
 
  private:
   size_t nextUnwrittenIndex() const noexcept { return numRowsWritten(); }
@@ -334,24 +340,30 @@ void OutputAqlItemRow::doCopyRow(InputAqlItemRow const& sourceRow, bool ignoreMi
 
   if (mustClone) {
     for (auto itemId : registersToKeep()) {
-      TRI_ASSERT(sourceRow.isInitialized());
+#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+      if (!_allowSourceRowUninitialized) {
+        TRI_ASSERT(sourceRow.isInitialized());
+      }
+#endif
       if (ignoreMissing && itemId >= sourceRow.getNrRegisters()) {
         continue;
       }
-      auto const& value = sourceRow.getValue(itemId);
-      if (!value.isEmpty()) {
-        AqlValue clonedValue = value.clone();
-        AqlValueGuard guard(clonedValue, true);
+      if (ADB_LIKELY(!_allowSourceRowUninitialized || sourceRow.isInitialized())) {
+        auto const& value = sourceRow.getValue(itemId);
+        if (!value.isEmpty()) {
+          AqlValue clonedValue = value.clone();
+          AqlValueGuard guard(clonedValue, true);
 
-        TRI_IF_FAILURE("OutputAqlItemRow::copyRow") {
-          THROW_ARANGO_EXCEPTION(TRI_ERROR_DEBUG);
-        }
-        TRI_IF_FAILURE("ExecutionBlock::inheritRegisters") {
-          THROW_ARANGO_EXCEPTION(TRI_ERROR_DEBUG);
-        }
+          TRI_IF_FAILURE("OutputAqlItemRow::copyRow") {
+            THROW_ARANGO_EXCEPTION(TRI_ERROR_DEBUG);
+          }
+          TRI_IF_FAILURE("ExecutionBlock::inheritRegisters") {
+            THROW_ARANGO_EXCEPTION(TRI_ERROR_DEBUG);
+          }
 
-        block().setValue(_baseIndex, itemId, clonedValue);
-        guard.steal();
+          block().setValue(_baseIndex, itemId, clonedValue);
+          guard.steal();
+        }
       }
     }
   } else {
diff --git a/arangod/Aql/SortedCollectExecutor.cpp b/arangod/Aql/SortedCollectExecutor.cpp
index 945c9e9315..0d7252fb2e 100644
--- a/arangod/Aql/SortedCollectExecutor.cpp
+++ b/arangod/Aql/SortedCollectExecutor.cpp
@@ -236,9 +236,14 @@ void SortedCollectExecutor::CollectGroup::groupValuesToArray(VPackBuilder& build
   builder.close();
 }
 
-void SortedCollectExecutor::CollectGroup::writeToOutput(OutputAqlItemRow& output) {
+void SortedCollectExecutor::CollectGroup::writeToOutput(OutputAqlItemRow& output, InputAqlItemRow& input) {
   // Thanks to the edge case that we have to emit a row even if we have no
   // input We cannot assert here that the input row is valid ;(
+
+  if (!input.isInitialized()) {
+    output.setAllowSourceRowUninitialized();
+  }
+
   size_t i = 0;
   for (auto& it : infos.getGroupRegisters()) {
     AqlValue val = this->groupValues[i];
@@ -289,7 +294,7 @@ std::pair<ExecutionState, NoStats> SortedCollectExecutor::produceRows(OutputAqlI
   while (true) {
     if (_fetcherDone) {
       if (_currentGroup.isValid()) {
-        _currentGroup.writeToOutput(output);
+        _currentGroup.writeToOutput(output, input);
         InputAqlItemRow input{CreateInvalidInputRowHint{}};
         _currentGroup.reset(input);
         TRI_ASSERT(!_currentGroup.isValid());
@@ -321,7 +326,7 @@ std::pair<ExecutionState, NoStats> SortedCollectExecutor::produceRows(OutputAqlI
 
       if (state == ExecutionState::DONE) {
         TRI_ASSERT(!output.produced());
-        _currentGroup.writeToOutput(output);
+        _currentGroup.writeToOutput(output, input);
         // Invalidate group
         input = InputAqlItemRow{CreateInvalidInputRowHint{}};
         _currentGroup.reset(input);
@@ -331,7 +336,7 @@ std::pair<ExecutionState, NoStats> SortedCollectExecutor::produceRows(OutputAqlI
       if (_currentGroup.isValid()) {
         // Write the current group.
         // Start a new group from input
-        _currentGroup.writeToOutput(output);
+        _currentGroup.writeToOutput(output, input);
         TRI_ASSERT(output.produced());
         _currentGroup.reset(input);  // reset and recreate new group
         if (input.isInitialized()) {
@@ -344,7 +349,7 @@ std::pair<ExecutionState, NoStats> SortedCollectExecutor::produceRows(OutputAqlI
           if (_infos.getGroupRegisters().empty()) {
             // we got exactly 0 rows as input.
             // by definition we need to emit one collect row
-            _currentGroup.writeToOutput(output);
+            _currentGroup.writeToOutput(output, input);
             TRI_ASSERT(output.produced());
           }
           TRI_ASSERT(state == ExecutionState::DONE);
diff --git a/arangod/Aql/SortedCollectExecutor.h b/arangod/Aql/SortedCollectExecutor.h
index 4f4f6a2c5c..9a3ae8adce 100644
--- a/arangod/Aql/SortedCollectExecutor.h
+++ b/arangod/Aql/SortedCollectExecutor.h
@@ -157,7 +157,7 @@ class SortedCollectExecutor {
     void addLine(InputAqlItemRow& input);
     bool isSameGroup(InputAqlItemRow& input);
     void groupValuesToArray(VPackBuilder& builder);
-    void writeToOutput(OutputAqlItemRow& output);
+    void writeToOutput(OutputAqlItemRow& output, InputAqlItemRow& input);
   };
 
  public:
diff --git a/tests/js/server/aql/aql-optimizer-collect-count.js b/tests/js/server/aql/aql-optimizer-collect-count.js
index 5dac71ff00..4b6c78fbfc 100644
--- a/tests/js/server/aql/aql-optimizer-collect-count.js
+++ b/tests/js/server/aql/aql-optimizer-collect-count.js
@@ -379,6 +379,20 @@ function optimizerCountTestSuite () {
         var plan = AQL_EXPLAIN(query).plan;
         assertNotEqual(-1, plan.rules.indexOf("collect-in-cluster"));
       }
+    },
+
+    testCollectAggregateUndefined: function () {
+      var randomDocumentID = db["UnitTestsCollection"].any()._id;
+      var query = 'LET start = DOCUMENT("' + randomDocumentID + '")._key for i in [] collect aggregate count = count(i) return {count, start}';
+      var bindParams = {};
+      var options = {optimizer: {rules: ['-remove-unnecessary-calculations','-remove-unnecessary-calculations-2']}};
+
+      var results = db._query(query, bindParams, options).toArray();
+      // expectation is that we exactly get one result
+      // count will be 0, start will be undefined
+      assertEqual(1, results.length);
+      assertEqual(0, results[0].count);
+      assertEqual(undefined, results[0].start);
     }
 
   };