arangodb · mchacki · Mar 20, 2020 · Oct 22, 2019 · Oct 22, 2019 · Oct 22, 2019
diff --git a/arangod/Aql/AqlItemBlockInputRange.cpp b/arangod/Aql/AqlItemBlockInputRange.cpp
@@ -181,3 +181,23 @@ auto AqlItemBlockInputRange::skipAll() noexcept -> std::size_t {
   _skipped = 0;
   return skipped;
 }
+
+auto AqlItemBlockInputRange::countDataRows() const noexcept -> std::size_t {
+  if (_block == nullptr) {
+    return 0;
+  }
+  auto const& block = getBlock();
+  return block->size() - block->getShadowRowIndexes().size();
+}
+
+auto AqlItemBlockInputRange::countShadowRows() const noexcept -> std::size_t {
+  if (_block == nullptr) {
+    return 0;
+  }
+  auto const& block = getBlock();
+  return block->getShadowRowIndexes().size();
+}
+
+[[nodiscard]] auto AqlItemBlockInputRange::finalState() const noexcept -> ExecutorState {
+  return _finalState;
+}
diff --git a/arangod/Aql/AqlItemBlockInputRange.h b/arangod/Aql/AqlItemBlockInputRange.h
@@ -72,6 +72,22 @@ class AqlItemBlockInputRange {
 
   [[nodiscard]] auto skippedInFlight() const noexcept -> std::size_t;
 
+  /**
+   * @brief Count how many datarows are expected in this range
+   *        Used to estimate amount of produced rows
+   * @return std::size_t
+   */
+  [[nodiscard]] auto countDataRows() const noexcept -> std::size_t;
+
+  /**
+   * @brief Count how many shadowRows are expected in this range
+   *        Used to estimate amount of produced rows
+   * @return std::size_t
+   */
+  [[nodiscard]] auto countShadowRows() const noexcept -> std::size_t;
+
+  [[nodiscard]] auto finalState() const noexcept -> ExecutorState;
+
  private:
   bool isIndexValid(std::size_t index) const noexcept;
 

diff --git a/arangod/Aql/CountCollectExecutor.cpp b/arangod/Aql/CountCollectExecutor.cpp
@@ -108,6 +108,12 @@ std::pair<ExecutionState, size_t> CountCollectExecutor::expectedNumberOfRows(siz
   return {ExecutionState::HASMORE, 1};
 }
 
+auto CountCollectExecutor::expectedNumberOfRowsNew(AqlItemBlockInputRange const& input,
+                                                   AqlCall const& call) const
+    noexcept -> size_t {
+  return std::min<size_t>(1, call.getLimit());
+}
+
 const CountCollectExecutor::Infos& CountCollectExecutor::infos() const noexcept {
   return _infos;
 }

diff --git a/arangod/Aql/CountCollectExecutor.h b/arangod/Aql/CountCollectExecutor.h
@@ -105,6 +105,9 @@ class CountCollectExecutor {
 
   std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t atMost) const;
 
+  [[nodiscard]] auto expectedNumberOfRowsNew(AqlItemBlockInputRange const& input,
+                                             AqlCall const& call) const noexcept -> size_t;
+
  private:
   Infos const& infos() const noexcept;
 

diff --git a/arangod/Aql/DistinctCollectExecutor.cpp b/arangod/Aql/DistinctCollectExecutor.cpp
@@ -85,6 +85,18 @@ std::pair<ExecutionState, size_t> DistinctCollectExecutor::expectedNumberOfRows(
   return _fetcher.preFetchNumberOfRows(atMost);
 }
 
+[[nodiscard]] auto DistinctCollectExecutor::expectedNumberOfRowsNew(
+    AqlItemBlockInputRange const& input, AqlCall const& call) const noexcept -> size_t {
+  if (input.finalState() == ExecutorState::DONE) {
+    // Worst case assumption:
+    // For every input row we have a new group.
+    // We will never produce more then asked for
+    return std::min(call.getLimit(), input.countDataRows());
+  }
+  // Otherwise we do not know.
+  return call.getLimit();
+}
+
 void DistinctCollectExecutor::destroyValues() {
   // destroy all AqlValues captured
   for (auto& value : _seen) {

diff --git a/arangod/Aql/DistinctCollectExecutor.h b/arangod/Aql/DistinctCollectExecutor.h
@@ -120,6 +120,9 @@ class DistinctCollectExecutor {
 
   std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t atMost) const;
 
+  [[nodiscard]] auto expectedNumberOfRowsNew(AqlItemBlockInputRange const& input,
+                                             AqlCall const& call) const noexcept -> size_t;
+
  private:
   Infos const& infos() const noexcept;
   void destroyValues();

diff --git a/arangod/Aql/ExecutionBlockImpl.cpp b/arangod/Aql/ExecutionBlockImpl.cpp
@@ -114,6 +114,7 @@ CREATE_HAS_MEMBER_CHECK(skipRows, hasSkipRows);
 CREATE_HAS_MEMBER_CHECK(fetchBlockForPassthrough, hasFetchBlockForPassthrough);
 CREATE_HAS_MEMBER_CHECK(expectedNumberOfRows, hasExpectedNumberOfRows);
 CREATE_HAS_MEMBER_CHECK(skipRowsRange, hasSkipRowsRange);
+CREATE_HAS_MEMBER_CHECK(expectedNumberOfRowsNew, hasExpectedNumberOfRowsNew);
 
 #ifdef ARANGODB_USE_GOOGLE_TESTS
 // Forward declaration of Test Executors.
@@ -1032,7 +1033,7 @@ std::pair<ExecutionState, SharedAqlItemBlockPtr> ExecutionBlockImpl<Executor>::r
 
 // TODO: We need to define the size of this block based on Input / Executor / Subquery depth
 template <class Executor>
-auto ExecutionBlockImpl<Executor>::allocateOutputBlock(AqlCall&& call)
+auto ExecutionBlockImpl<Executor>::allocateOutputBlock(AqlCall&& call, DataRange const& inputRange)
     -> std::unique_ptr<OutputAqlItemRow> {
   if constexpr (Executor::Properties::allowsBlockPassthrough == BlockPassthrough::Enable) {
     SharedAqlItemBlockPtr newBlock{nullptr};
@@ -1047,18 +1048,56 @@ auto ExecutionBlockImpl<Executor>::allocateOutputBlock(AqlCall&& call)
 
     return createOutputRow(newBlock, std::move(call));
   } else {
+    if constexpr (isMultiDepExecutor<Executor>) {
+      // MultiDepExecutor would require dependency handling.
+      // We do not have it here.
+      if (!inputRange.hasShadowRow() && !inputRange.hasDataRow()) {
+        // On empty input do not yet create output.
+        // We are going to ask again later
+        SharedAqlItemBlockPtr newBlock{nullptr};
+        return createOutputRow(newBlock, std::move(call));
+      }
+    } else {
+      if (!inputRange.hasShadowRow() && !inputRange.hasDataRow() &&
+          inputRange.upstreamState() == ExecutorState::HASMORE) {
+        // On empty input do not yet create output.
+        // We are going to ask again later
+        SharedAqlItemBlockPtr newBlock{nullptr};
+        return createOutputRow(newBlock, std::move(call));
+      }
+    }
+
     // Non-Passthrough variant, we need to allocate the block ourselfs
     size_t blockSize = ExecutionBlock::DefaultBatchSize;
+    if constexpr (hasExpectedNumberOfRowsNew<Executor>::value) {
+      blockSize = _executor.expectedNumberOfRowsNew(inputRange, call);
+      // The executor cannot expect to produce more then the limit!
+      if constexpr (!std::is_same_v<Executor, SubqueryStartExecutor>) {
+        // Except the subqueryStartExecutor, it's limit differs
+        // from it's output (it needs to count the new ShadowRows in addition)
+        TRI_ASSERT(blockSize <= call.getLimit());
+      }
+
+      blockSize += inputRange.countShadowRows();
+      // We have an upper bound by DefaultBatchSize;
+      blockSize = std::min(ExecutionBlock::DefaultBatchSize, blockSize);
+    }
+    if (blockSize == 0) {
+      // There is no data to be produced
+      SharedAqlItemBlockPtr newBlock{nullptr};
+      return createOutputRow(newBlock, std::move(call));
+    }
     SharedAqlItemBlockPtr newBlock =
         _engine->itemBlockManager().requestBlock(blockSize, _infos.numberOfOutputRegisters());
     return createOutputRow(newBlock, std::move(call));
   }
 }
 
 template <class Executor>
-void ExecutionBlockImpl<Executor>::ensureOutputBlock(AqlCall&& call) {
+void ExecutionBlockImpl<Executor>::ensureOutputBlock(AqlCall&& call,
+                                                     DataRange const& inputRange) {
   if (_outputItemRow == nullptr || !_outputItemRow->isInitialized()) {
-    _outputItemRow = allocateOutputBlock(std::move(call));
+    _outputItemRow = allocateOutputBlock(std::move(call), inputRange);
   } else {
     _outputItemRow->setCall(std::move(call));
   }
@@ -1251,7 +1290,7 @@ auto ExecutionBlockImpl<Executor>::executeFetcher(AqlCallStack& stack, AqlCallTy
       return {state, skipped, _lastRange};
     } else if constexpr (executorHasSideEffects<Executor>) {
       // If the executor has side effects, we cannot bypass any subqueries
-      // by skipping them. SO we need to fetch all shadow rows in order to
+      // by skipping them. So we need to fetch all shadow rows in order to
       // trigger this Executor with everthing from above.
       // NOTE: The Executor needs to discard shadowRows, and do the accouting.
       static_assert(std::is_same_v<AqlCall, std::decay_t<decltype(aqlCall)>>);
@@ -1939,9 +1978,9 @@ ExecutionBlockImpl<Executor>::executeWithoutTrace(AqlCallStack stack) {
             TRI_ASSERT(!stack.empty());
             AqlCall const& subqueryCall = stack.peek();
             AqlCall copyCall = subqueryCall;
-            ensureOutputBlock(std::move(copyCall));
+            ensureOutputBlock(std::move(copyCall), _lastRange);
           } else {
-            ensureOutputBlock(std::move(clientCall));
+            ensureOutputBlock(std::move(clientCall), _lastRange);
           }
           TRI_ASSERT(_outputItemRow);
           TRI_ASSERT(!_executorReturnedDone);
@@ -2133,9 +2172,9 @@ ExecutionBlockImpl<Executor>::executeWithoutTrace(AqlCallStack stack) {
             TRI_ASSERT(!stack.empty());
             AqlCall const& subqueryCall = stack.peek();
             AqlCall copyCall = subqueryCall;
-            ensureOutputBlock(std::move(copyCall));
+            ensureOutputBlock(std::move(copyCall), _lastRange);
           } else {
-            ensureOutputBlock(std::move(clientCall));
+            ensureOutputBlock(std::move(clientCall), _lastRange);
           }
 
           TRI_ASSERT(!_outputItemRow->allRowsUsed());

diff --git a/arangod/Aql/ExecutionBlockImpl.h b/arangod/Aql/ExecutionBlockImpl.h
@@ -309,12 +309,12 @@ class ExecutionBlockImpl final : public ExecutionBlock {
   [[nodiscard]] ExecutionState fetchShadowRowInternal();
 
   // Allocate an output block and install a call in it
-  [[nodiscard]] auto allocateOutputBlock(AqlCall&& call)
+  [[nodiscard]] auto allocateOutputBlock(AqlCall&& call, DataRange const& inputRange)
       -> std::unique_ptr<OutputAqlItemRow>;
 
-  // Ensure that we have an output block of the desired dimenstions
+  // Ensure that we have an output block of the desired dimensions
   // Will as a side effect modify _outputItemRow
-  void ensureOutputBlock(AqlCall&& call);
+  void ensureOutputBlock(AqlCall&& call, DataRange const& inputRange);
 
   // Compute the next state based on the given call.
   // Can only be one of Skip/Produce/FullCount/FastForward/Done

diff --git a/arangod/Aql/FilterExecutor.cpp b/arangod/Aql/FilterExecutor.cpp
@@ -153,3 +153,14 @@ auto FilterExecutor::produceRows(AqlItemBlockInputRange& inputRange, OutputAqlIt
                            (std::min)(clientCall.softLimit, clientCall.hardLimit);
   return {inputRange.upstreamState(), stats, upstreamCall};
 }
+
+[[nodiscard]] auto FilterExecutor::expectedNumberOfRowsNew(AqlItemBlockInputRange const& input,
+                                                           AqlCall const& call) const
+    noexcept -> size_t {
+  if (input.finalState() == ExecutorState::DONE) {
+    return std::min(call.getLimit(), input.countDataRows());
+  }
+  // We do not know how many more rows will be returned from upstream.
+  // So we can only overestimate
+  return call.getLimit();
+}
diff --git a/arangod/Aql/FilterExecutor.h b/arangod/Aql/FilterExecutor.h
@@ -108,6 +108,9 @@ class FilterExecutor {
 
   [[nodiscard]] std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t atMost) const;
 
+  [[nodiscard]] auto expectedNumberOfRowsNew(AqlItemBlockInputRange const& input,
+                                             AqlCall const& call) const noexcept -> size_t;
+
  private:
   Infos& _infos;
   Fetcher& _fetcher;

diff --git a/arangod/Aql/HashedCollectExecutor.cpp b/arangod/Aql/HashedCollectExecutor.cpp
@@ -360,6 +360,28 @@ std::pair<ExecutionState, size_t> HashedCollectExecutor::expectedNumberOfRows(si
   THROW_ARANGO_EXCEPTION(TRI_ERROR_NOT_IMPLEMENTED);
 }
 
+[[nodiscard]] auto HashedCollectExecutor::expectedNumberOfRowsNew(
+    AqlItemBlockInputRange const& input, AqlCall const& call) const noexcept -> size_t {
+  if (!_isInitialized) {
+    if (input.finalState() == ExecutorState::DONE) {
+      // Worst case assumption:
+      // For every input row we have a new group.
+      // We will never produce more then asked for
+      auto estOnInput = input.countDataRows();
+      if (estOnInput == 0 && _infos.getGroupRegisters().empty()) {
+        // Special case, on empty input we will produce 1 output
+        estOnInput = 1;
+      }
+      return std::min(call.getLimit(), estOnInput);
+    }
+    // Otherwise we do not know.
+    return call.getLimit();
+10000
  }
+  // We know how many groups we have left
+  return std::min<size_t>(call.getLimit(),
+                          std::distance(_currentGroup, _allGroups.end()));
+}
+
 const HashedCollectExecutor::Infos& HashedCollectExecutor::infos() const noexcept {
   return _infos;
 }
diff --git a/arangod/Aql/HashedCollectExecutor.h b/arangod/Aql/HashedCollectExecutor.h
@@ -168,6 +168,15 @@ class HashedCollectExecutor {
    */
   std::pair<ExecutionState, size_t> expectedNumberOfRows(size_t atMost) const;
 
+  /**
+   * @brief This Executor does not know how many distinct rows will be fetched
+   * from upstream, it can only report how many it has found by itself, plus
+   * it knows that it can only create as many new rows as pulled from upstream.
+   * So it will overestimate.
+   */
+  [[nodiscard]] auto expectedNumberOfRowsNew(AqlItemBlockInputRange const& input,
+                                             AqlCall const& call) const noexcept -> size_t;
+
  private:
   using AggregateValuesType = std::vector<std::unique_ptr<Aggregator>>;
   using GroupKeyType = std::vector<AqlValue>;
@@ -215,7 +224,7 @@ class HashedCollectExecutor {
 
   /// @brief hashmap of all encountered groups
   GroupMapType _allGroups;
-  GroupMapType::iterator _currentGroup;
+  GroupMapType::const_iterator _currentGroup;
 
   bool _isInitialized;  // init() was called successfully (e.g. it returned DONE)
 

diff --git a/arangod/Aql/IdExecutor.cpp b/arangod/Aql/IdExecutor.cpp
@@ -88,25 +88,22 @@ auto IdExecutor<UsedFetcher>::produceRows(AqlItemBlockInputRange& inputRange,
     -> std::tuple<ExecutorState, CountStats, AqlCall> {
   CountStats stats;
   TRI_ASSERT(output.numRowsWritten() == 0);
-  // TODO: We can implement a fastForward copy here.
-  // We know that all rows we have will fit into the output
-  while (!output.isFull() && inputRange.hasDataRow()) {
-    auto const& [state, inputRow] = inputRange.nextDataRow();
-    TRI_ASSERT(inputRow);
-
+  if (inputRange.hasDataRow()) {
+    TRI_ASSERT(!output.isFull());
     TRI_IF_FAILURE("SingletonBlock::getOrSkipSome") {
       THROW_ARANGO_EXCEPTION(TRI_ERROR_DEBUG);
     }
+    auto const& [state, inputRow] = inputRange.peekDataRow();
+
+    output.fastForwardAllRows(inputRow, inputRange.countDataRows());
 
-    /*Second parameter are to ignore registers that should be kept but are missing in the input row*/
-    output.copyRow(inputRow, std::is_same_v<UsedFetcher, ConstFetcher>);
-    TRI_ASSERT(output.produced());
-    output.advanceRow();
+    std::ignore = inputRange.skipAllRemainingDataRows();
 
     TRI_IF_FAILURE("SingletonBlock::getOrSkipSomeSet") {
       THROW_ARANGO_EXCEPTION(TRI_ERROR_DEBUG);
     }
   }
+  TRI_ASSERT(!inputRange.hasDataRow());
   if (_infos.doCount()) {
     stats.addCounted(output.numRowsWritten());
   }

diff --git a/arangod/Aql/NoResultsExecutor.cpp b/arangod/Aql/NoResultsExecutor.cpp
@@ -42,4 +42,11 @@ auto NoResultsExecutor::skipRowsRange(AqlItemBlockInputRange& inputRange, AqlCal
     noexcept -> std::tuple<ExecutorState, Stats, size_t, AqlCall> {
   return {inputRange.upstreamState(), NoStats{}, 0,
           AqlCall{0, false, 0, AqlCall::LimitType::HARD}};
-};
+};
+
+[[nodiscard]] auto NoResultsExecutor::expectedNumberOfRowsNew(AqlItemBlockInputRange const& input,
+                                                              AqlCall const& call) const
+    noexcept -> size_t {
+  // Well nevermind the input, but we will always return 0 rows here.
+  return 0;
+}
diff --git a/arangod/Aql/NoResultsExecutor.h b/arangod/Aql/NoResultsExecutor.h
@@ -76,6 +76,9 @@ class NoResultsExecutor {
     // Well nevermind the input, but we will always return 0 rows here.
     return {ExecutionState::DONE, 0};
   }
+
+  [[nodiscard]] auto expectedNumberOfRowsNew(AqlItemBlockInputRange const& input,
+                                             AqlCall const& call) const noexcept -> size_t;
 };
 }  // namespace aql
 }  // namespace arangodb

diff --git a/arangod/Aql/OutputAqlItemRow.cpp b/arangod/Aql/OutputAqlItemRow.cpp
@@ -166,6 +166,30 @@ void OutputAqlItemRow::copyRow(ItemRowType const& sourceRow, bool ignoreMissing)
   doCopyRow(sourceRow, ignoreMissing);
 }
 
+auto OutputAqlItemRow::fastForwardAllRows(InputAqlItemRow const& sourceRow, size_t rows)
+    -> void {
+#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+  TRI_ASSERT(sourceRow.internalBlockIs(_block));
+#endif
+  TRI_ASSERT(_doNotCopyInputRow);
+  TRI_ASSERT(_call.getLimit() >= rows);
+  TRI_ASSERT(rows > 0);
+  // We have the guarantee that we have all data in our block.
+  // We only need to adjust internal indexes.
+#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+  // Safely assert that the API is not missused.
+  TRI_ASSERT(_baseIndex + rows <= _block->size());
+  for (size_t i = _baseIndex; i < _baseIndex + rows; ++i) {
+    TRI_ASSERT(!_block->isShadowRow(i));
+  }
+#endif
+  _baseIndex += rows;
+  TRI_ASSERT(_baseIndex > 0);
+  _lastBaseIndex = _baseIndex - 1;
+  _lastSourceRow = InputAqlItemRow{_block, _lastBaseIndex};
+  _call.didProduce(rows);
+}
+
 void OutputAqlItemRow::copyBlockInternalRegister(InputAqlItemRow const& sourceRow,
                                                  RegisterId input, RegisterId output) {
   // This method is only allowed if the block of the input row is the same as