Feature/aql subquery operations stack aql item block input matrix (#1…

…1167) * added new file to cmakelist * added new method skipAllRemainingDataRows * added first implementation of execute - AllRowsFetcher * impl is now using skipAllRemainingDataRows method, also execute methods now using Fetcher::DataRange instead of inputRange * impl is now using skipAllRemainingDataRows method, also execute methods now using Fetcher::DataRange instead of inputRange - header file * AllRowsFetcher now using AqlItemBlockInputMatrix as DataRange, first implementation of execute * first implementation of AqlItemBlockInputMatrix * more cleanup * optimized execut elogic * optimized logic in inputmatrix, still a few todos left * added SortExecutor as newStyle * added produceRows and skipRows * logs (to be removed), skipAllRemainingRows * logs * inputMatrix * changed return type of skipAllRemainingDataRows * impl logs * logs, added clear method * sort test * typo, log * latest changes * cleanup, assert * rm logs * added more todos * todo * new method names, compile again * Reset ExecutorReturnedDone for SubqueryStart and SubqueryEnd. * InputMatrix needs to return state DONE on every input request. It guarantees that we have all input * Removed invalid Assertion, and removed early return if output is full. * Removed DebugLogging from CalculationTest * Fixed skipping in SortExecutor * Fixed upstreamState return of AqlItemMatrix. It is only finalState iff there is no data * Let the InputMatrix drop all rows after FullCount * Allow bypassing with hardLimit 0 in InputMatrix * Disabled the old-code assertion. This way the SortExecutor tests we have will be red, but not crash, allowing the jenkins to produce meaningfull results * Fix assert in WAITING case * Add rudimentary sort executor GTests * Fixed assertion on InputMatrix that starts right away with a shadowRow * Removed declared but not implemented function * changed from: peekDataRowAndState -> to: peekDataRow Co-authored-by: Michael Hackstein <michael@arangodb.com>
arangodb · mchacki · Mar 13, 2020 · Oct 22, 2019 · Oct 22, 2019 · Oct 22, 2019
commit a0d6ec193f02203e44e2270be0146d8aa5147b86
diff --git a/arangod/Aql/AllRowsFetcher.cpp b/arangod/Aql/AllRowsFetcher.cpp
@@ -21,6 +21,7 @@
 ////////////////////////////////////////////////////////////////////////////////
 
 #include "AllRowsFetcher.h"
+#include <Logger/LogMacros.h>
 
 #include "Aql/AqlItemBlock.h"
 #include "Aql/AqlItemMatrix.h"
@@ -59,6 +60,52 @@ std::pair<ExecutionState, AqlItemMatrix const*> AllRowsFetcher::fetchAllRows() {
   return {ExecutionState::DONE, nullptr};
 }
 
