arangodb · mchacki · Apr 2, 2020 · Oct 22, 2019 · Oct 22, 2019 · Oct 22, 2019
diff --git a/arangod/Aql/ExecutionBlockImpl.cpp b/arangod/Aql/ExecutionBlockImpl.cpp
@@ -135,7 +135,7 @@ constexpr bool is_one_of_v = (std::is_same_v<T, Es> || ...);
  */
 template <typename Executor>
 constexpr bool isNewStyleExecutor = is_one_of_v<
-    Executor, FilterExecutor, SortedCollectExecutor, IdExecutor<ConstFetcher>,
+    Executor, FilterExecutor, SortedCollectExecutor, IdExecutor<ConstFetcher>, ParallelUnsortedGatherExecutor,
     IdExecutor<SingleRowFetcher<BlockPassthrough::Enable>>, ReturnExecutor, DistinctCollectExecutor, IndexExecutor,
     EnumerateCollectionExecutor, SubqueryExecutor<true>, SubqueryExecutor<false>, CountCollectExecutor,
     CalculationExecutor<CalculationType::Condition>, CalculationExecutor<CalculationType::Reference>,
@@ -865,7 +865,7 @@ static SkipRowsRangeVariant constexpr skipRowsType() {
   static_assert(
       useExecutor ==
           (is_one_of_v<
-              Executor, FilterExecutor, ShortestPathExecutor, ReturnExecutor, KShortestPathsExecutor,
+              Executor, FilterExecutor, ShortestPathExecutor, ReturnExecutor, KShortestPathsExecutor, ParallelUnsortedGatherExecutor,
               IdExecutor<SingleRowFetcher<BlockPassthrough::Enable>>, IdExecutor<ConstFetcher>,
               HashedCollectExecutor, IndexExecutor, EnumerateCollectionExecutor, DistinctCollectExecutor,
               ConstrainedSortExecutor, CountCollectExecutor, SubqueryExecutor<true>,
@@ -990,7 +990,7 @@ auto ExecutionBlockImpl<Executor>::executeProduceRows(typename Fetcher::DataRang
                                                       OutputAqlItemRow& output)
     -> std::tuple<ExecutorState, typename Executor::Stats, AqlCall, size_t> {
   if constexpr (isNewStyleExecutor<Executor>) {
-    if constexpr (is_one_of_v<Executor, UnsortedGatherExecutor, SortingGatherExecutor>) {
+    if constexpr (is_one_of_v<Executor, UnsortedGatherExecutor, SortingGatherExecutor, ParallelUnsortedGatherExecutor>) {
       return _executor.produceRows(input, output);
     } else if constexpr (is_one_of_v<Executor, SubqueryExecutor<true>, SubqueryExecutor<false>>) {
       // The SubqueryExecutor has it's own special handling outside.
@@ -1013,7 +1013,7 @@ auto ExecutionBlockImpl<Executor>::executeSkipRowsRange(typename Fetcher::DataRa
   if constexpr (isNewStyleExecutor<Executor>) {
     call.skippedRows = 0;
     if constexpr (skipRowsType<Executor>() == SkipRowsRangeVariant::EXECUTOR) {
-      if constexpr (is_one_of_v<Executor, UnsortedGatherExecutor, SortingGatherExecutor>) {
+      if constexpr (is_one_of_v<Executor, UnsortedGatherExecutor, SortingGatherExecutor, ParallelUnsortedGatherExecutor>) {
         // If the executor has a method skipRowsRange, to skip outputs.
         // Every non-passthrough executor needs to implement this.
         auto res = _executor.skipRowsRange(inputRange, call);

diff --git a/arangod/Aql/MultiAqlItemBlockInputRange.cpp b/arangod/Aql/MultiAqlItemBlockInputRange.cpp
@@ -65,6 +65,11 @@ auto MultiAqlItemBlockInputRange::hasDataRow() const noexcept -> bool {
                      });
 }
 
+auto MultiAqlItemBlockInputRange::rangeForDependency(size_t const dependency)
+    -> AqlItemBlockInputRange& {
+  return _inputs.at(dependency);
+}
+
 auto MultiAqlItemBlockInputRange::peekDataRow(size_t const dependency) const
     -> std::pair<ExecutorState, arangodb::aql::InputAqlItemRow> {
   TRI_ASSERT(dependency < _inputs.size());

diff --git a/arangod/Aql/MultiAqlItemBlockInputRange.h b/arangod/Aql/MultiAqlItemBlockInputRange.h
@@ -49,6 +49,15 @@ class MultiAqlItemBlockInputRange {
   bool hasDataRow() const noexcept;
   bool hasDataRow(size_t const dependency) const noexcept;
 
+  /**
+   * @brief Get a reference to the range of a given dependency
+   * NOTE: Modifing this range will modify the state of this class as well
+   *
+   * @param dependency index of the dependency
+   * @return AqlItemBlockInputRange& Modifyable reference to the input data stream
+   */
+  auto rangeForDependency(size_t const dependency) -> AqlItemBlockInputRange&;
+
   std::pair<ExecutorState, arangodb::aql::InputAqlItemRow> peekDataRow(size_t const dependency) const;
   std::pair<ExecutorState, arangodb::aql::InputAqlItemRow> nextDataRow(size_t const dependency);
   auto skipAll(size_t const dependency) noexcept -> std::size_t;

diff --git a/arangod/Aql/ParallelUnsortedGatherExecutor.cpp b/arangod/Aql/ParallelUnsortedGatherExecutor.cpp
@@ -39,11 +39,15 @@ ParallelUnsortedGatherExecutorInfos::ParallelUnsortedGatherExecutorInfos(
                     nrInOutRegisters, nrInOutRegisters,
                     std::move(registersToClear), std::move(registersToKeep)) {}
 
-ParallelUnsortedGatherExecutor::ParallelUnsortedGatherExecutor(Fetcher& fetcher, Infos& infos)
-    : _fetcher(fetcher), _numberDependencies(0), _currentDependency(0), _skipped(0) {}
+ParallelUnsortedGatherExecutor::ParallelUnsortedGatherExecutor(Fetcher&, Infos& infos) {}
 
 ParallelUnsortedGatherExecutor::~ParallelUnsortedGatherExecutor() = default;
 
+auto ParallelUnsortedGatherExecutor::upstreamCall(AqlCall const& clientCall) const
+    noexcept -> AqlCall {
+  return clientCall;
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief Guarantees requiredby this this block:
 ///        1) For every dependency the input is sorted, according to the same strategy.
@@ -58,117 +62,74 @@ ParallelUnsortedGatherExecutor::~ParallelUnsortedGatherExecutor() = default;
 ///
 ////////////////////////////////////////////////////////////////////////////////
 
-std::pair<ExecutionState, NoStats> ParallelUnsortedGatherExecutor::produceRows(OutputAqlItemRow& output) {
-  initDependencies();
-
-  ExecutionState state;
-  InputAqlItemRow inputRow = InputAqlItemRow{CreateInvalidInputRowHint{}};
-
-  size_t x;
-  for (x = 0; x < _numberDependencies; ++x) {
-    size_t i = (_currentDependency + x) % _numberDependencies;
-
-    if (_upstream[i] == ExecutionState::DONE) {
-      continue;
-    }
-
-    size_t tmp = 0;
-
-    state = ExecutionState::HASMORE;
-    while (!output.isFull() && state == ExecutionState::HASMORE) {
-      std::tie(state, inputRow) = _fetcher.fetchRowForDependency(i, output.numRowsLeft() /*atMost*/);
-      if (inputRow) {
-        output.copyRow(inputRow);
-        TRI_ASSERT(output.produced());
+auto ParallelUnsortedGatherExecutor::produceRows(typename Fetcher::DataRange& input,
+                                                 OutputAqlItemRow& output)
+    -> std::tuple<ExecutorState, Stats, AqlCall, size_t> {
+  // Illegal dependency, on purpose to trigger asserts
+  size_t waitingDep = input.numberDependencies();
+  for (size_t dep = 0; dep < input.numberDependencies(); ++dep) {
+    while (!output.isFull()) {
+      auto [state, row] = input.nextDataRow(dep);
+      if (row) {
+        output.copyRow(row);
         output.advanceRow();
-        tmp++;
+      } else {
+        // This output did not produce anything
+        if (state == ExecutorState::HASMORE) {
+          waitingDep = dep;
+        }
+        break;
       }
     }
-
-    _upstream[i] = state;
-    if (output.isFull()) {
-      break;
-    }
-  }
-  _currentDependency = x;
-
-  NoStats stats;
-
-  // fix assert in ExecutionBlockImpl<E>::getSomeWithoutTrace
-  if (output.isFull()) {
-    return {ExecutionState::HASMORE, stats};
-  }
-
-  size_t numWaiting = 0;
-  for (x = 0; x < _numberDependencies; ++x) {
    if (_upstream[x] == ExecutionState::HASMORE) {
-      return {ExecutionState::HASMORE, stats};
-    } else if (_upstream[x] == ExecutionState::WAITING) {
-      numWaiting++;
-    }
   }
-  if (numWaiting > 0) {
-    return {ExecutionState::WAITING, stats};
+  if (input.isDone()) {
+    // We cannot have one that we are waiting on, if we are done.
+    TRI_ASSERT(waitingDep == input.numberDependencies());
+    return {ExecutorState::DONE, NoStats{}, AqlCall{}, waitingDep};
   }
-
-  TRI_ASSERT(std::all_of(_upstream.begin(), _upstream.end(), [](auto const& s) { return s == ExecutionState::DONE; } ));
-  return {ExecutionState::DONE, stats};
+  return {ExecutorState::HASMORE, NoStats{}, upstreamCall(output.getClientCall()), waitingDep};
 }
 
-std::tuple<ExecutionState, ParallelUnsortedGatherExecutor::Stats, size_t>
-ParallelUnsortedGatherExecutor::skipRows(size_t const toSkip) {
-  initDependencies();
-  TRI_ASSERT(_skipped <= toSkip);
-
-  ExecutionState state = ExecutionState::HASMORE;
-  while (_skipped < toSkip) {
-
-    const size_t i = _currentDependency;
-    if (_upstream[i] == ExecutionState::DONE) {
-      if (std::all_of(_upstream.begin(), _upstream.end(),
-                      [](auto s) { return s == ExecutionState::DONE; })) {
-         state = ExecutionState::DONE;
-         break;
-       }
-      _currentDependency = (i + 1) % _numberDependencies;
-       continue;
-     }
-
-    TRI_ASSERT(_skipped <= toSkip);
+auto ParallelUnsortedGatherExecutor::skipRowsRange(typename Fetcher::DataRange& input,
+                                                   AqlCall& call)
+    -> std::tuple<ExecutorState, Stats, size_t, AqlCall, size_t> {
+  size_t waitingDep = input.numberDependencies();
+  for (size_t dep = 0; dep < input.numberDependencies(); ++dep) {
+    auto& range = input.rangeForDependency(dep);
+    while (call.needSkipMore()) {
+      if (!range.hasDataRow() && range.skippedInFlight() == 0) {
+        // Consumed this range,
+        // consume the next one
 
-    size_t skippedNow;
-    std::tie(state, skippedNow) = _fetcher.skipRowsForDependency(i, toSkip - _skipped);
-    _upstream[i] = state;
-    if (state == ExecutionState::WAITING) {
-      TRI_ASSERT(skippedNow == 0);
-      return {ExecutionState::WAITING, NoStats{}, 0};
-    }
-    _skipped += skippedNow;
-
-    if (_upstream[i] == ExecutionState::DONE) {
-      if (std::all_of(_upstream.begin(), _upstream.end(),
-                      [](auto s) { return s == ExecutionState::DONE; })) {
+        // Guarantee:
+        // While in offsetPhase, we will only send requests to the first
+        // NON-DONE dependency.
+        if (range.upstreamState() == ExecutorState::HASMORE &&
+            waitingDep == input.numberDependencies()) {
+          waitingDep = dep;
+        }
         break;
       }
-       _currentDependency = (i + 1) % _numberDependencies;
-       continue;
-     }
+      if (range.hasDataRow()) {
+        // We overfetched, skipLocally
+        // By gurantee we will only see data, if
+        // we are past the offset phase.
+        TRI_ASSERT(call.getOffset() == 0);
+      } else {
+        if (call.getOffset() > 0) {
+          call.didSkip(range.skip(call.getOffset()));
+        } else {
+          // Fullcount Case
+          call.didSkip(range.skipAll());
+        }
+      }
+    }
   }
-
-  size_t skipped = _skipped;
-  _skipped = 0;
-
-  TRI_ASSERT(skipped <= toSkip);
-  return {state, NoStats{}, skipped};
-}
-
-void ParallelUnsortedGatherExecutor::initDependencies() {
-  if (_numberDependencies == 0) {
-    // We need to initialize the dependencies once, they are injected
-    // after the fetcher is created.
-    _numberDependencies = _fetcher.numberDependencies();
-    TRI_ASSERT(_numberDependencies > 0);
-    _upstream.resize(_numberDependencies, ExecutionState::HASMORE);
-    TRI_ASSERT(std::all_of(_upstream.begin(), _upstream.end(), [](auto const& s) { return s == ExecutionState::HASMORE; } ));
+  if (input.isDone()) {
+    // We cannot have one that we are waiting on, if we are done.
+    TRI_ASSERT(waitingDep == input.numberDependencies());
+    return {ExecutorState::DONE, NoStats{}, call.getSkipCount(), AqlCall{}, waitingDep};
   }
+  return {ExecutorState::HASMORE, NoStats{}, call.getSkipCount(),
+          upstreamCall(call), waitingDep};
 }
diff --git a/arangod/Aql/ParallelUnsortedGatherExecutor.h b/arangod/Aql/ParallelUnsortedGatherExecutor.h
@@ -37,10 +37,11 @@ class Methods;
 
 namespace aql {
 
+struct AqlCall;
+class MultiAqlItemBlockInputRange;
 class MultiDependencySingleRowFetcher;
 class NoStats;
 class OutputAqlItemRow;
-struct SortRegister;
 
 class ParallelUnsortedGatherExecutorInfos : public ExecutorInfos {
  public:
@@ -55,7 +56,6 @@ class ParallelUnsortedGatherExecutorInfos : public ExecutorInfos {
 
 class ParallelUnsortedGatherExecutor {
  public:
-
  public:
   struct Properties {
     static constexpr bool preservesOrder = true;
@@ -71,31 +71,36 @@ class ParallelUnsortedGatherExecutor {
   ~ParallelUnsortedGatherExecutor();
 
   /**
-   * @brief produce the next Row of Aql Values.
+   * @brief Produce rows
    *
-   * @return ExecutionState,
-   *         if something was written output.hasValue() == true
+   * @param input DataRange delivered by the fetcher
+   * @param output place to write rows to
+   * @return std::tuple<ExecutorState, Stats, AqlCall, size_t>
+   *   ExecutorState: DONE or HASMORE (only within a subquery)
+   *   Stats: Stats gerenated here
+   *   AqlCall: Request to upstream
+   *   size:t: Dependency to request
    */
-  std::pair<ExecutionState, Stats> produceRows(OutputAqlItemRow& output);
+  [[nodiscard]] auto produceRows(MultiAqlItemBlockInputRange& input, OutputAqlItemRow& output)
+      -> std::tuple<ExecutorState, Stats, AqlCall, size_t>;
 
-  std::tuple<ExecutionState, Stats, size_t> skipRows(size_t atMost);
-
- private:
-
-  void initDependencies();
-
- private:
-  Fetcher& _fetcher;
-  // 64: default size of buffer; 8: Alignment size; computed to 4 but breaks in windows debug build.
-  ::arangodb::containers::SmallVector<ExecutionState, 64, 8>::allocator_type::arena_type _arena;
-  ::arangodb::containers::SmallVector<ExecutionState, 64, 8> _upstream{_arena};
+  /**
+   * @brief Skip rows
+   *
+   * @param input DataRange delivered by the fetcher
+   * @param call skip request form consumer
+   * @return std::tuple<ExecutorState, Stats, AqlCall, size_t>
+   *   ExecutorState: DONE or HASMORE (only within a subquery)
+   *   Stats: Stats gerenated here
+   *   size_t: Number of rows skipped
+   *   AqlCall: Request to upstream
+   *   size:t: Dependency to request
+   */
+  [[nodiscard]] auto skipRowsRange(MultiAqlItemBlockInputRange& input, AqlCall& call)
+      -> std::tuple<ExecutorState, Stats, size_t, AqlCall, size_t>;
 
-  // Total Number of dependencies
-  size_t _numberDependencies;
-
-  size_t _currentDependency;
-
-  size_t _skipped;
+ private:
+  auto upstreamCall(AqlCall const& clientCall) const noexcept -> AqlCall;
 };
 
 }  // namespace aql