Feature/limit executor passthrough (#9162)

* Some refactoring to implement helper methods iff necessary

* Updated comments

* Added static assertions

* Re-enabled check for maintainer mode

* Allow pass-through for LimitExecutor. This is not yet working with fullCount correctly!

* Get fullCount before returning the last row

* Adapted fullCount tests: LimitExecutor doesn't lie about the state of the last row anymore

* Use correct row

* Handle LimitExecutor-stats during fetchBlockForPassthrough

* Fixed LimitExecutorTest

* Removed _stats member

* Added more catch tests

* Bugfix for fullStats

* Removed an erroneous assertion

* Implemented LimitExecutor::skipRows

* Avoid name clash of class enum member to please MSVC

* Test MSVC

* Revert "Test MSVC"

This reverts commit d8325d95318bb83b79c8c4cc9f886d36e484c870.

* CREATE_NEW seems to be taken, too. Next try for MSVC

* Built VPack to ItemBlock helpers, and made SingleRowFetcherHelper more general

* Removed ostream operator because MSVC doesn't like it...

* Implemented SingleRowFetcherHelper::fetchBlockForPassthrough, plus a small framework to test executors

* Removed erroneous include

* Moved code in separate files

* Began writing the extended schema of runExecutor

* Added output operator for ExecutorCall

* Fixed removeWaiting

* Fixed expected output

* Added skipRows to runExecutor

* Added two tests

* Allow empty blocks

* Added two more tests

* Added another test

* Built two different parametrized classes for LimitExecutor

* Some cleanup

* Even more tests

* Fixed a bug found by the new tests

* Fix compile error on windows

* Use native matrix representation in tests

* Updated some comments

arangod/Aql/CalculationExecutor.h

@@ -108,11 +108,9 @@ class CalculationExecutor {
    */
   inline std::pair<ExecutionState, Stats> produceRows(OutputAqlItemRow& output);
 
-  inline std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t) const {
+  inline std::tuple<ExecutionState, Stats, SharedAqlItemBlockPtr> fetchBlockForPassthrough(size_t atMost) {
-    TRI_ASSERT(false);
+    auto rv = _fetcher.fetchBlockForPassthrough(atMost);
-    THROW_ARANGO_EXCEPTION_MESSAGE(
+    return {rv.first, {}, std::move(rv.second)};
-        TRI_ERROR_INTERNAL,
-        "Logic_error, prefetching number fo rows not supported");
   }
 
  private:

arangod/Aql/EnumerateCollectionExecutor.h

@@ -123,13 +123,6 @@ class EnumerateCollectionExecutor {
   void setProducingFunction(DocumentProducingFunction const& documentProducer) {
     _documentProducer = documentProducer;
   };
-  
-  inline std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t) const {
-    TRI_ASSERT(false);
-    THROW_ARANGO_EXCEPTION_MESSAGE(
-        TRI_ERROR_INTERNAL,
-        "Logic_error, prefetching number fo rows not supported");
-  }
 
   void initializeCursor();
 

arangod/Aql/EnumerateListExecutor.h

@@ -96,13 +96,6 @@ class EnumerateListExecutor {
    */
   std::pair<ExecutionState, Stats> produceRows(OutputAqlItemRow& output);
 
-  inline std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t atMost) const {
-    TRI_ASSERT(false);
-    THROW_ARANGO_EXCEPTION_MESSAGE(
-        TRI_ERROR_INTERNAL,
-        "Logic_error, prefetching number fo rows not supported");
-  }
-
  private:
   AqlValue getAqlValue(AqlValue const& inVarReg, size_t const& pos, bool& mustDestroy);
   void initialize();

arangod/Aql/ExecutionBlock.h

@@ -83,6 +83,14 @@ class ExecutionBlock {
   /// @brief batch size value
   static constexpr inline size_t DefaultBatchSize() { return 1000; }
 
+  /// @brief Number to use when we skip all. Should really be inf, but don't
+  /// use something near std::numeric_limits<size_t>::max() to avoid overflows
+  /// in calculations.
+  /// This is used as an argument for skipSome(), e.g. when counting everything.
+  /// Setting this to any other value >0 does not (and must not) affect the
+  /// results. It's only to reduce the number of necessary skipSome calls.
+  static constexpr inline size_t SkipAllSize() { return 1000000000; }
+
   /// @brief Methods for execution
   /// Lifecycle is:
   ///    CONSTRUCTOR

arangod/Aql/ExecutionBlockImpl.cpp

@@ -89,6 +89,8 @@ using namespace arangodb::aql;
 
 CREATE_HAS_MEMBER_CHECK(initializeCursor, hasInitializeCursor);
 CREATE_HAS_MEMBER_CHECK(skipRows, hasSkipRows);
+CREATE_HAS_MEMBER_CHECK(fetchBlockForPassthrough, hasFetchBlockForPassthrough);
+CREATE_HAS_MEMBER_CHECK(expectedNumberOfRows, hasExpectedNumberOfRows);
 
 template <class Executor>
 ExecutionBlockImpl<Executor>::ExecutionBlockImpl(ExecutionEngine* engine,
@@ -217,7 +219,7 @@ std::unique_ptr<OutputAqlItemRow> ExecutionBlockImpl<Executor>::createOutputRow(
 namespace arangodb {
 namespace aql {
 
-enum class SkipVariants { FETCHER, EXECUTOR, DEFAULT };
+enum class SkipVariants { FETCHER, EXECUTOR, GET_SOME };
 
 // Specifying the namespace here is important to MSVC.
 template <enum arangodb::aql::SkipVariants>
@@ -229,7 +231,7 @@ struct ExecuteSkipVariant<SkipVariants::FETCHER> {
   static std::tuple<ExecutionState, typename Executor::Stats, size_t> executeSkip(
       Executor& executor, typename Executor::Fetcher& fetcher, size_t toSkip) {
     auto res = fetcher.skipRows(toSkip);
-    return std::make_tuple(res.first, typename Executor::Stats{}, res.second); // tupple, cannot use initializer list due to build failure
+    return std::make_tuple(res.first, typename Executor::Stats{}, res.second);  // tuple, cannot use initializer list due to build failure
   }
 };
 
@@ -243,14 +245,14 @@ struct ExecuteSkipVariant<SkipVariants::EXECUTOR> {
 };
 
 template <>
-struct ExecuteSkipVariant<SkipVariants::DEFAULT> {
+struct ExecuteSkipVariant<SkipVariants::GET_SOME> {
   template <class Executor>
   static std::tuple<ExecutionState, typename Executor::Stats, size_t> executeSkip(
       Executor& executor, typename Executor::Fetcher& fetcher, size_t toSkip) {
     // this function should never be executed
     TRI_ASSERT(false);
     // Make MSVC happy:
-    return std::make_tuple(ExecutionState::DONE, typename Executor::Stats{}, 0); // tupple, cannot use initializer list due to build failure
+    return std::make_tuple(ExecutionState::DONE, typename Executor::Stats{}, 0);  // tuple, cannot use initializer list due to build failure
   }
 };
 
@@ -278,15 +280,20 @@ static SkipVariants constexpr skipType() {
                      std::is_same<Executor, IResearchViewExecutor<true>>::value ||
                      std::is_same<Executor, IResearchViewMergeExecutor<false>>::value ||
                      std::is_same<Executor, IResearchViewMergeExecutor<true>>::value ||
-                     std::is_same<Executor, EnumerateCollectionExecutor>::value),
+                     std::is_same<Executor, EnumerateCollectionExecutor>::value ||
+                     std::is_same<Executor, LimitExecutor>::value),
                 "Unexpected executor for SkipVariants::EXECUTOR");
 
+  // The LimitExecutor will not work correctly with SkipVariants::FETCHER!
+  static_assert(!std::is_same<Executor, LimitExecutor>::value || useFetcher,
+      "LimitExecutor needs to implement skipRows() to work correctly");
+
   if (useExecutor) {
     return SkipVariants::EXECUTOR;
   } else if (useFetcher) {
     return SkipVariants::FETCHER;
   } else {
-    return SkipVariants::DEFAULT;
+    return SkipVariants::GET_SOME;
   }
 }
 
@@ -299,7 +306,7 @@ std::pair<ExecutionState, size_t> ExecutionBlockImpl<Executor>::skipSome(size_t
 
   constexpr SkipVariants customSkipType = skipType<Executor>();
 
-  if (customSkipType == SkipVariants::DEFAULT) {
+  if (customSkipType == SkipVariants::GET_SOME) {
     atMost = std::min(atMost, DefaultBatchSize());
     auto res = getSomeWithoutTrace(atMost);
 
@@ -496,16 +503,58 @@ std::pair<ExecutionState, Result> ExecutionBlockImpl<SubqueryExecutor<false>>::s
 }  // namespace aql
 }  // namespace arangodb
 
-template <class Executor>
+namespace arangodb {
-std::pair<ExecutionState, SharedAqlItemBlockPtr> ExecutionBlockImpl<Executor>::requestWrappedBlock(
+namespace aql {
-    size_t nrItems, RegisterCount nrRegs) {
+
-  SharedAqlItemBlockPtr block;
+// The constant "PASSTHROUGH" is somehow reserved with MSVC.
-  if /* constexpr */ (Executor::Properties::allowsBlockPassthrough) {
+enum class RequestWrappedBlockVariant { DEFAULT , PASS_THROUGH , INPUTRESTRICTED };
-    // If blocks can be passed through, we do not create new blocks.
+
-    // Instead, we take the input blocks from the fetcher and reuse them.
+// Specifying the namespace here is important to MSVC.
+template <enum arangodb::aql::RequestWrappedBlockVariant>
+struct RequestWrappedBlock {};
+
+template <>
+struct RequestWrappedBlock<RequestWrappedBlockVariant::DEFAULT> {
+  /**
+   * @brief Default requestWrappedBlock() implementation. Just get a new block
+   *        from the AqlItemBlockManager.
+   */
+  template <class Executor>
+  static std::pair<ExecutionState, SharedAqlItemBlockPtr> run(
+#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+      typename Executor::Infos const&,
+#endif
+      Executor& executor, ExecutionEngine& engine, size_t nrItems, RegisterCount nrRegs) {
+    return {ExecutionState::HASMORE,
+            engine.itemBlockManager().requestBlock(nrItems, nrRegs)};
+  }
+};
+
+template <>
+struct RequestWrappedBlock<RequestWrappedBlockVariant::PASS_THROUGH> {
+  /**
+   * @brief If blocks can be passed through, we do not create new blocks.
+   *        Instead, we take the input blocks and reuse them.
+   */
+  template <class Executor>
+  static std::pair<ExecutionState, SharedAqlItemBlockPtr> run(
+#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+      typename Executor::Infos const& infos,
+#endif
+      Executor& executor, ExecutionEngine& engine, size_t nrItems, RegisterCount nrRegs) {
+    static_assert(
+        Executor::Properties::allowsBlockPassthrough,
+        "This function can only be used with executors supporting `allowsBlockPassthrough`");
+    static_assert(hasFetchBlockForPassthrough<Executor>::value,
+                  "An Executor with allowsBlockPassthrough must implement "
+                  "fetchBlockForPassthrough");
+
+    SharedAqlItemBlockPtr block;
 
     ExecutionState state;
-    std::tie(state, block) = _rowFetcher.fetchBlockForPassthrough(nrItems);
+    typename Executor::Stats executorStats;
+    std::tie(state, executorStats, block) = executor.fetchBlockForPassthrough(nrItems);
+    engine._stats += executorStats;
 
     if (state == ExecutionState::WAITING) {
       TRI_ASSERT(block == nullptr);
@@ -523,22 +572,45 @@ std::pair<ExecutionState, SharedAqlItemBlockPtr> ExecutionBlockImpl<Executor>::r
     TRI_ASSERT(block->getNrRegs() == nrRegs);
 #ifdef ARANGODB_ENABLE_MAINTAINER_MODE
     // Check that all output registers are empty.
-    for (auto const& reg : *infos().getOutputRegisters()) {
+    for (auto const& reg : *infos.getOutputRegisters()) {
       for (size_t row = 0; row < block->size(); row++) {
         AqlValue const& val = block->getValueReference(row, reg);
         TRI_ASSERT(val.isEmpty());
       }
     }
 #endif
-  } else if /* constexpr */ (Executor::Properties::inputSizeRestrictsOutputSize) {
+
-    // The SortExecutor should refetch a block to save memory in case if only few elements to sort
+    return {ExecutionState::HASMORE, block};
+  }
+};
+
+template <>
+struct RequestWrappedBlock<RequestWrappedBlockVariant::INPUTRESTRICTED> {
+  /**
+   * @brief If the executor can set an upper bound on the output size knowing
+   *        the input size, usually because size(input) >= size(output), let it
+   *        prefetch an input block to give us this upper bound.
+   *        Only then we allocate a new block with at most this upper bound.
+   */
+  template <class Executor>
+  static std::pair<ExecutionState, SharedAqlItemBlockPtr> run(
+#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+      typename Executor::Infos const&,
+#endif
+      Executor& executor, ExecutionEngine& engine, size_t nrItems, RegisterCount nrRegs) {
+    static_assert(
+        Executor::Properties::inputSizeRestrictsOutputSize,
+        "This function can only be used with executors supporting `inputSizeRestrictsOutputSize`");
+    static_assert(hasExpectedNumberOfRows<Executor>::value,
+                  "An Executor with inputSizeRestrictsOutputSize must "
+                  "implement expectedNumberOfRows");
+
+    SharedAqlItemBlockPtr block;
+
     ExecutionState state;
     size_t expectedRows = 0;
     // Note: this might trigger a prefetch on the rowFetcher!
-    // TODO For the LimitExecutor, this call happens too early. See the more
+    std::tie(state, expectedRows) = executor.expectedNumberOfRows(nrItems);
-    //  elaborate comment on
-    //  LimitExecutor::Properties::inputSizeRestrictsOutputSize.
-    std::tie(state, expectedRows) = _executor.expectedNumberOfRows(nrItems);
     if (state == ExecutionState::WAITING) {
       return {state, nullptr};
     }
@@ -547,12 +619,46 @@ std::pair<ExecutionState, SharedAqlItemBlockPtr> ExecutionBlockImpl<Executor>::r
       TRI_ASSERT(state == ExecutionState::DONE);
       return {state, nullptr};
     }
-    block = requestBlock(nrItems, nrRegs);
+    block = engine.itemBlockManager().requestBlock(nrItems, nrRegs);
-  } else {
-    block = requestBlock(nrItems, nrRegs);
-  }
 
-  return {ExecutionState::HASMORE, std::move(block)};
+    return {ExecutionState::HASMORE, block};
+  }
+};
+
+}  // namespace aql
+}  // namespace arangodb
+
+template <class Executor>
+std::pair<ExecutionState, SharedAqlItemBlockPtr> ExecutionBlockImpl<Executor>::requestWrappedBlock(
+    size_t nrItems, RegisterCount nrRegs) {
+  static_assert(!Executor::Properties::allowsBlockPassthrough ||
+                    !Executor::Properties::inputSizeRestrictsOutputSize,
+                "At most one of Properties::allowsBlockPassthrough or "
+                "Properties::inputSizeRestrictsOutputSize should be true for "
+                "each Executor");
+  static_assert(
+      Executor::Properties::allowsBlockPassthrough ==
+          hasFetchBlockForPassthrough<Executor>::value,
+      "Executors should implement the method fetchBlockForPassthrough() iff "
+      "Properties::allowsBlockPassthrough is true");
+  static_assert(
+      Executor::Properties::inputSizeRestrictsOutputSize ==
+          hasExpectedNumberOfRows<Executor>::value,
+      "Executors should implement the method expectedNumberOfRows() iff "
+      "Properties::inputSizeRestrictsOutputSize is true");
+
+  constexpr RequestWrappedBlockVariant variant =
+      Executor::Properties::allowsBlockPassthrough
+          ? RequestWrappedBlockVariant::PASS_THROUGH
+          : Executor::Properties::inputSizeRestrictsOutputSize
+                ? RequestWrappedBlockVariant::INPUTRESTRICTED
+                : RequestWrappedBlockVariant::DEFAULT;
+
+  return RequestWrappedBlock<variant>::run(
+#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+      infos(),
+#endif
+      executor(), *_engine, nrItems, nrRegs);
 }
 
 /// @brief request an AqlItemBlock from the memory manager

arangod/Aql/IResearchViewExecutor.h

@@ -148,13 +148,6 @@ class IResearchViewExecutorBase {
   using Infos = IResearchViewExecutorInfos;
   using Stats = IResearchViewStats;
 
-  inline std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t atMost) const {
-    TRI_ASSERT(false);
-    THROW_ARANGO_EXCEPTION_MESSAGE(
-        TRI_ERROR_INTERNAL,
-        "Logic_error, prefetching number fo rows not supported");
-  }
-
   /**
    * @brief produce the next Row of Aql Values.
    *

arangod/Aql/IdExecutor.h

@@ -73,7 +73,6 @@ class ExecutionBlockImpl<IdExecutor<void>> : public ExecutionBlock {
         _currentDependency(0),
         _outputRegister(outputRegister),
         _doCount(doCount) {
-
     // already insert ourselves into the statistics results
     if (_profile >= PROFILE_LEVEL_BLOCKS) {
       _engine->_stats.nodes.emplace(node->id(), ExecutionStats::Node());
@@ -176,12 +175,9 @@ class IdExecutor {
    */
   std::pair<ExecutionState, Stats> produceRows(OutputAqlItemRow& output);
 
-  inline std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t atMost) const {
+  inline std::tuple<ExecutionState, Stats, SharedAqlItemBlockPtr> fetchBlockForPassthrough(size_t atMost) {
-    // This is passthrough!
+    auto rv = _fetcher.fetchBlockForPassthrough(atMost);
-    TRI_ASSERT(false);
+    return {rv.first, {}, std::move(rv.second)};
-    THROW_ARANGO_EXCEPTION_MESSAGE(
-        TRI_ERROR_INTERNAL,
-        "Logic_error, prefetching number fo rows not supported");
   }
 
  private:

arangod/Aql/IndexExecutor.h

@@ -218,13 +218,6 @@ class IndexExecutor {
   std::tuple<ExecutionState, Stats, size_t> skipRows(size_t toSkip);
 
  public:
-  inline std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t atMost) const {
-    TRI_ASSERT(false);
-    THROW_ARANGO_EXCEPTION_MESSAGE(
-        TRI_ERROR_INTERNAL,
-        "Logic_error, prefetching number fo rows not supported");
-  }
-
   void initializeCursor();
 
  private:

arangod/Aql/KShortestPathsExecutor.h

@@ -156,12 +156,6 @@ class KShortestPathsExecutor {
    * @return ExecutionState, and if successful exactly one new Row of AqlItems.
    */
   std::pair<ExecutionState, Stats> produceRows(OutputAqlItemRow& output);
-  inline std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t) const {
-    TRI_ASSERT(false);
-    THROW_ARANGO_EXCEPTION_MESSAGE(
-        TRI_ERROR_INTERNAL,
-        "Logic_error, prefetching number fo rows not supported");
-  }
 
  private:
   /**

arangod/Aql/LimitExecutor.cpp

@@ -53,50 +53,80 @@ LimitExecutorInfos::LimitExecutorInfos(RegisterId nrInputRegisters, RegisterId n
       _fullCount(fullCount) {}
 
 LimitExecutor::LimitExecutor(Fetcher& fetcher, Infos& infos)
-    : _infos(infos), _fetcher(fetcher){};
+    : _infos(infos),
+      _fetcher(fetcher),
+      _lastRowToOutput(CreateInvalidInputRowHint{}),
+      _stateOfLastRowToOutput(ExecutionState::HASMORE) {}
 LimitExecutor::~LimitExecutor() = default;
 
+std::pair<ExecutionState, LimitStats> LimitExecutor::skipOffset() {
+  ExecutionState state;
+  size_t skipped;
+  std::tie(state, skipped) = _fetcher.skipRows(maxRowsLeftToSkip());
+
+  // WAITING => skipped == 0
+  TRI_ASSERT(state != ExecutionState::WAITING || skipped == 0);
+
+  _counter += skipped;
+
+  LimitStats stats{};
+  if (infos().isFullCountEnabled()) {
+    stats.incrFullCountBy(skipped);
+  }
+
+  return {state, stats};
+}
+
+std::pair<ExecutionState, LimitStats> LimitExecutor::skipRestForFullCount() {
+  ExecutionState state;
+  size_t skipped;
+  LimitStats stats{};
+  // skip ALL the rows
+  std::tie(state, skipped) = _fetcher.skipRows(ExecutionBlock::SkipAllSize());
+
+  if (state == ExecutionState::WAITING) {
+    TRI_ASSERT(skipped == 0);
+    return {state, stats};
+  }
+
+  // We must not update _counter here. It is only used to count until offset+limit
+  // is reached.
+
+  if (infos().isFullCountEnabled()) {
+    stats.incrFullCountBy(skipped);
+  }
+
+  return {state, stats};
+}
+
 std::pair<ExecutionState, LimitStats> LimitExecutor::produceRows(OutputAqlItemRow& output) {
   TRI_IF_FAILURE("LimitExecutor::produceRows") {
     THROW_ARANGO_EXCEPTION(TRI_ERROR_DEBUG);
   }
-  LimitStats stats{};
   InputAqlItemRow input{CreateInvalidInputRowHint{}};
 
   ExecutionState state;
-  LimitState limitState;
+  LimitStats stats{};
 
   while (LimitState::SKIPPING == currentState()) {
-    size_t skipped;
+    LimitStats tmpStats;
-    std::tie(state, skipped) = _fetcher.skipRows(maxRowsLeftToSkip());
+    std::tie(state, tmpStats) = skipOffset();
-
+    stats += tmpStats;
-    if (state == ExecutionState::WAITING) {
+    if (state == ExecutionState::WAITING || state == ExecutionState::DONE) {
-      return {state, stats};
-    }
-
-    _counter += skipped;
-
-    if (infos().isFullCountEnabled()) {
-      stats.incrFullCountBy(skipped);
-    }
-
-    // Abort if upstream is done
-    if (state == ExecutionState::DONE) {
       return {state, stats};
     }
   }
 
-  while (LimitState::LIMIT_REACHED != (limitState = currentState()) && LimitState::COUNTING != limitState) {
+  while (LimitState::RETURNING == currentState()) {
     std::tie(state, input) = _fetcher.fetchRow(maxRowsLeftToFetch());
 
     if (state == ExecutionState::WAITING) {
       return {state, stats};
     }
 
-    if (!input) {
+    // This executor is pass-through. Thus we will never get asked to write an
-      TRI_ASSERT(state == ExecutionState::DONE);
+    // output row for which there is no input, as in- and output rows have a
-      return {state, stats};
+    // 1:1 correspondence.
-    }
     TRI_ASSERT(input.isInitialized());
 
     // We've got one input row
@@ -107,113 +137,152 @@ std::pair<ExecutionState, LimitStats> LimitExecutor::produceRows(OutputAqlItemRo
     }
 
     // Return one row
-    if (limitState == LimitState::RETURNING) {
+    output.copyRow(input);
-      output.copyRow(input);
+    return {state, stats};
-      return {state, stats};
-    }
-    if (limitState == LimitState::RETURNING_LAST_ROW) {
-      output.copyRow(input);
-      return {ExecutionState::DONE, stats};
-    }
-
-    // Abort if upstream is done
-    if (state == ExecutionState::DONE) {
-      return {state, stats};
-    }
-
-    TRI_ASSERT(false);
   }
 
-  while (LimitState::LIMIT_REACHED != currentState()) {
+  // This case is special for two reasons.
-    size_t skipped;
+  // First, after this we want to return DONE, regardless of the upstream's
-    // TODO: skip ALL the rows
+  // state.
-    std::tie(state, skipped) = _fetcher.skipRows(ExecutionBlock::DefaultBatchSize());
+  // Second, when fullCount is enabled, we need to get the fullCount before
+  // returning the last row, as the count is returned with the stats (and we
+  // would not be asked again by ExecutionBlockImpl in any case).
+  if (LimitState::RETURNING_LAST_ROW == currentState()) {
+    if (_lastRowToOutput.isInitialized()) {
+      // Use previously saved row iff there is one. We can get here only if
+      // fullCount is enabled. If it is, we can get here multiple times (until
+      // we consumed the whole upstream, which might return WAITING repeatedly).
+      TRI_ASSERT(infos().isFullCountEnabled());
+      state = _stateOfLastRowToOutput;
+      TRI_ASSERT(state != ExecutionState::WAITING);
+      input = std::move(_lastRowToOutput);
+      TRI_ASSERT(!_lastRowToOutput.isInitialized()); // rely on the move
+    } else {
+      std::tie(state, input) = _fetcher.fetchRow(maxRowsLeftToFetch());
 
-    if (state == ExecutionState::WAITING) {
+      if (state == ExecutionState::WAITING) {
-      return {state, stats};
+        return {state, stats};
+      }
     }
 
-    _counter += skipped;
+    // This executor is pass-through. Thus we will never get asked to write an
+    // output row for which there is no input, as in- and output rows have a
+    // 1:1 correspondence.
+    TRI_ASSERT(input.isInitialized());
 
     if (infos().isFullCountEnabled()) {
-      stats.incrFullCountBy(skipped);
+      // Save the state now. The _stateOfLastRowToOutput will not be used unless
+      // _lastRowToOutput gets set.
+      _stateOfLastRowToOutput = state;
+      LimitStats tmpStats;
+      std::tie(state, tmpStats) = skipRestForFullCount();
+      stats += tmpStats;
+      if (state == ExecutionState::WAITING) {
+        // Save the row
+        _lastRowToOutput = std::move(input);
+        return {state, stats};
+      }
     }
 
-    // Abort if upstream is done
+    // It's important to increase the counter for the last row only *after*
-    if (state == ExecutionState::DONE) {
+    // skipRestForFullCount() is done, because we need currentState() to stay
-      return {state, stats};
+    // at RETURNING_LAST_ROW until we've actually returned the last row.
+    _counter++;
+    if (infos().isFullCountEnabled()) {
+      stats.incrFullCount();
     }
+
+    output.copyRow(input);
+    return {ExecutionState::DONE, stats};
   }
 
+  // We should never be COUNTING, this must already be done in the
+  // RETURNING_LAST_ROW-handler.
+  TRI_ASSERT(LimitState::LIMIT_REACHED == currentState());
   // When fullCount is enabled, the loop may only abort when upstream is done.
   TRI_ASSERT(!infos().isFullCountEnabled());
 
   return {ExecutionState::DONE, stats};
 }
 
-std::pair<ExecutionState, size_t> LimitExecutor::expectedNumberOfRows(size_t atMost) const {
+std::tuple<ExecutionState, LimitStats, SharedAqlItemBlockPtr> LimitExecutor::fetchBlockForPassthrough(size_t atMost) {
   switch (currentState()) {
     case LimitState::LIMIT_REACHED:
       // We are done with our rows!
-      return {ExecutionState::DONE, 0};
+      return {ExecutionState::DONE, LimitStats{}, nullptr};
-    case LimitState::COUNTING:
+    case LimitState::COUNTING: {
-      // We are actually done with our rows,
+      LimitStats stats{};
-      // BUt we need to make sure that we get asked more
+      while (LimitState::LIMIT_REACHED != currentState()) {
-      return {ExecutionState::DONE, 1};
+        ExecutionState state;
+        LimitStats tmpStats{};
+        std::tie(state, tmpStats) = skipRestForFullCount();
+        stats += tmpStats;
+
+        if (state == ExecutionState::WAITING || state == ExecutionState::DONE) {
+          return {state, stats, nullptr};
+        }
+      }
+      return {ExecutionState::DONE, stats, nullptr};
+    }
     case LimitState::SKIPPING: {
-      // This is the best guess we can make without calling
+      LimitStats stats{};
-      // preFetchNumberOfRows(), which, however, would prevent skipping.
+      while (LimitState::SKIPPING == currentState()) {
-      // The problem is not here, but in ExecutionBlockImpl which calls this to
+        ExecutionState state;
-      // allocate a block before we had a chance to skip here.
+        LimitStats tmpStats{};
-      // There is a corresponding todo note on
+        std::tie(state, tmpStats) = skipOffset();
-      // LimitExecutor::Properties::inputSizeRestrictsOutputSize.
+        stats += tmpStats;
-
+        if (state == ExecutionState::WAITING || state == ExecutionState::DONE) {
-      TRI_ASSERT(_counter < infos().getOffset());
+          return {state, stats, nullptr};
-
-      // Note on fullCount we might get more lines from upstream then required.
-      size_t leftOverIncludingSkip = infos().getLimitPlusOffset() - _counter;
-      size_t leftOver = infos().getLimit();
-      if (_infos.isFullCountEnabled()) {
-        // Add one for the fullcount.
-        if (leftOverIncludingSkip < atMost) {
-          leftOverIncludingSkip++;
-        }
-        if (leftOver < atMost) {
-          leftOver++;
         }
       }
 
-      ExecutionState const state =
+      // We should have reached the next state now
-          leftOverIncludingSkip > 0 ? ExecutionState::HASMORE : ExecutionState::DONE;
+      TRI_ASSERT(currentState() != LimitState::SKIPPING);
-
+      // Now jump to the correct case
-      if (state != ExecutionState::DONE) {
+      auto rv = fetchBlockForPassthrough(atMost);
-        // unless we're DONE, never return 0.
+      // Add the stats we collected to the return value
-        leftOver = (std::max)(std::size_t{1}, leftOver);
+      std::get<LimitStats>(rv) += stats;
-      }
+      return rv;
-
-      return {state, leftOver};
     }
     case LimitState::RETURNING_LAST_ROW:
-    case LimitState::RETURNING: {
+    case LimitState::RETURNING:
-      auto res = _fetcher.preFetchNumberOfRows(maxRowsLeftToFetch());
+      auto rv =_fetcher.fetchBlockForPassthrough(std::min(atMost, maxRowsLeftToFetch()));
-      if (res.first == ExecutionState::WAITING) {
+      return {rv.first, LimitStats{}, std::move(rv.second)};
-        return res;
-      }
-      // Note on fullCount we might get more lines from upstream then required.
-      size_t leftOver = (std::min)(infos().getLimitPlusOffset() - _counter, res.second);
-      if (_infos.isFullCountEnabled() && leftOver < atMost) {
-        // Add one for the fullcount.
-        leftOver++;
-      }
-      if (leftOver > 0) {
-        return {ExecutionState::HASMORE, leftOver};
-      }
-      return {ExecutionState::DONE, 0};
-    }
   }
-  TRI_ASSERT(false);
+}
-  // This should not be reached, (the switch case is covering all enum values)
+
-  // Nevertheless if it is reached this will fall back to the non.optimal, but
+std::tuple<ExecutionState, LimitExecutor::Stats, size_t> LimitExecutor::skipRows(size_t const toSkipRequested) {
-  // working variant
+  // fullCount can only be enabled on the last top-level LIMIT block. Thus
-  return {ExecutionState::DONE, atMost};
+  // skip cannot be called on it! If this requirement is changed for some
+  // reason, the current implementation will not work.
+  TRI_ASSERT(!infos().isFullCountEnabled());
+
+  // If we're still skipping ourselves up to offset, this needs to be done first.
+  size_t const toSkipOffset =
+      currentState() == LimitState::SKIPPING ? maxRowsLeftToSkip() : 0;
+
+  // We have to skip
+  //   our offset (toSkipOffset or maxRowsLeftToSkip()),
+  //   plus what we were requested to skip (toSkipRequested),
+  //   but not more than our total limit (maxRowsLeftToFetch()).
+  size_t const toSkipTotal =
+      std::min(toSkipRequested + toSkipOffset, maxRowsLeftToFetch());
+
+  ExecutionState state;
+  size_t skipped;
+  std::tie(state, skipped) = _fetcher.skipRows(toSkipTotal);
+
+  // WAITING => skipped == 0
+  TRI_ASSERT(state != ExecutionState::WAITING || skipped == 0);
+
+  _counter += skipped;
+
+  // Do NOT report the rows we skipped up to the offset, they don't count.
+  size_t const reportSkipped = toSkipOffset >= skipped ? 0 : skipped - toSkipOffset;
+
+  if (currentState() == LimitState::LIMIT_REACHED) {
+    state = ExecutionState::DONE;
+  }
+
+  return std::make_tuple(state, LimitStats{}, reportSkipped);
 }

arangod/Aql/LimitExecutor.h

@@ -34,6 +34,7 @@
 #include "Aql/OutputAqlItemRow.h"
 #include "Aql/types.h"
 
+#include <iosfwd>
 #include <memory>
 
 namespace arangodb {
@@ -80,24 +81,8 @@ class LimitExecutor {
  public:
   struct Properties {
     static const bool preservesOrder = true;
-    // TODO Maybe we can and want to allow passthrough. For this it would be
+    static const bool allowsBlockPassthrough = true;
-    //  necessary to allow the LimitExecutor to skip before ExecutionBlockImpl
+    static const bool inputSizeRestrictsOutputSize = false;
-    //  prefetches a block. This is related to the comment on
-    //  inputSizeRestrictsOutputSize.
-    static const bool allowsBlockPassthrough = false;
-    //TODO:
-    // The implementation of this is currently suboptimal for the LimitExecutor.
-    // ExecutionBlockImpl allocates a block before calling produceRows();
-    // that means before LimitExecutor had a chance to skip;
-    // that means we cannot yet call expectedNumberOfRows() on the Fetcher,
-    // because it would call getSome on the parent when we actually want to
-    // skip.
-    // One possible solution is to call skipSome during expectedNumberOfRows(),
-    // which is more than a little ugly. Perhaps we can find a better way.
-    // Note that there are corresponding comments in
-    // ExecutionBlockImpl::requestWrappedBlock() and
-    // LimitExecutor::expectedNumberOfRows().
-    static const bool inputSizeRestrictsOutputSize = true;
   };
   using Fetcher = SingleRowFetcher<Properties::allowsBlockPassthrough>;
   using Infos = LimitExecutorInfos;
@@ -116,16 +101,30 @@ class LimitExecutor {
    */
   std::pair<ExecutionState, Stats> produceRows(OutputAqlItemRow& output);
 
-  std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t atMost) const;
+  /**
-  
+   * @brief Custom skipRows() implementation. This is obligatory to increase
+   * _counter!
+   *
+   * Semantically, we first skip until our local offset. We may not report the
+   * number of rows skipped this way. Second, we skip up to the number of rows
+   * requested; but at most up to our limit.
+   */
+  std::tuple<ExecutionState, Stats, size_t> skipRows(size_t toSkipRequested);
+
+  std::tuple<ExecutionState, LimitStats, SharedAqlItemBlockPtr> fetchBlockForPassthrough(size_t atMost);
+
  private:
   Infos const& infos() const noexcept { return _infos; };
 
   size_t maxRowsLeftToFetch() const noexcept {
+    // counter should never exceed this count!
+    TRI_ASSERT(infos().getLimitPlusOffset() >= _counter);
     return infos().getLimitPlusOffset() - _counter;
   }
 
   size_t maxRowsLeftToSkip() const noexcept {
+    // should not be called after skipping the offset!
+    TRI_ASSERT(infos().getOffset() >= _counter);
     return infos().getOffset() - _counter;
   }
 
@@ -134,8 +133,8 @@ class LimitExecutor {
     SKIPPING,
     // state is RETURNING until the limit is reached
     RETURNING,
-    // state is RETURNING_LAST_ROW only if fullCount is disabled, and we've seen
+    // state is RETURNING_LAST_ROW if we've seen the second to last row before
-    // the second to last row until the limit is reached
+    // the limit is reached
     RETURNING_LAST_ROW,
     // state is COUNTING when the limit is reached and fullcount is enabled
     COUNTING,
@@ -156,7 +155,7 @@ class LimitExecutor {
     if (_counter < infos().getOffset()) {
       return LimitState::SKIPPING;
     }
-    if (!infos().isFullCountEnabled() && _counter + 1 == infos().getLimitPlusOffset()) {
+    if (_counter + 1 == infos().getLimitPlusOffset()) {
       return LimitState::RETURNING_LAST_ROW;
     }
     if (_counter < infos().getLimitPlusOffset()) {
@@ -169,9 +168,14 @@ class LimitExecutor {
     return LimitState::LIMIT_REACHED;
   }
 
+  std::pair<ExecutionState, Stats> skipOffset();
+  std::pair<ExecutionState, Stats> skipRestForFullCount();
+
  private:
   Infos const& _infos;
   Fetcher& _fetcher;
+  InputAqlItemRow _lastRowToOutput;
+  ExecutionState _stateOfLastRowToOutput;
   // Number of input lines seen
   size_t _counter = 0;
 };

arangod/Aql/LimitStats.h

@@ -34,6 +34,19 @@ class LimitStats {
  public:
   LimitStats() noexcept : _fullCount(0) {}
 
+  LimitStats(LimitStats const&) = default;
+  LimitStats& operator=(LimitStats const&) = default;
+
+  // It is relied upon that other._fullcount is zero after the move!
+  LimitStats(LimitStats&& other) noexcept : _fullCount(other._fullCount) {
+    other._fullCount = 0;
+  }
+  LimitStats& operator=(LimitStats&& other) noexcept {
+    _fullCount = other._fullCount;
+    other._fullCount = 0;
+    return *this;
+  };
+
   void incrFullCount() noexcept { _fullCount++; }
   void incrFullCountBy(size_t amount) noexcept { _fullCount += amount; }
 
@@ -49,6 +62,11 @@ inline ExecutionStats& operator+=(ExecutionStats& executionStats,
   return executionStats;
 }
 
+inline LimitStats& operator+=(LimitStats& limitStats, LimitStats const& other) noexcept {
+  limitStats.incrFullCountBy(other.getFullCount());
+  return limitStats;
+}
+
 }
 }
 

arangod/Aql/ModificationExecutor.h

@@ -229,17 +229,6 @@ class ModificationExecutor : public ModificationExecutorBase<FetcherType> {
    */
   std::pair<ExecutionState, Stats> produceRows(OutputAqlItemRow& output);
 
-  /**
-   * This executor immedieately  returns every actually consumed row
-   * All other rows belong to the fetcher.
-   */
-  inline std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t) const {
-    TRI_ASSERT(false);
-    THROW_ARANGO_EXCEPTION_MESSAGE(
-        TRI_ERROR_INTERNAL,
-        "Logic_error, prefetching number fo rows not supported");
-  }
-
  private:
   Modifier _modifier;
 };

arangod/Aql/ShortestPathExecutor.h

@@ -172,13 +172,6 @@ class ShortestPathExecutor {
    */
   std::pair<ExecutionState, Stats> produceRows(OutputAqlItemRow& output);
 
-  inline std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t) const {
-    TRI_ASSERT(false);
-    THROW_ARANGO_EXCEPTION_MESSAGE(
-        TRI_ERROR_INTERNAL,
-        "Logic_error, prefetching number fo rows not supported");
-  }
-
  private:
   /**
    * @brief Fetch input row(s) and compute path

arangod/Aql/SingleRemoteModificationExecutor.h

@@ -86,13 +86,6 @@ struct SingleRemoteModificationExecutor {
    */
   std::pair<ExecutionState, Stats> produceRows(OutputAqlItemRow& output);
 
-  inline std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t) const {
-    TRI_ASSERT(false);
-    THROW_ARANGO_EXCEPTION_MESSAGE(
-        TRI_ERROR_INTERNAL,
-        "Logic_error, prefetching number fo rows not supported");
-  }
-
  protected:
   bool doSingleRemoteModificationOperation(InputAqlItemRow&, OutputAqlItemRow&, Stats&);
 

arangod/Aql/SingleRowFetcher.h

@@ -88,7 +88,7 @@ class SingleRowFetcher {
   TEST_VIRTUAL std::pair<ExecutionState, size_t> skipRows(size_t atMost);
 
   // TODO enable_if<passBlocksThrough>
-  std::pair<ExecutionState, SharedAqlItemBlockPtr> fetchBlockForPassthrough(size_t atMost);
+  TEST_VIRTUAL std::pair<ExecutionState, SharedAqlItemBlockPtr> fetchBlockForPassthrough(size_t atMost);
 
   std::pair<ExecutionState, size_t> preFetchNumberOfRows(size_t atMost) {
     if (_upstreamState != ExecutionState::DONE && !indexIsValid()) {

arangod/Aql/SubqueryExecutor.h

@@ -93,12 +93,9 @@ class SubqueryExecutor {
    */
   std::pair<ExecutionState, Stats> produceRows(OutputAqlItemRow& output);
 
-  inline std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t) const {
+  inline std::tuple<ExecutionState, Stats, SharedAqlItemBlockPtr> fetchBlockForPassthrough(size_t atMost) {
-    // Passthrough does not need to implement this!
+    auto rv = _fetcher.fetchBlockForPassthrough(atMost);
-    TRI_ASSERT(false);
+    return {rv.first, {}, std::move(rv.second)};
-    THROW_ARANGO_EXCEPTION_MESSAGE(
-        TRI_ERROR_INTERNAL,
-        "Logic_error, prefetching number fo rows not supported");
   }
 
  private:

arangod/Aql/TraversalExecutor.h

@@ -137,13 +137,6 @@ class TraversalExecutor {
    */
   std::pair<ExecutionState, Stats> produceRows(OutputAqlItemRow& output);
 
-  inline std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t) const {
-    TRI_ASSERT(false);
-    THROW_ARANGO_EXCEPTION_MESSAGE(
-        TRI_ERROR_INTERNAL,
-        "Logic_error, prefetching number fo rows not supported");
-  }
-
  private:
   /**
    * @brief compute the return state

tests/Aql/AqlHelper.cpp

@@ -0,0 +1,100 @@
+////////////////////////////////////////////////////////////////////////////////
+/// DISCLAIMER
+///
+/// Copyright 2019 ArangoDB GmbH, Cologne, Germany
+///
+/// Licensed under the Apache License, Version 2.0 (the "License");
+/// you may not use this file except in compliance with the License.
+/// You may obtain a copy of the License at
+///
+///     http://www.apache.org/licenses/LICENSE-2.0
+///
+/// Unless required by applicable law or agreed to in writing, software
+/// distributed under the License is distributed on an "AS IS" BASIS,
+/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+/// See the License for the specific language governing permissions and
+/// limitations under the License.
+///
+/// Copyright holder is ArangoDB GmbH, Cologne, Germany
+///
+/// @author Tobias Gödderz
+////////////////////////////////////////////////////////////////////////////////
+
+#include "AqlHelper.h"
+
+#include "Aql/ExecutionStats.h"
+
+using namespace arangodb;
+using namespace arangodb::aql;
+
+std::ostream& arangodb::aql::operator<<(std::ostream& stream, ExecutionStats const& stats) {
+  VPackBuilder builder{};
+  stats.toVelocyPack(builder, true);
+  return stream << builder.toJson();
+}
+
+std::ostream& arangodb::aql::operator<<(std::ostream& stream, AqlItemBlock const& block) {
+  stream << "[";
+  for (size_t row = 0; row < block.size(); row++) {
+    if (row > 0) {
+      stream << ",";
+    }
+    stream << " ";
+    VPackBuilder builder{};
+    builder.openArray();
+    for (RegisterId reg = 0; reg < block.getNrRegs(); reg++) {
+      if (reg > 0) {
+        stream << ",";
+      }
+      // will not work for docvecs or ranges
+      builder.add(block.getValueReference(row, reg).slice());
+    }
+    builder.close();
+    stream << builder.toJson();
+  }
+  stream << " ]";
+  return stream;
+}
+
+bool arangodb::aql::operator==(arangodb::aql::ExecutionStats const& left,
+                               arangodb::aql::ExecutionStats const& right) {
+  TRI_ASSERT(left.nodes.empty());
+  TRI_ASSERT(right.nodes.empty());
+  TRI_ASSERT(left.executionTime == 0.0);
+  TRI_ASSERT(right.executionTime == 0.0);
+  TRI_ASSERT(left.peakMemoryUsage == 0);
+  TRI_ASSERT(right.peakMemoryUsage == 0);
+  // clang-format off
+  return left.writesExecuted == right.writesExecuted
+      && left.writesIgnored == right.writesIgnored
+      && left.scannedFull == right.scannedFull
+      && left.scannedIndex == right.scannedIndex
+      && left.filtered == right.filtered
+      && left.requests == right.requests
+      && left.fullCount == right.fullCount
+      && left.count == right.count;
+  // clang-format on
+}
+bool arangodb::aql::operator==(arangodb::aql::AqlItemBlock const& left,
+                               arangodb::aql::AqlItemBlock const& right) {
+  if (left.size() != right.size()) {
+    return false;
+  }
+  if (left.getNrRegs() != right.getNrRegs()) {
+    return false;
+  }
+  size_t const rows = left.size();
+  RegisterCount const regs = left.getNrRegs();
+  for (size_t row = 0; row < rows; row++) {
+    for (RegisterId reg = 0; reg < regs; reg++) {
+      AqlValue const& l = left.getValueReference(row, reg);
+      AqlValue const& r = right.getValueReference(row, reg);
+      // Doesn't work for docvecs or ranges
+      if (l.slice() != r.slice()) {
+        return false;
+      }
+    }
+  }
+
+  return true;
+}

tests/Aql/AqlHelper.h

@@ -0,0 +1,42 @@
+////////////////////////////////////////////////////////////////////////////////
+/// DISCLAIMER
+///
+/// Copyright 2019 ArangoDB GmbH, Cologne, Germany
+///
+/// Licensed under the Apache License, Version 2.0 (the "License");
+/// you may not use this file except in compliance with the License.
+/// You may obtain a copy of the License at
+///
+///     http://www.apache.org/licenses/LICENSE-2.0
+///
+/// Unless required by applicable law or agreed to in writing, software
+/// distributed under the License is distributed on an "AS IS" BASIS,
+/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+/// See the License for the specific language governing permissions and
+/// limitations under the License.
+///
+/// Copyright holder is ArangoDB GmbH, Cologne, Germany
+///
+/// @author Tobias Gödderz
+////////////////////////////////////////////////////////////////////////////////
+
+#ifndef TESTS_AQL_AQLHELPER_H
+#define TESTS_AQL_AQLHELPER_H
+
+#include "Aql/AqlItemBlock.h"
+#include "Aql/ExecutionState.h"
+#include "Aql/ExecutionStats.h"
+
+namespace arangodb {
+namespace aql {
+
+std::ostream& operator<<(std::ostream&, arangodb::aql::ExecutionStats const&);
+std::ostream& operator<<(std::ostream&, arangodb::aql::AqlItemBlock const&);
+
+bool operator==(arangodb::aql::ExecutionStats const&, arangodb::aql::ExecutionStats const&);
+bool operator==(arangodb::aql::AqlItemBlock const&, arangodb::aql::AqlItemBlock const&);
+
+}
+}
+
+#endif  // TESTS_AQL_AQLHELPER_H

tests/Aql/AqlItemBlockHelper.h

@@ -32,6 +32,7 @@
 #include "Aql/ResourceUsage.h"
 #include "Aql/SharedAqlItemBlockPtr.h"
 
+#include "AqlHelper.h"
 #include "VelocyPackHelper.h"
 
 /* * * * * * * *
@@ -81,10 +82,6 @@ template <::arangodb::aql::RegisterId columns>
 }  // namespace tests
 }  // namespace arangodb
 
-namespace std {
-std::ostream& operator<<(std::ostream&, ::arangodb::aql::AqlItemBlock const&);
-}
-
 namespace arangodb {
 namespace tests {
 namespace aql {
@@ -104,6 +101,9 @@ class EntryToAqlValueVisitor : public boost::static_visitor<AqlValue> {
 template <RegisterId columns>
 SharedAqlItemBlockPtr buildBlock(AqlItemBlockManager& manager,
                                  MatrixBuilder<columns>&& matrix) {
+  if (matrix.size() == 0) {
+    return nullptr;
+  }
   SharedAqlItemBlockPtr block{new AqlItemBlock(manager, matrix.size(), columns)};
 
   for (size_t row = 0; row < matrix.size(); row++) {

tests/Aql/CalculationExecutorTest.cpp

@@ -111,7 +111,7 @@ class CalculationExecutorTest : public ::testing::Test {
 TEST_F(CalculationExecutorTest, there_are_no_rows_upstream_the_producer_does_not_wait) {
   SharedAqlItemBlockPtr block{new AqlItemBlock(itemBlockManager, 1000, 2)};
   VPackBuilder input;
-  SingleRowFetcherHelper<true> fetcher(input.steal(), false);
+  SingleRowFetcherHelper<true> fetcher(itemBlockManager, input.steal(), false);
   CalculationExecutor<CalculationType::Condition> testee(fetcher, infos);
   // Use this instead of std::ignore, so the tests will be noticed and
   // updated when someone changes the stats type in the return value of
@@ -128,7 +128,7 @@ TEST_F(CalculationExecutorTest, there_are_no_rows_upstream_the_producer_does_not
 TEST_F(CalculationExecutorTest, there_are_no_rows_upstream_the_producer_waits) {
   SharedAqlItemBlockPtr block{new AqlItemBlock(itemBlockManager, 1000, 2)};
   VPackBuilder input;
-  SingleRowFetcherHelper<true> fetcher(input.steal(), true);
+  SingleRowFetcherHelper<true> fetcher(itemBlockManager, input.steal(), true);
   CalculationExecutor<CalculationType::Condition> testee(fetcher, infos);
   // Use this instead of std::ignore, so the tests will be noticed and
   // updated when someone changes the stats type in the return value of
@@ -149,7 +149,7 @@ TEST_F(CalculationExecutorTest, there_are_no_rows_upstream_the_producer_waits) {
 TEST_F(CalculationExecutorTest, there_are_rows_in_the_upstream_the_producer_does_not_wait) {
   SharedAqlItemBlockPtr block{new AqlItemBlock(itemBlockManager, 1000, 2)};
   auto input = VPackParser::fromJson("[ [0], [1], [2] ]");
-  SingleRowFetcherHelper<true> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<true> fetcher(itemBlockManager, input->steal(), false);
   CalculationExecutor<CalculationType::Condition> testee(fetcher, infos);
   NoStats stats{};
 
@@ -193,7 +193,7 @@ TEST_F(CalculationExecutorTest, there_are_rows_in_the_upstream_the_producer_does
 TEST_F(CalculationExecutorTest, there_are_rows_in_the_upstream_the_producer_waits) {
   SharedAqlItemBlockPtr block{new AqlItemBlock(itemBlockManager, 1000, 2)};
   auto input = VPackParser::fromJson("[ [0], [1], [2] ]");
-  SingleRowFetcherHelper<true> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<true> fetcher(itemBlockManager, input->steal(), true);
   CalculationExecutor<CalculationType::Condition> testee(fetcher, infos);
   NoStats stats{};
 

tests/Aql/CountCollectExecutorTest.cpp

@@ -60,7 +60,7 @@ class CountCollectExecutorTest : public ::testing::Test {
 TEST_F(CountCollectExecutorTest, there_are_no_rows_upstream_the_producer_doesnt_wait) {
   CountCollectExecutorInfos infos(1 /* outputRegId */, 1 /* nrIn */, nrOutputReg, {}, {});
   VPackBuilder input;
-  SingleRowFetcherHelper<false> fetcher(input.steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), false);
   CountCollectExecutor testee(fetcher, infos);
   NoStats stats{};
 
@@ -79,7 +79,7 @@ TEST_F(CountCollectExecutorTest, there_are_no_rows_upstream_the_producer_doesnt_
 TEST_F(CountCollectExecutorTest, there_are_now_rows_upstream_the_producer_waits) {
   CountCollectExecutorInfos infos(1 /* outputRegId */, 1 /* nrIn */, nrOutputReg, {}, {});
   VPackBuilder input;
-  SingleRowFetcherHelper<false> fetcher(input.steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), true);
   CountCollectExecutor testee(fetcher, infos);
   NoStats stats{};
 
@@ -102,7 +102,7 @@ TEST_F(CountCollectExecutorTest, there_are_now_rows_upstream_the_producer_waits)
 TEST_F(CountCollectExecutorTest, there_are_rows_in_the_upstream_the_producer_doesnt_wait) {
   CountCollectExecutorInfos infos(1 /* outputRegId */, 1 /* nrIn */, nrOutputReg, {}, {});
   auto input = VPackParser::fromJson("[ [1], [2], [3] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   CountCollectExecutor testee(fetcher, infos);
   NoStats stats{};
 
@@ -121,7 +121,7 @@ TEST_F(CountCollectExecutorTest, there_are_rows_in_the_upstream_the_producer_doe
 TEST_F(CountCollectExecutorTest, there_are_rows_in_the_upstream_the_producer_waits) {
   CountCollectExecutorInfos infos(1 /* outputRegId */, 1 /* nrIn */, nrOutputReg, {}, {});
   auto input = VPackParser::fromJson("[ [1], [2], [3] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   CountCollectExecutor testee(fetcher, infos);
   NoStats stats{};
   OutputAqlItemRow result{std::move(block), outputRegisters,

tests/Aql/DistinctCollectExecutorTest.cpp

@@ -85,7 +85,7 @@ TEST_F(DistinctCollectExecutorTest, if_no_rows_in_upstream_the_producer_doesnt_w
                                      std::move(groupRegisters), trx);
   block.reset(new AqlItemBlock(itemBlockManager, 1000, 2));
 
-  SingleRowFetcherHelper<false> fetcher(input.steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), false);
   DistinctCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -103,7 +103,7 @@ TEST_F(DistinctCollectExecutorTest, if_no_rows_in_upstream_the_producer_waits) {
                                      std::move(groupRegisters), trx);
   block.reset(new AqlItemBlock(itemBlockManager, 1000, 2));
 
-  SingleRowFetcherHelper<false> fetcher(input.steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), true);
   DistinctCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -131,7 +131,7 @@ TEST_F(DistinctCollectExecutorTest,
   block.reset(new AqlItemBlock(itemBlockManager, 1000, nrOutputRegister));
 
   auto input = VPackParser::fromJson("[ [1], [2] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   DistinctCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -175,7 +175,7 @@ TEST_F(DistinctCollectExecutorTest,
   block.reset(new AqlItemBlock(itemBlockManager, 1000, nrOutputRegister));
 
   auto input = VPackParser::fromJson("[ [1], [2] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   DistinctCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -227,7 +227,7 @@ TEST_F(DistinctCollectExecutorTest,
   block.reset(new AqlItemBlock(itemBlockManager, 1000, nrOutputRegister));
 
   auto input = VPackParser::fromJson("[ [1], [2], [3], [1], [2] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   DistinctCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -288,7 +288,7 @@ TEST_F(DistinctCollectExecutorTest,
   block.reset(new AqlItemBlock(itemBlockManager, 1000, nrOutputRegister));
 
   auto input = VPackParser::fromJson("[ [1], [2], [3], [1], [2] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   DistinctCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),

tests/Aql/EnumerateCollectionExecutorTest.cpp

@@ -139,7 +139,7 @@ class EnumerateCollectionExecutorTestNoRowsUpstream : public ::testing::Test {
 };
 
 TEST_F(EnumerateCollectionExecutorTestNoRowsUpstream, the_producer_does_not_wait) {
-  SingleRowFetcherHelper<false> fetcher(input.steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), false);
   EnumerateCollectionExecutor testee(fetcher, infos);
   // Use this instead of std::ignore, so the tests will be noticed and
   // updated when someone changes the stats type in the return value of
@@ -154,7 +154,7 @@ TEST_F(EnumerateCollectionExecutorTestNoRowsUpstream, the_producer_does_not_wait
 }
 
 TEST_F(EnumerateCollectionExecutorTestNoRowsUpstream, the_producer_waits) {
-  SingleRowFetcherHelper<false> fetcher(input.steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), true);
   EnumerateCollectionExecutor testee(fetcher, infos);
   // Use this instead of std::ignore, so the tests will be noticed and
   // updated when someone changes the stats type in the return value of

tests/Aql/EnumerateListExecutorTest.cpp

@@ -62,7 +62,7 @@ TEST_F(EnumerateListExecutorTest, there_are_no_rows_upstream_the_producer_does_n
   SharedAqlItemBlockPtr block{new AqlItemBlock(itemBlockManager, 1000, 2)};
   VPackBuilder input;
 
-  SingleRowFetcherHelper<false> fetcher(input.steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), false);
   EnumerateListExecutor testee(fetcher, infos);
   // Use this instead of std::ignore, so the tests will be noticed and
   // updated when someone changes the stats type in the return value of
@@ -80,7 +80,7 @@ TEST_F(EnumerateListExecutorTest, there_are_no_rows_upstream_the_producer_waits)
   SharedAqlItemBlockPtr block{new AqlItemBlock(itemBlockManager, 1000, 2)};
   VPackBuilder input;
 
-  SingleRowFetcherHelper<false> fetcher(input.steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), true);
   EnumerateListExecutor testee(fetcher, infos);
   // Use this instead of std::ignore, so the tests will be noticed and
   // updated when someone changes the stats type in the return value of
@@ -103,7 +103,7 @@ TEST_F(EnumerateListExecutorTest, there_is_one_row_in_the_upstream_the_producer_
   SharedAqlItemBlockPtr block{new AqlItemBlock(itemBlockManager, 1000, 5)};
   auto input = VPackParser::fromJson("[ [1, 2, 3, [true, true, true]] ]");
 
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   EnumerateListExecutor testee(fetcher, infos);
   // Use this instead of std::ignore, so the tests will be noticed and
   // updated when someone changes the stats type in the return value of
@@ -188,7 +188,7 @@ TEST_F(EnumerateListExecutorTest, there_is_one_empty_array_row_in_the_upstream_t
   SharedAqlItemBlockPtr block{new AqlItemBlock(itemBlockManager, 1000, 5)};
   auto input = VPackParser::fromJson("[ [1, 2, 3, [] ] ]");
 
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   EnumerateListExecutor testee(fetcher, infos);
   // Use this instead of std::ignore, so the tests will be noticed and
   // updated when someone changes the stats type in the return value of
@@ -216,7 +216,7 @@ TEST_F(EnumerateListExecutorTest, there_are_rows_in_the_upstream_the_producer_wa
   auto input = VPackParser::fromJson(
       "[ [1, 2, 3, [true, true, true]], [1, 2, 3, [true, true, true]] ]");
 
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   EnumerateListExecutor testee(fetcher, infos);
   // Use this instead of std::ignore, so the tests will be noticed and
   // updated when someone changes the stats type in the return value of

tests/Aql/ExecutorTestHelper.cpp

@@ -1,7 +1,7 @@
 ////////////////////////////////////////////////////////////////////////////////
 /// DISCLAIMER
 ///
-/// Copyright 2018 ArangoDB GmbH, Cologne, Germany
+/// Copyright 2019 ArangoDB GmbH, Cologne, Germany
 ///
 /// Licensed under the Apache License, Version 2.0 (the "License");
 /// you may not use this file except in compliance with the License.
@@ -20,19 +20,20 @@
 /// @author Tobias Gödderz
 ////////////////////////////////////////////////////////////////////////////////
 
-#include "AqlItemBlockHelper.h"
+#include "ExecutorTestHelper.h"
 
-std::ostream& std::operator<<(
+std::ostream& arangodb::tests::aql::operator<<(std::ostream& stream,
-    std::ostream& out, ::arangodb::aql::AqlItemBlock const& block) {
+                                               arangodb::tests::aql::ExecutorCall call) {
-  for (size_t i = 0; i < block.size(); i++) {
+  return stream << [call]() {
-    for (arangodb::aql::RegisterCount j = 0; j < block.getNrRegs(); j++) {
+    switch (call) {
-      out << block.getValue(i, j).slice().toJson();
+      case ExecutorCall::SKIP_ROWS:
-      if (j + 1 != block.getNrRegs()) out << ", ";
+        return "SKIP_ROWS";
+      case ExecutorCall::PRODUCE_ROWS:
+        return "PRODUCE_ROWS";
+      case ExecutorCall::FETCH_FOR_PASSTHROUGH:
+        return "FETCH_FOR_PASSTHROUGH";
+      case ExecutorCall::EXPECTED_NR_ROWS:
+        return "EXPECTED_NR_ROWS";
     }
-    if (i + 1 != block.size()) out << std::endl;
+  }();
-  }
-
-  out << std::endl;
-
-  return out;
 }

tests/Aql/ExecutorTestHelper.h

@@ -0,0 +1,174 @@
+////////////////////////////////////////////////////////////////////////////////
+/// DISCLAIMER
+///
+/// Copyright 2019 ArangoDB GmbH, Cologne, Germany
+///
+/// Licensed under the Apache License, Version 2.0 (the "License");
+/// you may not use this file except in compliance with the License.
+/// You may obtain a copy of the License at
+///
+///     http://www.apache.org/licenses/LICENSE-2.0
+///
+/// Unless required by applicable law or agreed to in writing, software
+/// distributed under the License is distributed on an "AS IS" BASIS,
+/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+/// See the License for the specific language governing permissions and
+/// limitations under the License.
+///
+/// Copyright holder is ArangoDB GmbH, Cologne, Germany
+///
+/// @author Tobias Gödderz
+////////////////////////////////////////////////////////////////////////////////
+
+#ifndef TESTS_AQL_EXECUTORTESTHELPER_H
+#define TESTS_AQL_EXECUTORTESTHELPER_H
+
+#include "gtest/gtest.h"
+
+#include "Aql/ExecutionBlock.h"
+#include "Aql/ExecutionState.h"
+#include "Aql/ExecutionStats.h"
+#include "Aql/OutputAqlItemRow.h"
+#include "Aql/SharedAqlItemBlockPtr.h"
+
+#include <tuple>
+
+namespace arangodb {
+namespace tests {
+namespace aql {
+
+enum class ExecutorCall {
+  SKIP_ROWS,
+  PRODUCE_ROWS,
+  FETCH_FOR_PASSTHROUGH,
+  EXPECTED_NR_ROWS,
+};
+
+std::ostream& operator<<(std::ostream& stream, ExecutorCall call);
+
+using ExecutorStepResult = std::tuple<ExecutorCall, arangodb::aql::ExecutionState, size_t>;
+
+// TODO Add skipRows by passing 3 additional integers i, j, k, saying we should
+//  - skip i rows
+//  - produce j rows
+//  - skip k rows
+// TODO Make the calls to skipRows, fetchBlockForPassthrough and (later) expectedNumberOfRows
+//  somehow optional. e.g. call a templated function or so.
+// TODO Add calls to expectedNumberOfRows
+
+template <class Executor>
+std::tuple<arangodb::aql::SharedAqlItemBlockPtr, std::vector<ExecutorStepResult>, arangodb::aql::ExecutionStats>
+runExecutor(arangodb::aql::AqlItemBlockManager& manager, Executor& executor,
+            arangodb::aql::OutputAqlItemRow& outputRow, size_t const numSkip,
+            size_t const numProduce, bool const skipRest) {
+  using namespace arangodb::aql;
+  ExecutionState state = ExecutionState::HASMORE;
+  std::vector<ExecutorStepResult> results{};
+  ExecutionStats stats{};
+
+  uint64_t rowsLeft = 0;
+  size_t skippedTotal = 0;
+  size_t producedTotal = 0;
+
+  enum class RunState {
+    SKIP_OFFSET,
+    FETCH_FOR_PASSTHROUGH,
+    PRODUCE,
+    SKIP_REST,
+    BREAK
+  };
+
+  while (state != ExecutionState::DONE) {
+    RunState const runState = [&]() {
+      if (skippedTotal < numSkip) {
+        return RunState::SKIP_OFFSET;
+      }
+      if (rowsLeft == 0 && (producedTotal < numProduce || numProduce == 0)) {
+        return RunState::FETCH_FOR_PASSTHROUGH;
+      }
+      if (producedTotal < numProduce || !skipRest) {
+        return RunState::PRODUCE;
+      }
+      if (skipRest) {
+        return RunState::SKIP_REST;
+      }
+      return RunState::BREAK;
+    }();
+
+    switch (runState) {
+      // Skip first
+      // TODO don't do this for executors which don't have skipRows
+      case RunState::SKIP_OFFSET: {
+        size_t skipped;
+        typename Executor::Stats executorStats{};
+        std::tie(state, executorStats, skipped) = executor.skipRows(numSkip);
+        results.emplace_back(std::make_tuple(ExecutorCall::SKIP_ROWS, state, skipped));
+        stats += executorStats;
+        skippedTotal += skipped;
+      } break;
+      // Get a new block for pass-through if we still need to produce rows and
+      // the current (imagined, via rowsLeft) block is "empty".
+      // TODO: Don't do this at all for non-passThrough blocks
+      case RunState::FETCH_FOR_PASSTHROUGH: {
+        ExecutionState fetchBlockState;
+        typename Executor::Stats executorStats{};
+        SharedAqlItemBlockPtr block{};
+        std::tie(fetchBlockState, executorStats, block) =
+            executor.fetchBlockForPassthrough(1000);
+        size_t const blockSize = block != nullptr ? block->size() : 0;
+        results.emplace_back(std::make_tuple(ExecutorCall::FETCH_FOR_PASSTHROUGH,
+                                             fetchBlockState, blockSize));
+        stats += executorStats;
+        rowsLeft = blockSize;
+        if (fetchBlockState != ExecutionState::WAITING &&
+            fetchBlockState != ExecutionState::DONE) {
+          EXPECT_GT(rowsLeft, 0);
+        }
+        if (fetchBlockState != ExecutionState::WAITING && block == nullptr) {
+          EXPECT_EQ(ExecutionState::DONE, fetchBlockState);
+          // Abort
+          state = ExecutionState::DONE;
+        }
+      } break;
+      // Produce rows
+      case RunState::PRODUCE: {
+        EXPECT_GT(rowsLeft, 0);
+        typename Executor::Stats executorStats{};
+        size_t const rowsBefore = outputRow.numRowsWritten();
+        std::tie(state, executorStats) = executor.produceRows(outputRow);
+        size_t const rowsAfter = outputRow.numRowsWritten();
+        size_t const rowsProduced = rowsAfter - rowsBefore;
+        results.emplace_back(std::make_tuple(ExecutorCall::PRODUCE_ROWS, state, rowsProduced));
+        stats += executorStats;
+        EXPECT_LE(rowsProduced, rowsLeft);
+        rowsLeft -= rowsProduced;
+        producedTotal += rowsProduced;
+
+        if (outputRow.produced()) {
+          outputRow.advanceRow();
+        }
+      } break;
+      // TODO don't do this for executors which don't have skipRows
+      case RunState::SKIP_REST: {
+        size_t skipped;
+        typename Executor::Stats executorStats{};
+        std::tie(state, executorStats, skipped) =
+            executor.skipRows(ExecutionBlock::SkipAllSize());
+        results.emplace_back(std::make_tuple(ExecutorCall::SKIP_ROWS, state, skipped));
+        stats += executorStats;
+      } break;
+      // We're done
+      case RunState::BREAK: {
+        state = ExecutionState::DONE;
+      } break;
+    }
+  }
+
+  return {outputRow.stealBlock(), results, stats};
+}
+
+}  // namespace aql
+}  // namespace tests
+}  // namespace arangodb
+
+#endif  // TESTS_AQL_EXECUTORTESTHELPER_H

tests/Aql/FilterExecutorTest.cpp

@@ -63,7 +63,7 @@ class FilterExecutorTest : public ::testing::Test {
 
 TEST_F(FilterExecutorTest, there_are_no_rows_upstream_the_producer_does_not_wait) {
   VPackBuilder input;
-  SingleRowFetcherHelper<false> fetcher(input.steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), false);
   FilterExecutor testee(fetcher, infos);
   FilterStats stats{};
 
@@ -76,7 +76,7 @@ TEST_F(FilterExecutorTest, there_are_no_rows_upstream_the_producer_does_not_wait
 
 TEST_F(FilterExecutorTest, there_are_no_rows_upstream_the_producer_waits) {
   VPackBuilder input;
-  SingleRowFetcherHelper<false> fetcher(input.steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), true);
   FilterExecutor testee(fetcher, infos);
   FilterStats stats{};
 
@@ -96,7 +96,7 @@ TEST_F(FilterExecutorTest, there_are_no_rows_upstream_the_producer_waits) {
 TEST_F(FilterExecutorTest, there_are_rows_in_the_upstream_the_producer_does_not_wait) {
   auto input = VPackParser::fromJson(
       "[ [true], [false], [true], [false], [false], [true] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   FilterExecutor testee(fetcher, infos);
   FilterStats stats{};
 
@@ -133,7 +133,7 @@ TEST_F(FilterExecutorTest, there_are_rows_in_the_upstream_the_producer_does_not_
 TEST_F(FilterExecutorTest, there_are_rows_in_the_upstream_the_producer_waits) {
   auto input = VPackParser::fromJson(
       "[ [true], [false], [true], [false], [false], [true] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   FilterExecutor testee(fetcher, infos);
   FilterStats stats{};
 
@@ -215,7 +215,7 @@ TEST_F(FilterExecutorTest,
        there_are_rows_in_the_upstream_and_the_last_one_has_to_be_filtered_the_producer_does_not_wait) {
   auto input = VPackParser::fromJson(
       "[ [true], [false], [true], [false], [false], [true], [false] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   FilterExecutor testee(fetcher, infos);
   FilterStats stats{};
 
@@ -258,7 +258,7 @@ TEST_F(FilterExecutorTest,
        there_are_rows_in_the_upstream_and_the_last_one_has_to_be_filtered_the_producer_waits) {
   auto input = VPackParser::fromJson(
       "[ [true], [false], [true], [false], [false], [true], [false] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   FilterExecutor testee(fetcher, infos);
   FilterStats stats{};
 

tests/Aql/HashedCollectExecutorTest.cpp

@@ -97,7 +97,7 @@ class HashedCollectExecutorTestNoRows : public ::testing::Test {
 };
 
 TEST_F(HashedCollectExecutorTestNoRows, the_producer_doesnt_wait) {
-  SingleRowFetcherHelper<false> fetcher(input.steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), false);
   HashedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -108,7 +108,7 @@ TEST_F(HashedCollectExecutorTestNoRows, the_producer_doesnt_wait) {
 }
 
 TEST_F(HashedCollectExecutorTestNoRows, the_producer_waits) {
-  SingleRowFetcherHelper<false> fetcher(input.steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), true);
   HashedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -172,7 +172,7 @@ class HashedCollectExecutorTestRowsNoCount : public ::testing::Test {
 
 TEST_F(HashedCollectExecutorTestRowsNoCount, the_producer_doesnt_wait_1) {
   auto input = VPackParser::fromJson("[ [1], [2] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   HashedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -212,7 +212,7 @@ TEST_F(HashedCollectExecutorTestRowsNoCount, the_producer_doesnt_wait_1) {
 
 TEST_F(HashedCollectExecutorTestRowsNoCount, the_producer_doesnt_wait_2) {
   auto input = VPackParser::fromJson("[ [1], [2], [3] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   HashedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -262,7 +262,7 @@ TEST_F(HashedCollectExecutorTestRowsNoCount, the_producer_doesnt_wait_2) {
 
 TEST_F(HashedCollectExecutorTestRowsNoCount, the_producer_doesnt_wait_3) {
   auto input = VPackParser::fromJson("[ [1], [2], [3], [1], [2] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   HashedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -312,7 +312,7 @@ TEST_F(HashedCollectExecutorTestRowsNoCount, the_producer_doesnt_wait_3) {
 
 TEST_F(HashedCollectExecutorTestRowsNoCount, the_producer_doesnt_wait_4) {
   auto input = VPackParser::fromJson("[ [1], [2], [1], [2] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   HashedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -352,7 +352,7 @@ TEST_F(HashedCollectExecutorTestRowsNoCount, the_producer_doesnt_wait_4) {
 
 TEST_F(HashedCollectExecutorTestRowsNoCount, the_producer_waits) {
   auto input = VPackParser::fromJson("[ [1], [2] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   HashedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -442,7 +442,7 @@ TEST(HashedCollectExecutorTestRowsCount, the_producer_doesnt_wait) {
   NoStats stats{};
 
   auto input = VPackParser::fromJson("[ [1], [2] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   HashedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -539,7 +539,7 @@ TEST(HashedCollectExecutorTestRowsCountNumbers, the_producer_doesnt_wait) {
   NoStats stats{};
 
   auto input = VPackParser::fromJson("[ [1], [2], [3] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   HashedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -652,7 +652,7 @@ TEST(HashedCollectExecutorTestRowsCountStrings, the_producer_doesnt_wait) {
   NoStats stats{};
 
   auto input = VPackParser::fromJson("[ [\"a\"], [\"aa\"], [\"aaa\"] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   HashedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),

tests/Aql/NoResultsExecutorTest.cpp

@@ -70,7 +70,7 @@ class NoResultsExecutorTest : public ::testing::Test {
 
 TEST_F(NoResultsExecutorTest, no_rows_upstream_the_producer_doesnt_wait) {
   VPackBuilder input;
-  SingleRowFetcherHelper<false> fetcher(input.steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), false);
   NoResultsExecutor testee(fetcher, infos);
   NoStats stats{};
 
@@ -82,7 +82,7 @@ TEST_F(NoResultsExecutorTest, no_rows_upstream_the_producer_doesnt_wait) {
 
 TEST_F(NoResultsExecutorTest, no_rows_upstream_the_producer_waits) {
   VPackBuilder input;
-  SingleRowFetcherHelper<false> fetcher(input.steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), true);
   NoResultsExecutor testee(fetcher, infos);
   NoStats stats{};
 
@@ -99,7 +99,7 @@ TEST_F(NoResultsExecutorTest, no_rows_upstream_the_producer_waits) {
 
 TEST_F(NoResultsExecutorTest, rows_upstream_the_producer_doesnt_wait) {
   auto input = VPackParser::fromJson("[ [true], [false], [true] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   NoResultsExecutor testee(fetcher, infos);
   NoStats stats{};
 
@@ -116,7 +116,7 @@ TEST_F(NoResultsExecutorTest, rows_upstream_the_producer_doesnt_wait) {
 
 TEST_F(NoResultsExecutorTest, rows_upstream_the_producer_waits) {
   auto input = VPackParser::fromJson("[ [true], [false], [true] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   NoResultsExecutor testee(fetcher, infos);
   NoStats stats{};
 

tests/Aql/ReturnExecutorTest.cpp

@@ -61,7 +61,7 @@ TEST_F(ReturnExecutorTest, NoRowsUpstreamProducerDoesNotWait) {
   ReturnExecutorInfos infos(inputRegister, 1 /*nr in*/, 1 /*nr out*/, true /*do count*/);
   auto& outputRegisters = infos.getOutputRegisters();
   VPackBuilder input;
-  SingleRowFetcherHelper<false> fetcher(input.steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), false);
   ReturnExecutor testee(fetcher, infos);
   CountStats stats{};
 
@@ -76,7 +76,7 @@ TEST_F(ReturnExecutorTest, NoRowsUpstreamProducerWaits) {
   ReturnExecutorInfos infos(inputRegister, 1 /*nr in*/, 1 /*nr out*/, true /*do count*/);
   auto& outputRegisters = infos.getOutputRegisters();
   VPackBuilder input;
-  SingleRowFetcherHelper<false> fetcher(input.steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), true);
   ReturnExecutor testee(fetcher, infos);
   CountStats stats{};
 
@@ -95,7 +95,7 @@ TEST_F(ReturnExecutorTest, RowsUpstreamProducerDoesNotWait) {
   ReturnExecutorInfos infos(inputRegister, 1 /*nr in*/, 1 /*nr out*/, true /*do count*/);
   auto& outputRegisters = infos.getOutputRegisters();
   auto input = VPackParser::fromJson("[ [true], [false], [true] ]");
-  SingleRowFetcherHelper<false> fetcher(input->buffer(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->buffer(), false);
   ReturnExecutor testee(fetcher, infos);
   CountStats stats{};
 
@@ -135,7 +135,7 @@ TEST_F(ReturnExecutorTest, RowsUpstreamProducerWaits) {
   ReturnExecutorInfos infos(inputRegister, 1 /*nr in*/, 1 /*nr out*/, true /*do count*/);
   auto& outputRegisters = infos.getOutputRegisters();
   auto input = VPackParser::fromJson("[ [true], [false], [true] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   ReturnExecutor testee(fetcher, infos);
   CountStats stats{};
 

tests/Aql/RowFetcherHelper.cpp

@@ -24,6 +24,7 @@
 ////////////////////////////////////////////////////////////////////////////////
 
 #include "RowFetcherHelper.h"
+#include "VelocyPackHelper.h"
 
 #include "Aql/AllRowsFetcher.h"
 #include "Aql/AqlItemBlock.h"
@@ -44,23 +45,6 @@ using namespace arangodb::tests::aql;
 using namespace arangodb::aql;
 
 namespace {
-void VPackToAqlItemBlock(VPackSlice data, uint64_t nrRegs, AqlItemBlock& block) {
-  // coordinates in the matrix rowNr, entryNr
-  size_t rowIndex = 0;
-  RegisterId entry = 0;
-  for (auto const& row : VPackArrayIterator(data)) {
-    // Walk through the rows
-    TRI_ASSERT(row.isArray());
-    TRI_ASSERT(row.length() == nrRegs);
-    for (auto const& oneEntry : VPackArrayIterator(row)) {
-      // Walk through on row values
-      block.setValue(rowIndex, entry, AqlValue{oneEntry});
-      entry++;
-    }
-    rowIndex++;
-    entry = 0;
-  }
-}
 }  // namespace
 
 // -----------------------------------------
@@ -69,40 +53,24 @@ void VPackToAqlItemBlock(VPackSlice data, uint64_t nrRegs, AqlItemBlock& block)
 
 template <bool passBlocksThrough>
 SingleRowFetcherHelper<passBlocksThrough>::SingleRowFetcherHelper(
-    std::shared_ptr<VPackBuffer<uint8_t>> vPackBuffer, bool returnsWaiting)
+    AqlItemBlockManager& manager,
+    std::shared_ptr<VPackBuffer<uint8_t>> const& vPackBuffer, bool returnsWaiting)
+    : SingleRowFetcherHelper(manager, 1, returnsWaiting,
+                             vPackBufferToAqlItemBlock(manager, vPackBuffer)) {}
+
+template <bool passBlocksThrough>
+SingleRowFetcherHelper<passBlocksThrough>::SingleRowFetcherHelper(::arangodb::aql::AqlItemBlockManager& manager,
+                       size_t const blockSize, bool const returnsWaiting,
+                       ::arangodb::aql::SharedAqlItemBlockPtr input)
     : SingleRowFetcher<passBlocksThrough>(),
-      _vPackBuffer(std::move(vPackBuffer)),
       _returnsWaiting(returnsWaiting),
-      _nrItems(0),
+      _nrItems(input == nullptr ? 0 : input->size()),
-      _nrCalled(0),
+      _blockSize(blockSize),
-      _didWait(false),
+      _itemBlockManager(manager),
-      _resourceMonitor(),
+      _itemBlock(std::move(input)),
-      _itemBlockManager(&_resourceMonitor),
-      _itemBlock(nullptr),
       _lastReturnedRow{CreateInvalidInputRowHint{}} {
-  if (_vPackBuffer != nullptr) {
+  TRI_ASSERT(_blockSize > 0);
-    _data = VPackSlice(_vPackBuffer->data());
+}
-  } else {
-    _data = VPackSlice::nullSlice();
-  }
-  if (_data.isArray()) {
-    _nrItems = _data.length();
-    if (_nrItems > 0) {
-      VPackSlice oneRow = _data.at(0);
-      TRI_ASSERT(oneRow.isArray());
-      arangodb::aql::RegisterCount nrRegs = static_cast<arangodb::aql::RegisterCount>(oneRow.length());
-      // Add all registers as valid input registers:
-      auto inputRegisters = std::make_shared<std::unordered_set<RegisterId>>();
-      for (RegisterId i = 0; i < nrRegs; i++) {
-        inputRegisters->emplace(i);
-      }
-      _itemBlock =
-          SharedAqlItemBlockPtr{new AqlItemBlock(_itemBlockManager, _nrItems, nrRegs)};
-      // std::make_unique<AqlItemBlock>(&_resourceMonitor, _nrItems, nrRegs);
-      VPackToAqlItemBlock(_data, nrRegs, *_itemBlock);
-    }
-  }
-};
 
 template <bool passBlocksThrough>
 SingleRowFetcherHelper<passBlocksThrough>::~SingleRowFetcherHelper() = default;
@@ -112,24 +80,20 @@ template <bool passBlocksThrough>
 std::pair<ExecutionState, InputAqlItemRow> SingleRowFetcherHelper<passBlocksThrough>::fetchRow(size_t) {
   // If this assertion fails, the Executor has fetched more rows after DONE.
   TRI_ASSERT(_nrCalled <= _nrItems);
-  if (_returnsWaiting) {
+  if (wait()) {
-    if (!_didWait) {
+    // if once DONE is returned, always return DONE
-      _didWait = true;
+    if (_returnedDone) {
-      // if once DONE is returned, always return DONE
+      return {ExecutionState::DONE, InputAqlItemRow{CreateInvalidInputRowHint{}}};
-      if (_returnedDone) {
-        return {ExecutionState::DONE, InputAqlItemRow{CreateInvalidInputRowHint{}}};
-      }
-      return {ExecutionState::WAITING, InputAqlItemRow{CreateInvalidInputRowHint{}}};
     }
-    _didWait = false;
+    return {ExecutionState::WAITING, InputAqlItemRow{CreateInvalidInputRowHint{}}};
   }
   _nrCalled++;
   if (_nrCalled > _nrItems) {
     _returnedDone = true;
     return {ExecutionState::DONE, InputAqlItemRow{CreateInvalidInputRowHint{}}};
   }
-  TRI_ASSERT(_itemBlock != nullptr);
+  _lastReturnedRow = InputAqlItemRow{getItemBlock(), _curRowIndex};
-  _lastReturnedRow = InputAqlItemRow{_itemBlock, _nrCalled - 1};
+  nextRow();
   ExecutionState state;
   if (_nrCalled < _nrItems) {
     state = ExecutionState::HASMORE;
@@ -142,22 +106,49 @@ std::pair<ExecutionState, InputAqlItemRow> SingleRowFetcherHelper<passBlocksThro
 
 template <bool passBlocksThrough>
 std::pair<ExecutionState, size_t> SingleRowFetcherHelper<passBlocksThrough>::skipRows(size_t const atMost) {
-  size_t skipped = 0;
   ExecutionState state = ExecutionState::HASMORE;
 
-  while (atMost > skipped) {
+  while (atMost > _skipped) {
-    std::tie(state, std::ignore) = fetchRow();
+    InputAqlItemRow row{CreateInvalidInputRowHint{}};
+    std::tie(state, row) = fetchRow();
     if (state == ExecutionState::WAITING) {
-      return {state, skipped};
+      return {state, 0};
+    }
+    if (row.isInitialized()) {
+      ++_skipped;
     }
-    ++skipped;
     if (state == ExecutionState::DONE) {
+      size_t skipped = _skipped;
+      _skipped = 0;
       return {state, skipped};
     }
   }
 
+  size_t skipped = _skipped;
+  _skipped = 0;
   return {state, skipped};
-};
+}
+
+template <bool passBlocksThrough>
+std::pair<arangodb::aql::ExecutionState, arangodb::aql::SharedAqlItemBlockPtr>
+SingleRowFetcherHelper<passBlocksThrough>::fetchBlockForPassthrough(size_t const atMost) {
+  if (wait()) {
+    return {ExecutionState::WAITING, nullptr};
+  }
+
+  size_t const remainingRows = _blockSize - _curIndexInBlock;
+  size_t const from = _curRowIndex;
+  size_t const to = _curRowIndex + remainingRows;
+
+  bool const isLastBlock = _curRowIndex + _blockSize >= _nrItems;
+  bool const askingForMore = _curRowIndex + atMost > _nrItems;
+
+  bool const done = isLastBlock && askingForMore;
+
+  ExecutionState const state = done ? ExecutionState::DONE : ExecutionState::HASMORE;
+
+  return {state, _itemBlock->slice(from, to)};
+}
 
 // -----------------------------------------
 // - SECTION ALLROWSFETCHER                -

tests/Aql/RowFetcherHelper.h

@@ -31,8 +31,10 @@
 #include "Aql/ConstFetcher.h"
 #include "Aql/ExecutionBlock.h"
 #include "Aql/ExecutionState.h"
+#include "Aql/InputAqlItemRow.h"
 #include "Aql/ResourceUsage.h"
 #include "Aql/SingleRowFetcher.h"
+#include "Aql/VelocyPackHelper.h"
 
 #include <Basics/Common.h>
 #include <velocypack/Buffer.h>
@@ -56,48 +58,80 @@ namespace aql {
  */
 template <bool passBlocksThrough>
 class SingleRowFetcherHelper
-    : public ::arangodb::aql::SingleRowFetcher<passBlocksThrough> {
+    : public arangodb::aql::SingleRowFetcher<passBlocksThrough> {
  public:
-  SingleRowFetcherHelper(std::shared_ptr<arangodb::velocypack::Buffer<uint8_t>> vPackBuffer,
+  SingleRowFetcherHelper(arangodb::aql::AqlItemBlockManager& manager,
+                         size_t blockSize, bool returnsWaiting,
+                         arangodb::aql::SharedAqlItemBlockPtr input);
+
+  // backwards compatible constructor
+  SingleRowFetcherHelper(arangodb::aql::AqlItemBlockManager& manager,
+                         std::shared_ptr<arangodb::velocypack::Buffer<uint8_t>> const& vPackBuffer,
                          bool returnsWaiting);
+
   virtual ~SingleRowFetcherHelper();
 
   // NOLINTNEXTLINE google-default-arguments
-  std::pair<::arangodb::aql::ExecutionState, ::arangodb::aql::InputAqlItemRow> fetchRow(
+  std::pair<arangodb::aql::ExecutionState, arangodb::aql::InputAqlItemRow> fetchRow(
-      size_t atMost = ::arangodb::aql::ExecutionBlock::DefaultBatchSize()) override;
+      size_t atMost = arangodb::aql::ExecutionBlock::DefaultBatchSize()) override;
 
   uint64_t nrCalled() { return _nrCalled; }
 
   std::pair<arangodb::aql::ExecutionState, size_t> skipRows(size_t atMost) override;
 
-  ::arangodb::aql::SharedAqlItemBlockPtr getItemBlock() { return _itemBlock; }
+  std::pair<arangodb::aql::ExecutionState, arangodb::aql::SharedAqlItemBlockPtr> fetchBlockForPassthrough(
+      size_t atMost) override;
 
   bool isDone() const { return _returnedDone; }
 
+  arangodb::aql::AqlItemBlockManager& itemBlockManager() { return _itemBlockManager; }
+
+ private:
+  arangodb::aql::SharedAqlItemBlockPtr& getItemBlock() { return _itemBlock; }
+  arangodb::aql::SharedAqlItemBlockPtr const& getItemBlock() const { return _itemBlock; }
+
+  void nextRow() {
+    _curRowIndex++;
+    _curIndexInBlock = (_curIndexInBlock + 1) % _blockSize;
+  }
+
+  bool wait() {
+    // Wait on the first row of every block, if applicable
+    if (_returnsWaiting && _curIndexInBlock == 0) {
+      // If the insert succeeds, we have not yet waited at this index.
+      bool const waitNow = _didWaitAt.insert(_curRowIndex).second;
+
+      return waitNow;
+    }
+
+    return false;
+  }
+
  private:
-  std::shared_ptr<arangodb::velocypack::Buffer<uint8_t>> _vPackBuffer;
-  arangodb::velocypack::Slice _data;
   bool _returnedDone = false;
-  bool _returnsWaiting;
+  bool const _returnsWaiting;
   uint64_t _nrItems;
-  uint64_t _nrCalled;
+  uint64_t _nrCalled{};
-  bool _didWait;
+  size_t _skipped{};
-  arangodb::aql::ResourceMonitor _resourceMonitor;
+  size_t _curIndexInBlock{};
-  arangodb::aql::AqlItemBlockManager _itemBlockManager;
+  size_t _curRowIndex{};
+  size_t _blockSize;
+  std::unordered_set<size_t> _didWaitAt;
+  arangodb::aql::AqlItemBlockManager& _itemBlockManager;
   arangodb::aql::SharedAqlItemBlockPtr _itemBlock;
-  arangodb::aql::InputAqlItemRow _lastReturnedRow;
+  arangodb::aql::InputAqlItemRow _lastReturnedRow{arangodb::aql::CreateInvalidInputRowHint{}};
 };
 
 /**
  * @brief Mock for AllRowsFetcher
  */
-class AllRowsFetcherHelper : public ::arangodb::aql::AllRowsFetcher {
+class AllRowsFetcherHelper : public arangodb::aql::AllRowsFetcher {
  public:
   AllRowsFetcherHelper(std::shared_ptr<arangodb::velocypack::Buffer<uint8_t>> vPackBuffer,
                        bool returnsWaiting);
   ~AllRowsFetcherHelper();
 
-  std::pair<::arangodb::aql::ExecutionState, ::arangodb::aql::AqlItemMatrix const*> fetchAllRows() override;
+  std::pair<arangodb::aql::ExecutionState, arangodb::aql::AqlItemMatrix const*> fetchAllRows() override;
 
  private:
   std::shared_ptr<arangodb::velocypack::Buffer<uint8_t>> _vPackBuffer;
@@ -118,7 +152,7 @@ class ConstFetcherHelper : public arangodb::aql::ConstFetcher {
                      std::shared_ptr<arangodb::velocypack::Buffer<uint8_t>> vPackBuffer);
   virtual ~ConstFetcherHelper();
 
-  std::pair<::arangodb::aql::ExecutionState, ::arangodb::aql::InputAqlItemRow> fetchRow() override;
+  std::pair<arangodb::aql::ExecutionState, arangodb::aql::InputAqlItemRow> fetchRow() override;
 
  private:
   std::shared_ptr<arangodb::velocypack::Buffer<uint8_t>> _vPackBuffer;

tests/Aql/ShortestPathExecutorTest.cpp

@@ -252,7 +252,7 @@ class ShortestPathExecutorTest : public ::testing::Test {
     ExecutionState state = ExecutionState::HASMORE;
     auto& finder = dynamic_cast<FakePathFinder&>(infos.finder());
 
-    SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+    SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
     OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
                             infos.registersToKeep(), infos.registersToClear());
     ShortestPathExecutor testee(fetcher, infos);
@@ -300,7 +300,7 @@ class ShortestPathExecutorTest : public ::testing::Test {
     ExecutionState state = ExecutionState::HASMORE;
     auto& finder = dynamic_cast<FakePathFinder&>(infos.finder());
 
-    SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+    SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
     OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
                             infos.registersToKeep(), infos.registersToClear());
     ShortestPathExecutor testee(fetcher, infos);

tests/Aql/SortedCollectExecutorTest.cpp

@@ -105,7 +105,7 @@ class SortedCollectExecutorTestNoRowsUpstream : public ::testing::Test {
 };
 
 TEST_F(SortedCollectExecutorTestNoRowsUpstream, producer_doesnt_wait) {
-  SingleRowFetcherHelper<false> fetcher(input.steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), false);
   SortedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -116,7 +116,7 @@ TEST_F(SortedCollectExecutorTestNoRowsUpstream, producer_doesnt_wait) {
 }
 
 TEST_F(SortedCollectExecutorTestNoRowsUpstream, producer_waits) {
-  SingleRowFetcherHelper<false> fetcher(input.steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), true);
   SortedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -188,7 +188,7 @@ class SortedCollectExecutorTestRowsUpstream : public ::testing::Test {
 
 TEST_F(SortedCollectExecutorTestRowsUpstream, producer_doesnt_wait) {
   auto input = VPackParser::fromJson("[ [1], [2] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   SortedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -224,7 +224,7 @@ TEST_F(SortedCollectExecutorTestRowsUpstream, producer_doesnt_wait) {
 
 TEST_F(SortedCollectExecutorTestRowsUpstream, producer_doesnt_wait_2) {
   auto input = VPackParser::fromJson("[ [1], [2], [3] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   SortedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -268,7 +268,7 @@ TEST_F(SortedCollectExecutorTestRowsUpstream, producer_doesnt_wait_2) {
 TEST_F(SortedCollectExecutorTestRowsUpstream, producer_doesnt_wait_3) {
   // Input order needs to be guaranteed
   auto input = VPackParser::fromJson("[ [1], [1], [2], [2], [3] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   SortedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -312,7 +312,7 @@ TEST_F(SortedCollectExecutorTestRowsUpstream, producer_doesnt_wait_3) {
 
 TEST_F(SortedCollectExecutorTestRowsUpstream, producer_doesnt_wait_4) {
   auto input = VPackParser::fromJson("[ [1], [1], [2], [2] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   SortedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -347,7 +347,7 @@ TEST_F(SortedCollectExecutorTestRowsUpstream, producer_doesnt_wait_4) {
 
 TEST_F(SortedCollectExecutorTestRowsUpstream, producer_waits) {
   auto input = VPackParser::fromJson("[ [1], [2] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   SortedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -435,7 +435,7 @@ TEST(SortedCollectExecutorTestRowsUpstreamCount, test) {
   NoStats stats{};
 
   auto input = VPackParser::fromJson("[ [1], [2] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   SortedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -526,7 +526,7 @@ TEST(SortedCollectExecutorTestRowsUpstreamCountNumbers, test) {
   NoStats stats{};
 
   auto input = VPackParser::fromJson("[ [1], [2], [3] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   SortedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),
@@ -632,7 +632,7 @@ TEST(SortedCollectExecutorTestRowsUpstreamCountStrings, test) {
   NoStats stats{};
 
   auto input = VPackParser::fromJson("[ [\"a\"], [\"aa\"], [\"aaa\"] ]");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   SortedCollectExecutor testee(fetcher, infos);
 
   OutputAqlItemRow result(std::move(block), infos.getOutputRegisters(),

tests/Aql/TestEmptyExecutorHelper.h

@@ -77,13 +77,6 @@ class TestEmptyExecutorHelper {
    * @return ExecutionState, and if successful exactly one new Row of AqlItems.
    */
   std::pair<ExecutionState, Stats> produceRows(OutputAqlItemRow& output);
-
-  inline std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t) const {
-    TRI_ASSERT(false);
-    THROW_ARANGO_EXCEPTION_MESSAGE(
-        TRI_ERROR_INTERNAL,
-        "Logic_error, prefetching number of rows not supported");
-  }
 };
 
 }  // namespace aql

tests/Aql/TestExecutorHelper.h

@@ -62,9 +62,6 @@ class TestExecutorHelperInfos : public ExecutorInfos {
   RegisterId _inputRegister;
 };
 
-/**
- * @brief Implementation of Filter Node
- */
 class TestExecutorHelper {
  public:
   struct Properties {
@@ -89,13 +86,6 @@ class TestExecutorHelper {
    */
   std::pair<ExecutionState, Stats> produceRows(OutputAqlItemRow& output);
 
-  inline std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t) const {
-    TRI_ASSERT(false);
-    THROW_ARANGO_EXCEPTION_MESSAGE(
-        TRI_ERROR_INTERNAL,
-        "Logic_error, prefetching number fo rows not supported");
-  }
-
  public:
   Infos& _infos;
 

tests/Aql/TraversalExecutorTest.cpp

@@ -293,7 +293,7 @@ class TraversalExecutorTestInputStartVertex : public ::testing::Test {
 
 TEST_F(TraversalExecutorTestInputStartVertex, there_are_no_rows_upstream_producer_doesnt_wait) {
   VPackBuilder input;
-  SingleRowFetcherHelper<false> fetcher(input.steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), false);
   TraversalExecutor testee(fetcher, infos);
   TraversalStats stats{};
 
@@ -306,7 +306,7 @@ TEST_F(TraversalExecutorTestInputStartVertex, there_are_no_rows_upstream_produce
 
 TEST_F(TraversalExecutorTestInputStartVertex, there_are_no_rows_upstream_producer_waits) {
   VPackBuilder input;
-  SingleRowFetcherHelper<false> fetcher(input.steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), true);
   TraversalExecutor testee(fetcher, infos);
   TraversalStats stats{};
 
@@ -328,7 +328,7 @@ TEST_F(TraversalExecutorTestInputStartVertex, there_are_rows_upstream_producer_d
   myGraph.addVertex("2");
   myGraph.addVertex("3");
   auto input = VPackParser::fromJson(R"([["v/1"], ["v/2"], ["v/3"]])");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   TraversalExecutor testee(fetcher, infos);
   TraversalStats stats{};
 
@@ -360,7 +360,7 @@ TEST_F(TraversalExecutorTestInputStartVertex,
   myGraph.addVertex("2");
   myGraph.addVertex("3");
   auto input = VPackParser::fromJson(R"([["v/1"], ["v/2"], ["v/3"]])");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   TraversalExecutor testee(fetcher, infos);
   TraversalStats stats{};
 
@@ -398,7 +398,7 @@ TEST_F(TraversalExecutorTestInputStartVertex,
   myGraph.addVertex("2");
   myGraph.addVertex("3");
   auto input = VPackParser::fromJson(R"([["v/1"], ["v/2"], ["v/3"]])");
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   TraversalExecutor testee(fetcher, infos);
   TraversalStats stats{};
 
@@ -498,7 +498,7 @@ class TraversalExecutorTestConstantStartVertex : public ::testing::Test {
 
 TEST_F(TraversalExecutorTestConstantStartVertex, no_rows_upstream_producer_doesnt_wait) {
   VPackBuilder input;
-  SingleRowFetcherHelper<false> fetcher(input.steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), false);
   TraversalExecutor testee(fetcher, infos);
   TraversalStats stats{};
 
@@ -511,7 +511,7 @@ TEST_F(TraversalExecutorTestConstantStartVertex, no_rows_upstream_producer_doesn
 
 TEST_F(TraversalExecutorTestConstantStartVertex, no_rows_upstream_producer_waits) {
   VPackBuilder input;
-  SingleRowFetcherHelper<false> fetcher(input.steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input.steal(), true);
   TraversalExecutor testee(fetcher, infos);
   TraversalStats stats{};
 
@@ -534,7 +534,7 @@ TEST_F(TraversalExecutorTestConstantStartVertex, rows_upstream_producer_doesnt_w
   myGraph.addVertex("3");
   auto input = VPackParser::fromJson(R"([ ["v/1"], ["v/2"], ["v/3"] ])");
 
-  SingleRowFetcherHelper<false> fetcher(input->steal(), false);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), false);
   TraversalExecutor testee(fetcher, infos);
   TraversalStats stats{};
 
@@ -566,7 +566,7 @@ TEST_F(TraversalExecutorTestConstantStartVertex, rows_upstream_producer_waits_no
   myGraph.addVertex("3");
   auto input = VPackParser::fromJson(R"([ ["v/1"], ["v/2"], ["v/3"] ])");
 
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   TraversalExecutor testee(fetcher, infos);
   TraversalStats stats{};
   OutputAqlItemRow row(std::move(block), infos.getOutputRegisters(),
@@ -603,7 +603,7 @@ TEST_F(TraversalExecutorTestConstantStartVertex, rows_upstream_producer_waits_ed
   myGraph.addVertex("3");
   auto input = VPackParser::fromJson(R"([ ["v/1"], ["v/2"], ["v/3"] ])");
 
-  SingleRowFetcherHelper<false> fetcher(input->steal(), true);
+  SingleRowFetcherHelper<false> fetcher(itemBlockManager, input->steal(), true);
   TraversalExecutor testee(fetcher, infos);
   TraversalStats stats{};
   myGraph.addEdge("1", "2", "1->2");

tests/Aql/VelocyPackHelper.cpp

@@ -22,20 +22,85 @@
 
 #include "VelocyPackHelper.h"
 
-using namespace arangodb;
+#include <velocypack/velocypack-aliases.h>
-using namespace arangodb::tests::aql;
 
-VPackBufferPtr arangodb::tests::aql::vpackFromJsonString(char const* c) {
+using namespace arangodb;
-  velocypack::Options options;
+using namespace arangodb::aql;
+using namespace arangodb::tests;
+
+VPackBufferPtr arangodb::tests::vpackFromJsonString(char const* c) {
+  VPackOptions options;
   options.checkAttributeUniqueness = true;
-  velocypack::Parser parser(&options);
+  VPackParser parser(&options);
   parser.parse(c);
 
-  std::shared_ptr<velocypack::Builder> builder = parser.steal();
+  std::shared_ptr<VPackBuilder> builder = parser.steal();
 
   return builder->steal();
 }
 
-VPackBufferPtr arangodb::tests::aql::operator"" _vpack(const char* json, size_t) {
+VPackBufferPtr arangodb::tests::operator"" _vpack(const char* json, size_t) {
   return vpackFromJsonString(json);
 }
+
+void arangodb::tests::VPackToAqlItemBlock(VPackSlice data, RegisterCount nrRegs,
+                                          AqlItemBlock& block) {
+  // coordinates in the matrix rowNr, entryNr
+  size_t rowIndex = 0;
+  RegisterId entry = 0;
+  for (auto const& row : VPackArrayIterator(data)) {
+    // Walk through the rows
+    TRI_ASSERT(row.isArray());
+    TRI_ASSERT(row.length() == nrRegs);
+    for (auto const& oneEntry : VPackArrayIterator(row)) {
+      // Walk through on row values
+      block.setValue(rowIndex, entry, AqlValue{oneEntry});
+      entry++;
+    }
+    rowIndex++;
+    entry = 0;
+  }
+}
+
+arangodb::aql::SharedAqlItemBlockPtr arangodb::tests::vPackBufferToAqlItemBlock(
+    arangodb::aql::AqlItemBlockManager& manager, VPackBufferPtr const& buffer) {
+  if(VPackSlice(buffer->data()).isNone()) {
+    return nullptr;
+  }
+  return multiVPackBufferToAqlItemBlocks(manager, buffer)[0];
+}
+
+std::vector<SharedAqlItemBlockPtr> arangodb::tests::vPackToAqlItemBlocks(
+    AqlItemBlockManager& manager, VPackBufferPtr const& buffer) {
+  VPackSlice outer(buffer->data());
+  if (outer.isNone()) {
+    return {};
+  }
+  TRI_ASSERT(outer.isArray());
+  if (outer.length() == 0) {
+    return {};
+  }
+  size_t const nrRegs = [&]() {
+    VPackSlice firstRow(outer[0]);
+    TRI_ASSERT(firstRow.isArray());
+    return firstRow.length();
+  }();
+
+  auto wrap = [](VPackSlice slice) -> VPackBufferPtr {
+    VPackBuilder builder;
+    builder.openArray();
+    builder.add(slice);
+    builder.close();
+    return builder.steal();
+  };
+
+  std::vector<SharedAqlItemBlockPtr> blocks;
+
+  for (VPackSlice inner : VPackArrayIterator(outer)) {
+    SharedAqlItemBlockPtr block = manager.requestBlock(1, nrRegs);
+    VPackToAqlItemBlock(VPackSlice(wrap(inner)->data()), nrRegs, *block);
+    blocks.emplace_back(block);
+  }
+
+  return blocks;
+}

tests/Aql/VelocyPackHelper.h

@@ -23,15 +23,20 @@
 #ifndef ARANGOD_AQL_TESTS_VELOCYPACK_HELPER_H
 #define ARANGOD_AQL_TESTS_VELOCYPACK_HELPER_H
 
+#include "Aql/AqlItemBlock.h"
+#include "Aql/AqlItemBlockManager.h"
+#include "Aql/AqlValue.h"
+#include "Aql/SharedAqlItemBlockPtr.h"
+
 #include <velocypack/Buffer.h>
 #include <velocypack/Builder.h>
+#include <velocypack/Iterator.h>
 #include <velocypack/Options.h>
 #include <velocypack/Parser.h>
 #include <memory>
 
 namespace arangodb {
 namespace tests {
-namespace aql {
 
 using VPackBufferPtr = std::shared_ptr<velocypack::Buffer<uint8_t>>;
 
@@ -39,7 +44,61 @@ VPackBufferPtr vpackFromJsonString(char const* c);
 
 VPackBufferPtr operator"" _vpack(const char* json, size_t);
 
-}  // namespace aql
+void VPackToAqlItemBlock(velocypack::Slice data, arangodb::aql::RegisterCount nrRegs,
+                         arangodb::aql::AqlItemBlock& block);
+
+// Convert a single VPackBuffer into an AqlItemBlock
+arangodb::aql::SharedAqlItemBlockPtr vPackBufferToAqlItemBlock(
+  arangodb::aql::AqlItemBlockManager& manager, VPackBufferPtr const& buffer);
+
+/**
+ * @brief Convert a list of VPackBufferPtr to a vector of AqlItemBlocks.
+ * Does no error handling but for maintainer mode assertions: It's meant for
+ * tests with static input.
+ */
+template <typename... Ts>
+std::vector<arangodb::aql::SharedAqlItemBlockPtr> multiVPackBufferToAqlItemBlocks(
+    arangodb::aql::AqlItemBlockManager& manager, Ts... vPackBuffers) {
+  std::vector<VPackBufferPtr> buffers({std::forward<Ts>(vPackBuffers)...});
+  arangodb::aql::RegisterCount const nrRegs = [&]() -> arangodb::aql::RegisterCount {
+    if (buffers.empty()) {
+      return 0;
+    }
+    for (size_t i = 0; i < buffers.size(); i++) {
+      velocypack::Slice block(buffers[0]->data());
+      TRI_ASSERT(block.isArray());
+      if (block.length() > 0) {
+        velocypack::Slice firstRow(block[0]);
+        TRI_ASSERT(firstRow.isArray());
+        return static_cast<arangodb::aql::RegisterCount>(firstRow.length());
+      }
+    }
+    // no rows in any block
+    return 0;
+  }();
+
+  std::vector<arangodb::aql::SharedAqlItemBlockPtr> blocks{};
+
+  for (auto const& buffer : buffers) {
+    velocypack::Slice slice(buffer->data());
+    TRI_ASSERT(slice.isArray());
+    size_t const nrItems = slice.length();
+    arangodb::aql::SharedAqlItemBlockPtr block = nullptr;
+    if (nrItems > 0) {
+      block = manager.requestBlock(nrItems, nrRegs);
+      VPackToAqlItemBlock(slice, nrRegs, *block);
+    }
+    blocks.emplace_back(block);
+  }
+
+  return blocks;
+}
+
+// Expects buffer to be an array of arrays. For every inner array, an
+// AqlItemBlock with a single row matching the inner array is returned.
+std::vector<arangodb::aql::SharedAqlItemBlockPtr> vPackToAqlItemBlocks(
+    arangodb::aql::AqlItemBlockManager& manager, VPackBufferPtr const& buffer);
+
 }  // namespace tests
 }  // namespace arangodb
 

tests/CMakeLists.txt

@@ -20,9 +20,9 @@ set(ARANGODB_TESTS_SOURCES
   Agency/RemoveFollowerTest.cpp
   Agency/StoreTest.cpp
   Agency/SupervisionTest.cpp
-  Aql/AllRowsFetcherTest.cpp
+  Aql/AqlHelper.cpp
-  Aql/AqlItemBlockHelper.cpp
   Aql/AqlItemRowTest.cpp
+  Aql/AllRowsFetcherTest.cpp
   Aql/CalculationExecutorTest.cpp
   Aql/CountCollectExecutorTest.cpp
   Aql/DateFunctionsTest.cpp
@@ -33,6 +33,7 @@ set(ARANGODB_TESTS_SOURCES
   Aql/EnumerateListExecutorTest.cpp
   Aql/ExecutionBlockImplTest.cpp
   Aql/ExecutionBlockImplTestInstances.cpp
+  Aql/ExecutorTestHelper.cpp
   Aql/FilterExecutorTest.cpp
   Aql/HashedCollectExecutorTest.cpp
   Aql/IdExecutorTest.cpp

tests/js/server/aql/aql-queries-optimizer-sort-limit.js

@@ -217,7 +217,7 @@ function ahuacatlQueryOptimizerLimitTestSuite () {
         var query = "FOR c IN " + cn + " SORT c.value LIMIT " + test.offset + ", " + test.limit + " LIMIT " + test.offset2 + ", " + test.limit2 + " RETURN c";
 
         var actual = getQueryResults(query);
-        assertEqual(test.expectedLength, actual.length);
+        assertEqual(test.expectedLength, actual.length, `Test #${i}, query was: ${query}`);
 
         var sorts = getSorts(query);
         assertEqual(sorts.length, 1);