Merge branch 'feature/AqlSubqueryExecutionBlockImplExecuteImplementat…

…ion' of ssh://github.com/arangodb/ArangoDB into feature/AqlSubqueryExecutionBlockImplExecuteImplementation-subqueryStartExecutor
arangodb · mchacki · Feb 29, 2020 · Jan 29, 2020 · Jan 31, 2020 · Jan 31, 2020
commit 81443f6556ed1b9411e73a58f56ae9a9bc173815
diff --git a/arangod/Aql/ExecutionBlockImpl.cpp b/arangod/Aql/ExecutionBlockImpl.cpp
@@ -141,7 +141,7 @@ constexpr bool isNewStyleExecutor =
                 TestLambdaExecutor,
                 TestLambdaSkipExecutor,  // we need one after these to avoid compile errors in non-test mode
 #endif
-                SubqueryStartExecutor, SubqueryEndExecutor, ShortestPathExecutor, EnumerateListExecutor>;
+                SubqueryStartExecutor, SubqueryEndExecutor, ShortestPathExecutor, EnumerateListExecutor, LimitExecutor>;
 
 template <class Executor>
 ExecutionBlockImpl<Executor>::ExecutionBlockImpl(ExecutionEngine* engine,
@@ -1086,7 +1086,8 @@ static SkipRowsRangeVariant constexpr skipRowsType() {
 #ifdef ARANGODB_USE_GOOGLE_TESTS
                                  TestLambdaSkipExecutor,
 #endif
-                                 EnumerateListExecutor, SubqueryStartExecutor, SubqueryEndExecutor, SortedCollectExecutor>),
+                                 EnumerateListExecutor, SubqueryStartExecutor, SubqueryEndExecutor,
+                                 SortedCollectExecutor, LimitExecutor>),
                 "Unexpected executor for SkipVariants::EXECUTOR");
 
   // The LimitExecutor will not work correctly with SkipVariants::FETCHER!
@@ -1146,13 +1147,15 @@ auto ExecutionBlockImpl<SubqueryStartExecutor>::shadowRowForwarding() -> ExecSta
   TRI_ASSERT(_outputItemRow);
   TRI_ASSERT(_outputItemRow->isInitialized());
   TRI_ASSERT(_outputItemRow->allRowsUsed());
+  THROW_ARANGO_EXCEPTION(TRI_ERROR_NOT_IMPLEMENTED);
 }
 
 template <>
 auto ExecutionBlockImpl<SubqueryEndExecutor>::shadowRowForwarding() -> ExecState {
   TRI_ASSERT(_outputItemRow);
   TRI_ASSERT(_outputItemRow->isInitialized());
   TRI_ASSERT(_outputItemRow->allRowsUsed());
+  THROW_ARANGO_EXCEPTION(TRI_ERROR_NOT_IMPLEMENTED);
 }
 
 template <class Executor>
@@ -1485,6 +1488,9 @@ ExecutionBlockImpl<Executor>::executeWithoutTrace(AqlCallStack stack) {
           TRI_ASSERT(!_lastRange.hasShadowRow());
           size_t skippedLocal = 0;
 
+#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+          size_t subqueryLevelBefore = stack.subqueryLevel();
+#endif
           // If we are SubqueryStart, we remove the top element of the stack
           // which belongs to the subquery enclosed by this
           // SubqueryStart and the partnered SubqueryEnd by *not*
@@ -1512,7 +1518,6 @@ ExecutionBlockImpl<Executor>::executeWithoutTrace(AqlCallStack stack) {
               }
             }
           }
-
           std::tie(_upstreamState, skippedLocal, _lastRange) = _rowFetcher.execute(stack);
           // As the stack is copied into the fetcher, we need to pop off our call again.
           // If we use other datastructures or moving we may hand over ownership of the stack here

diff --git a/arangod/Aql/ExecutionBlockImpl.h b/arangod/Aql/ExecutionBlockImpl.h
@@ -289,7 +289,9 @@ class ExecutionBlockImpl final : public ExecutionBlock {
   // Executor is done, we need to handle ShadowRows of subqueries.
   // In most executors they are simply copied, in subquery executors
   // there needs to be actions applied here.
-  [[nodsicard]] auto shadowRowForwarding() -> ExecState;
+  [[nodiscard]] auto shadowRowForwarding() -> ExecState;
+
+  [[nodiscard]] auto outputIsFull() const noexcept -> bool;
 
  private:
   /**

diff --git a/tests/Aql/ExecutorTestHelper.h b/tests/Aql/ExecutorTestHelper.h
@@ -36,6 +36,7 @@
 #include "Aql/ExecutionBlock.h"
 #include "Aql/ExecutionBlockImpl.h"
 #include "Aql/ExecutionEngine.h"
+#include "Aql/ExecutionNode.h"
 #include "Aql/ExecutionState.h"
 #include "Aql/ExecutionStats.h"
 #include "Aql/OutputAqlItemRow.h"
@@ -291,20 +292,32 @@ struct ExecutorTestHelper {
     return *this;
   }
 
+  /**
+   * @brief Set the Execution Block object
+   *
+   * @tparam E The executor
+   * @param infos to build the executor
+   * @param nodeType The type of executor node, only used for debug printing, defaults to SINGLETON
+   * @return ExecutorTestHelper&
+   */
   template <typename E>
-  auto setExecBlock(typename E::Infos infos) -> ExecutorTestHelper& {
+  auto setExecBlock(typename E::Infos infos,
+                    ExecutionNode::NodeType nodeType = ExecutionNode::SINGLETON)
+      -> ExecutorTestHelper& {
     auto& testeeNode = _execNodes.emplace_back(std::move(
-        std::make_unique<SingletonNode>(_query.plan(), _execNodes.size())));
+        std::make_unique<MockTypedNode>(_query.plan(), _execNodes.size(), nodeType)));
     setPipeline(Pipeline{
         std::make_unique<ExecutionBlockImpl<E>>(_query.engine(), testeeNode.get(),
                                                 std::move(infos))});
     return *this;
   }
 
   template <typename E>