+std::tuple<ExecutionState, size_t, AqlItemBlockInputMatrix> AllRowsFetcher::execute(AqlCallStack& stack) {
+  if (!stack.isRelevant()) {
+    auto [state, skipped, block] = _dependencyProxy->execute(stack);
+    return {state, skipped, AqlItemBlockInputMatrix{block}};
+  }
+  TRI_ASSERT(stack.peek().getOffset() == 0);
+  TRI_ASSERT(!stack.peek().needsFullCount());
+  // We allow a 0 hardLimit for bypassing
+  // bot otherwise we do not allow any limit
+  TRI_ASSERT(!stack.peek().hasHardLimit() || stack.peek().getLimit() == 0);
+  TRI_ASSERT(!stack.peek().hasSoftLimit());
+
+  if (_aqlItemMatrix == nullptr) {
+    _aqlItemMatrix = std::make_unique<AqlItemMatrix>(getNrInputRegisters());
+  }
+  // We can only execute More if we are not Stopped yet.
+  TRI_ASSERT(!_aqlItemMatrix->stoppedOnShadowRow());
+  while (true) {
+    auto [state, skipped, block] = _dependencyProxy->execute(stack);
+    TRI_ASSERT(skipped == 0);
+
+    // we will either build a complete fetched AqlItemBlockInputMatrix or return an empty one
+    if (state == ExecutionState::WAITING) {
+      TRI_ASSERT(block == nullptr);
+      // On waiting we have nothing to return
+      return {state, 0, AqlItemBlockInputMatrix{ExecutorState::HASMORE}};
+    }
+    TRI_ASSERT(block != nullptr || state == ExecutionState::DONE);
+
+    if (block != nullptr) {
+      // we need to store the block for later creation of AqlItemBlockInputMatrix
+      _aqlItemMatrix->addBlock(std::move(block));
+    }
+
+    // If we find a ShadowRow or ExecutionState == Done, we're done fetching.
+    if (_aqlItemMatrix->stoppedOnShadowRow() || state == ExecutionState::DONE) {
+      if (state == ExecutionState::HASMORE) {
+        return {state, 0,
+                AqlItemBlockInputMatrix{ExecutorState::HASMORE, _aqlItemMatrix.get()}};
+      }
+      return {state, 0,
+              AqlItemBlockInputMatrix{ExecutorState::DONE, _aqlItemMatrix.get()}};
+    }
+  }
+}
+
 std::pair<ExecutionState, InputAqlItemRow> AllRowsFetcher::fetchRow(size_t atMost) {
   switch (_dataFetchedState) {
     case ALL_DATA_FETCHED:

diff --git a/arangod/Aql/AllRowsFetcher.h b/arangod/Aql/AllRowsFetcher.h
@@ -33,8 +33,7 @@
 #include <cstddef>
 #include <memory>
 
-// TODO REMOVE ME TEMPORARY
-#include "Aql/AqlItemBlockInputRange.h"
+#include "Aql/AqlItemBlockInputMatrix.h"
 
 namespace arangodb {
 namespace aql {
@@ -90,22 +89,28 @@ class AllRowsFetcher {
 
  public:
   explicit AllRowsFetcher(DependencyProxy<BlockPassthrough::Disable>& executionBlock);
-
   TEST_VIRTUAL ~AllRowsFetcher() = default;
 
-  // TODO FIXME, this Range does not work here.
-  using DataRange = AqlItemBlockInputRange;
+  using DataRange = AqlItemBlockInputMatrix;
 
  protected:
   // only for testing! Does not initialize _dependencyProxy!
   AllRowsFetcher() = default;
 
  public:
-  // TODO implement and document
-  std::tuple<ExecutionState, size_t, DataRange> execute(/* TODO: add"justDoIt"-style parameter */) {
-    TRI_ASSERT(false);
-    THROW_ARANGO_EXCEPTION(TRI_ERROR_NOT_IMPLEMENTED);
-  }
+  /**
+   * @brief Execute the given call stack
+   *
+   * @param stack Call stack, on top of stack there is current subquery, bottom is the main query.
+   * @return std::tuple<ExecutionState, size_t, DataRange>
+   *   ExecutionState => DONE, all queries are done, there will be no more
+   *   ExecutionState => HASMORE, there are more results for queries, might be on other subqueries
+   *   ExecutionState => WAITING, we need to do I/O to solve the request, save local state and return WAITING to caller immediately
+   *
+   *   size_t => Amount of documents skipped
+   *   DataRange => Resulting data
+   */
+  std::tuple<ExecutionState, size_t, DataRange> execute(AqlCallStack& stack);
 
   /**
    * @brief Fetch one new AqlItemRow from upstream.

diff --git a/arangod/Aql/AqlItemBlockInputMatrix.cpp b/arangod/Aql/AqlItemBlockInputMatrix.cpp
@@ -0,0 +1,142 @@
+////////////////////////////////////////////////////////////////////////////////
+/// 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 "AqlItemBlockInputMatrix.h"
+#include "Aql/ShadowAqlItemRow.h"
+
+#include <Logger/LogMacros.h>
+#include <velocypack/Builder.h>
+#include <velocypack/velocypack-aliases.h>
+#include <numeric>
+
+using namespace arangodb;
+using namespace arangodb::aql;
+
+AqlItemBlockInputMatrix::AqlItemBlockInputMatrix(ExecutorState state)
+    : _finalState{state}, _aqlItemMatrix{nullptr} {
+  TRI_ASSERT(_aqlItemMatrix == nullptr);
+  TRI_ASSERT(!hasDataRow());
+}
+
+// only used for block passthrough
+AqlItemBlockInputMatrix::AqlItemBlockInputMatrix(arangodb::aql::SharedAqlItemBlockPtr const& block)
+    : _block{block}, _aqlItemMatrix{nullptr} {
+  TRI_ASSERT(_aqlItemMatrix == nullptr);
+  TRI_ASSERT(!hasDataRow());
+}
+
+AqlItemBlockInputMatrix::AqlItemBlockInputMatrix(ExecutorState state, AqlItemMatrix* aqlItemMatrix)
+    : _finalState{state}, _aqlItemMatrix{aqlItemMatrix} {
+  TRI_ASSERT(_block == nullptr);
+  TRI_ASSERT(_aqlItemMatrix != nullptr);
+  if (_aqlItemMatrix->size() == 0 && _aqlItemMatrix->stoppedOnShadowRow()) {
+    // Fast forward to initialize the _shadowRow
+    skipAllRemainingDataRows();
+  }
+}
+
+SharedAqlItemBlockPtr AqlItemBlockInputMatrix::getBlock() const noexcept {
+  TRI_ASSERT(_aqlItemMatrix == nullptr);
+  return _block;
+}
+
+std::pair<ExecutorState, AqlItemMatrix const*> AqlItemBlockInputMatrix::getMatrix() noexcept {
+  TRI_ASSERT(_aqlItemMatrix != nullptr);
+  TRI_ASSERT(_block == nullptr);
+  TRI_ASSERT(!_shadowRow.isInitialized());
+
+  // We are always done. This InputMatrix
+  // guarantees that we have all data in our hand at once.
+  return {ExecutorState::DONE, _aqlItemMatrix};
+}
+
+ExecutorState AqlItemBlockInputMatrix::upstreamState() const noexcept {
+  if (hasDataRow() || hasShadowRow()) {
+    return ExecutorState::DONE;
+  }
+  return _finalState;
+}
+
+bool AqlItemBlockInputMatrix::upstreamHasMore() const noexcept {
+  return upstreamState() == ExecutorState::HASMORE;
+}
+
+bool AqlItemBlockInputMatrix::hasDataRow() const noexcept {
+  if (_aqlItemMatrix == nullptr) {
+    return false;
+  }
+  return (!_shadowRow.isInitialized() && _aqlItemMatrix->size() != 0);
+}
+
+std::pair<ExecutorState, ShadowAqlItemRow> AqlItemBlockInputMatrix::nextShadowRow() {
+  auto tmpShadowRow = _shadowRow;
+
+  if (_aqlItemMatrix->size() == 0 && _aqlItemMatrix->stoppedOnShadowRow() &&
+      !_aqlItemMatrix->peekShadowRow().isRelevant()) {
+    // next row will be a shadow row
+    _shadowRow = _aqlItemMatrix->popShadowRow();
+  } else {
+    _shadowRow = ShadowAqlItemRow{CreateInvalidShadowRowHint()};
+  }
+
+  auto state = ExecutorState::HASMORE;
+  if (_shadowRow.isInitialized()) {
+    TRI_ASSERT(!_shadowRow.isRelevant());
+    state = ExecutorState::HASMORE;
+  } else if (_aqlItemMatrix->stoppedOnShadowRow() &&
+             _aqlItemMatrix->peekShadowRow().isRelevant()) {
+    state = ExecutorState::DONE;
+  } else {
+    state = _finalState;
+  }
+
+  return {state, tmpShadowRow};
+}
+
+ShadowAqlItemRow AqlItemBlockInputMatrix::peekShadowRow() const {
+  return _shadowRow;
+}
+
+bool AqlItemBlockInputMatrix::hasShadowRow() const noexcept {
+  return _shadowRow.isInitialized();
+}
+
+void AqlItemBlockInputMatrix::skipAllRemainingDataRows() {
+  if (_aqlItemMatrix == nullptr) {
+    // Have not been initialized.
+    // We need to be called before.
+    TRI_ASSERT(!hasShadowRow());
+    TRI_ASSERT(!hasDataRow());
+    return;
+  }
+  if (!hasShadowRow()) {
+    if (_aqlItemMatrix->stoppedOnShadowRow()) {
+      _shadowRow = _aqlItemMatrix->popShadowRow();
+      TRI_ASSERT(_shadowRow.isRelevant());
+    } else {
+      TRI_ASSERT(_finalState == ExecutorState::DONE);
+      _aqlItemMatrix->clear();
+    }
+  }
+  // Else we did already skip once.
+  // nothing to do
+}
diff --git a/arangod/Aql/AqlItemBlockInputMatrix.h b/arangod/Aql/AqlItemBlockInputMatrix.h
@@ -0,0 +1,67 @@
+////////////////////////////////////////////////////////////////////////////////
+/// 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 ARANGOD_AQL_AQLITEMBLOCKMATRIXITERATOR_H
+#define ARANGOD_AQL_AQLITEMBLOCKMATRIXITERATOR_H
+
+#include "Aql/AqlItemMatrix.h"
+#include "Aql/ExecutionState.h"
+#include "Aql/InputAqlItemRow.h"
+#include "Aql/SharedAqlItemBlockPtr.h"
+
+namespace arangodb::aql {
+
+class ShadowAqlItemRow;
+
+class AqlItemBlockInputMatrix {
+ public:
+  explicit AqlItemBlockInputMatrix(ExecutorState state);
+
+  AqlItemBlockInputMatrix(arangodb::aql::SharedAqlItemBlockPtr const&);
+
+  AqlItemBlockInputMatrix(ExecutorState state, AqlItemMatrix* aqlItemMatrix);
+
+  std::pair<ExecutorState, ShadowAqlItemRow> nextShadowRow();
+  ShadowAqlItemRow peekShadowRow() const;
+  bool hasShadowRow() const noexcept;
+  bool hasDataRow() const noexcept;
+
+  arangodb::aql::SharedAqlItemBlockPtr getBlock() const noexcept;
+  std::pair<ExecutorState, AqlItemMatrix const*> getMatrix() noexcept;
+
+  ExecutorState upstreamState() const noexcept;
+  bool upstreamHasMore() const noexcept;
+  void skipAllRemainingDataRows();
+
+ private:
+  arangodb::aql::SharedAqlItemBlockPtr _block{nullptr};
+  ExecutorState _finalState{ExecutorState::HASMORE};
+
+  // Only if _aqlItemMatrix is set (and NOT a nullptr), we have a valid and usable
+  // DataRange object available to work with.
+  AqlItemMatrix* _aqlItemMatrix;
+  ShadowAqlItemRow _shadowRow{CreateInvalidShadowRowHint{}};
+};
+
+}  // namespace arangodb::aql
+
+#endif  // ARANGOD_AQL_AQLITEMBLOCKINPUTITERATOR_H
diff --git a/arangod/Aql/AqlItemBlockInputRange.cpp b/arangod/Aql/AqlItemBlockInputRange.cpp
@@ -57,6 +57,7 @@ bool AqlItemBlockInputRange::hasDataRow() const noexcept {
   return isIndexValid(_rowIndex) && !isShadowRowAtIndex(_rowIndex);
 }
 
+// TODO: Implement peekDataRow (without state). e.g. IResearchViewExecutor does not need the state!
 std::pair<ExecutorState, InputAqlItemRow> AqlItemBlockInputRange::peekDataRow() const {
   if (hasDataRow()) {
     return std::make_pair(nextState<LookAhead::NEXT, RowType::DATA>(),
@@ -95,7 +96,14 @@ bool AqlItemBlockInputRange::isShadowRowAtIndex(std::size_t index) const noexcep
   return _block->isShadowRow(index);
 }
 
-std::pair<ExecutorState, ShadowAqlItemRow> AqlItemBlockInputRange::peekShadowRow() const {
+arangodb::aql::ShadowAqlItemRow AqlItemBlockInputRange::peekShadowRow() const {
+  if (hasShadowRow()) {
+    return ShadowAqlItemRow{_block, _rowIndex};
+  }
+  return ShadowAqlItemRow{CreateInvalidShadowRowHint{}};
+}
+
+std::pair<ExecutorState, ShadowAqlItemRow> AqlItemBlockInputRange::peekShadowRowAndState() const {
   if (hasShadowRow()) {
     return std::make_pair(nextState<LookAhead::NEXT, RowType::SHADOW>(),
                           ShadowAqlItemRow{_block, _rowIndex});
@@ -105,14 +113,24 @@ std::pair<ExecutorState, ShadowAqlItemRow> AqlItemBlockInputRange::peekShadowRow
 }
 
 std::pair<ExecutorState, ShadowAqlItemRow> AqlItemBlockInputRange::nextShadowRow() {
-  auto res = peekShadowRow();
+  auto res = peekShadowRowAndState();
   if (res.second.isInitialized()) {
     // Advance the current row.
     _rowIndex++;
   }
   return res;
 }
 
+void AqlItemBlockInputRange::skipAllRemainingDataRows() {
+  ExecutorState state;
+  InputAqlItemRow row{CreateInvalidInputRowHint{}};
+
+  while (hasDataRow()) {
+    std::tie(state, row) = nextDataRow();
+    TRI_ASSERT(row.isInitialized());
+  }
+}
+
 template <AqlItemBlockInputRange::LookAhead doPeek, AqlItemBlockInputRange::RowType type>
 ExecutorState AqlItemBlockInputRange::nextState() const noexcept {
   size_t testRowIndex = _rowIndex;

diff --git a/arangod/Aql/AqlItemBlockInputRange.h b/arangod/Aql/AqlItemBlockInputRange.h
@@ -57,10 +57,13 @@ class AqlItemBlockInputRange {
 
   bool hasShadowRow() const noexcept;
 
-  std::pair<ExecutorState, arangodb::aql::ShadowAqlItemRow> peekShadowRow() const;
+  arangodb::aql::ShadowAqlItemRow peekShadowRow() const;
+  std::pair<ExecutorState, arangodb::aql::ShadowAqlItemRow> peekShadowRowAndState() const;
 
   std::pair<ExecutorState, arangodb::aql::ShadowAqlItemRow> nextShadowRow();
 
+  void skipAllRemainingDataRows();
+
   // Subtract up to this many rows from the local `_skipped` state; return
   // the number actually skipped. Does not skip data rows.
   [[nodiscard]] auto skip(std::size_t) noexcept -> std::size_t;