-  auto createExecBlock(typename E::Infos infos) -> ExecBlock {
+  auto createExecBlock(typename E::Infos infos,
+                       ExecutionNode::NodeType nodeType = ExecutionNode::SINGLETON)
+      -> ExecBlock {
     auto& testeeNode = _execNodes.emplace_back(std::move(
-        std::make_unique<SingletonNode>(_query.plan(), _execNodes.size())));
+        std::make_unique<MockTypedNode>(_query.plan(), _execNodes.size(), nodeType)));
     return std::make_unique<ExecutionBlockImpl<E>>(_query.engine(), testeeNode.get(),
                                                    std::move(infos));
   }
@@ -334,29 +347,48 @@ struct ExecutorTestHelper {
     return *this;
   }
 
-  auto run() -> void {
+  auto run(bool const loop = false) -> void {
     ResourceMonitor monitor;
     AqlItemBlockManager itemBlockManager(&monitor, SerializationFormat::SHADOWROWS);
 
     auto inputBlock = generateInputRanges(itemBlockManager);
 
+    auto skippedTotal = size_t{0};
+    auto finalState = ExecutionState::HASMORE;
+
     TRI_ASSERT(!_pipeline.empty());
     _pipeline.get().back()->addDependency(inputBlock.get());
 
-    AqlCallStack stack{_call};
+    BlockCollector allResults{&itemBlockManager};
 
-    ExecutionState state;
-    size_t skipped;
-    SharedAqlItemBlockPtr result;
-    do {
-      std::tie(state, skipped, result) = _pipeline.get().front()->execute(stack);
+    if (!loop) {
+      AqlCallStack stack{_call};
+      auto const [state, skipped, result] = _pipeline.get().front()->execute(stack);
+      skippedTotal = skipped;
+      finalState = state;
       if (result != nullptr) {
-        LOG_DEVEL << *result;  // .slice().toJson();
+        allResults.add(result);
       }
-    } while (state != ExecutionState::DONE);
+    } else {
+      auto call = _call;
+      do {
+        AqlCallStack stack{call};
+        auto const [state, skipped, result] = _pipeline.get().front()->execute(stack);
+        finalState = state;
+        skippedTotal += skipped;
+        if (result != nullptr) {
+          allResults.add(result);
+        }
+        call = _call;
+        call.didSkip(skippedTotal);
+        call.didProduce(allResults.totalSize());
+      } while (finalState != ExecutionState::DONE &&
+               (!call.hasSoftLimit() || (call.getLimit() + call.getOffset()) > 0));
+    }
 
-    EXPECT_EQ(skipped, _expectedSkip);
-    EXPECT_EQ(state, _expectedState);
+    EXPECT_EQ(skippedTotal, _expectedSkip);
+    EXPECT_EQ(finalState, _expectedState);
+    SharedAqlItemBlockPtr result = allResults.steal();
     if (result == nullptr) {
       // Empty output, possible if we skip all
       EXPECT_EQ(_output.size(), 0)
@@ -459,7 +491,7 @@ struct ExecutorTestHelper {
   arangodb::aql::Query& _query;
   std::unique_ptr<arangodb::aql::ExecutionNode> _dummyNode;
   Pipeline _pipeline;
-  std::vector<std::unique_ptr<ExecutionNode>> _execNodes;
+  std::vector<std::unique_ptr<MockTypedNode>> _execNodes;
 };
 
 enum class ExecutorCall {

diff --git a/tests/Aql/ReturnExecutorTest.cpp b/tests/Aql/ReturnExecutorTest.cpp
@@ -99,7 +99,7 @@ TEST_P(ReturnExecutorTest, returns_all_from_upstream) {
   ReturnExecutorInfos infos(0 /*input register*/, 1 /*nr in*/, 1 /*nr out*/, doCount());
   AqlCall call{};  // unlimited produce
   ExecutorTestHelper(*fakedQuery)
-      .setExecBlock<ReturnExecutor>(std::move(infos))
+      .setExecBlock<ReturnExecutor>(std::move(infos), ExecutionNode::RETURN)
       .setInputValueList(1, 2, 5, 2, 1, 5, 7, 1)
       .setInputSplitType(getSplit())
       .setCall(call)
@@ -115,7 +115,7 @@ TEST_P(ReturnExecutorTest, handle_soft_limit) {
   AqlCall call{};
   call.softLimit = 3;
   ExecutorTestHelper(*fakedQuery)
-      .setExecBlock<ReturnExecutor>(std::move(infos))
+      .setExecBlock<ReturnExecutor>(std::move(infos), ExecutionNode::RETURN)
       .setInputValueList(1, 2, 5, 2, 1, 5, 7, 1)
       .setInputSplitType(getSplit())
       .setCall(call)
@@ -131,7 +131,7 @@ TEST_P(ReturnExecutorTest, handle_hard_limit) {
   AqlCall call{};
   call.hardLimit = 5;
   ExecutorTestHelper(*fakedQuery)
-      .setExecBlock<ReturnExecutor>(std::move(infos))
+      .setExecBlock<ReturnExecutor>(std::move(infos), ExecutionNode::RETURN)
       .setInputValueList(1, 2, 5, 2, 1, 5, 7, 1)
       .setInputSplitType(getSplit())
       .setCall(call)
@@ -147,7 +147,7 @@ TEST_P(ReturnExecutorTest, handle_offset) {
   AqlCall call{};
   call.offset = 4;
   ExecutorTestHelper(*fakedQuery)
-      .setExecBlock<ReturnExecutor>(std::move(infos))
+      .setExecBlock<ReturnExecutor>(std::move(infos), ExecutionNode::RETURN)
       .setInputValueList(1, 2, 5, 2, 1, 5, 7, 1)
       .setInputSplitType(getSplit())
       .setCall(call)
@@ -164,7 +164,7 @@ TEST_P(ReturnExecutorTest, handle_fullcount) {
   call.hardLimit = 2;
   call.fullCount = true;
   ExecutorTestHelper(*fakedQuery)
-      .setExecBlock<ReturnExecutor>(std::move(infos))
+      .setExecBlock<ReturnExecutor>(std::move(infos), ExecutionNode::RETURN)
       .setInputValueList(1, 2, 5, 2, 1, 5, 7, 1)
       .setInputSplitType(getSplit())
       .setCall(call)
@@ -180,7 +180,7 @@ TEST_P(ReturnExecutorTest, handle_other_inputRegister) {
   AqlCall call{};
   call.hardLimit = 5;
   ExecutorTestHelper<2, 1>(*fakedQuery)
-      .setExecBlock<ReturnExecutor>(std::move(infos))
+      .setExecBlock<ReturnExecutor>(std::move(infos), ExecutionNode::RETURN)
       .setInputValue({{R"("invalid")", 1},
                       {R"("invalid")", 2},
                       {R"("invalid")", 5},

diff --git a/tests/Aql/SubqueryStartExecutorTest.cpp b/tests/Aql/SubqueryStartExecutorTest.cpp
@@ -114,8 +114,7 @@ TEST_F(SubqueryStartExecutorTest, adds_a_shadowrow_after_single_input) {
   auto state = ExecutorState::HASMORE;
 
   auto inputBlock = buildBlock<1>(itemBlockManager, MatrixBuilder<1>{{{{R"("a")"}}}});
-  auto input =
-      AqlItemBlockInputRange{ExecutorState::DONE, inputBlock, 0, inputBlock->size()};
+  auto input = AqlItemBlockInputRange{ExecutorState::DONE, 0, inputBlock, 0};
 
   OutputAqlItemRow output{std::move(block), infos.getOutputRegisters(),
                           infos.registersToKeep(), infos.registersToClear()};
@@ -141,8 +140,7 @@ TEST_F(SubqueryStartExecutorTest, adds_a_shadowrow_after_every_input_line_in_sin
   auto inputBlock =
       buildBlock<1>(itemBlockManager,
                     MatrixBuilder<1>{{{{{R"("a")"}}}, {{{R"("b")"}}}, {{R"("c")"}}}});
-  auto input =
-      AqlItemBlockInputRange{ExecutorState::DONE, inputBlock, 0, inputBlock->size()};
+  auto input = AqlItemBlockInputRange{ExecutorState::DONE, 0, inputBlock, 0};
 
   SharedAqlItemBlockPtr block = itemBlockManager.requestBlock(1000, 1);
   OutputAqlItemRow output{std::move(block), infos.getOutputRegisters(),
@@ -168,8 +166,7 @@ TEST_F(SubqueryStartExecutorTest, shadow_row_does_not_fit_in_current_block) {
   auto inputBlock =
       buildBlock<1>(itemBlockManager,
                     MatrixBuilder<1>{{{{{R"("a")"}}}, {{{R"("b")"}}}, {{R"("c")"}}}});
-  auto input =
-      AqlItemBlockInputRange{ExecutorState::DONE, inputBlock, 0, inputBlock->size()};
+  auto input = AqlItemBlockInputRange{ExecutorState::DONE, 0, inputBlock, 0};
 
   SharedAqlItemBlockPtr block = itemBlockManager.requestBlock(3, 1);
   OutputAqlItemRow output{std::move(block), infos.getOutputRegisters(),