SubqueryStart/SubqueryEnd

arangodb · mchacki · Feb 29, 2020 · Jan 29, 2020 · Jan 31, 2020 · Jan 31, 2020
commit 88ba3cd137781eafdd0b66bc59ed8ed8fdbb3f23
diff --git a/arangod/Aql/ExecutionBlockImpl.cpp b/arangod/Aql/ExecutionBlockImpl.cpp
@@ -1145,27 +1145,15 @@ ExecutionBlockImpl<Executor>::executeWithoutTrace(AqlCallStack stack) {
       std::tie(_upstreamState, skippedLocal, bypassedRange) = _rowFetcher.execute(stack);
       return {_upstreamState, skippedLocal, bypassedRange.getBlock()};
     }
-    AqlCall clientCall = stack.popCall();
 
-    // Callstack handling for subqueries
+    //    AqlCall subqueryCall{};
+    AqlCall clientCall = stack.popCall();
     if constexpr (std::is_same_v<Executor, SubqueryStartExecutor>) {
-      // inspect the top two calls
-      // clientCall has the "topmost" which is coming from the subquery
-      // but we need the next one as real clientCall
-      //      AqlCall subqueryCall = clientCall;
+      // TODO: what happens to the call from the subquery? Currently
+      // we're just discarding it
+      // subqueryCall = clientCall;
+      // (void)subqueryCall;
       clientCall = stack.popCall();
-
-      // TODO: what do we do with subquery call? We know that the subquery
-      //       needs one row to produce one row; If the subquery has an offset
-      //       we might use that, too?
-    } else if constexpr (std::is_same_v<Executor, SubqueryEndExecutor>) {
-      // push original call
-      auto outsideCall = clientCall;
-      stack.pushCall(std::move(outsideCall));
-      clientCall = AqlCall{};
-
-      // special hack
-      // clientCall = AqlCall{clientCall.getOffset(), 0, 0, false};
     }
 
     // We can only have returned the following internal states
@@ -1281,7 +1269,23 @@ ExecutionBlockImpl<Executor>::executeWithoutTrace(AqlCallStack stack) {
           TRI_ASSERT(!_lastRange.hasShadowRow());
           size_t skippedLocal = 0;
           auto callCopy = _upstreamRequest;
+
           stack.pushCall(std::move(callCopy));
+
+          // Callstack handling for subqueries
+          // TODO: If the subquery is modifying we need to do some
+          //       special handling here
+          if constexpr (std::is_same_v<Executor, SubqueryEndExecutor>) {
+            if (_executor.isModificationSubquery()) {
+            } else {
+              if (_upstreamRequest.getOffset() > 0) {
+                stack.pushCall(AqlCall{_upstreamRequest.getOffset(), 0, 0, false});
+              } else {
+                stack.pushCall(AqlCall{});
+              }
+            }
+          }
+
           std::tie(_upstreamState, skippedLocal, _lastRange) = _rowFetcher.execute(stack);
           if (_upstreamState == ExecutionState::WAITING) {
             // We need to persist the old call before we return.
@@ -1305,6 +1309,15 @@ ExecutionBlockImpl<Executor>::executeWithoutTrace(AqlCallStack stack) {
           // E.g. LastRow is releveant shadowRow. NextRow is non-relevant shadowRow.
           // NOTE: I do not think this is an issue, as the Executor will always say that it cannot do anything with
           // an empty input. Only exception might be COLLECT COUNT.
+
+          if constexpr (std::is_same_v<Executor, SubqueryStartExecutor>) {
+            // TODO: make sure ensureOutputBlock can be called here
+            ensureOutputBlock(std::move(clientCall));
+            _executor.produceShadowRow(*_outputItemRow);
+            // produceShadowRow modifies output
+            clientCall = _outputItemRow->getClientCall();
+          }
+
           if (_lastRange.hasShadowRow()) {
             auto const& [state, shadowRow] = _lastRange.nextShadowRow();
             TRI_ASSERT(shadowRow.isInitialized());

diff --git a/arangod/Aql/ExecutionState.cpp b/arangod/Aql/ExecutionState.cpp
@@ -50,6 +50,9 @@ std::ostream& operator<<(std::ostream& ostream, ExecutorState state) {
     case ExecutorState::HASMORE:
       ostream << "HASMORE";
       break;
+    default:
+      ostream << " WAT WAT WAT";
+      break;
   }
   return ostream;
 }

diff --git a/arangod/Aql/OptimizerRules.cpp b/arangod/Aql/OptimizerRules.cpp
diff --git a/arangod/Aql/SubqueryEndExecutionNode.cpp b/arangod/Aql/SubqueryEndExecutionNode.cpp
@@ -51,7 +51,18 @@ bool CompareVariables(Variable const* mine, Variable const* yours) {
 SubqueryEndNode::SubqueryEndNode(ExecutionPlan* plan, arangodb::velocypack::Slice const& base)
     : ExecutionNode(plan, base),
       _inVariable(Variable::varFromVPack(plan->getAst(), base, "inVariable", true)),
-      _outVariable(Variable::varFromVPack(plan->getAst(), base, "outVariable")) {}
+      _outVariable(Variable::varFromVPack(plan->getAst(), base, "outVariable")),
+      _isModificationSubquery(base.get("isModificationSubquery").getBoolean()) {}
+
+SubqueryEndNode::SubqueryEndNode(ExecutionPlan* plan, size_t id, Variable const* inVariable,
+                                 Variable const* outVariable, bool isModificationSubquery)
+    : ExecutionNode(plan, id),
+      _inVariable(inVariable),
+      _outVariable(outVariable),
+      _isModificationSubquery(isModificationSubquery) {
+  // _inVariable might be nullptr
+  TRI_ASSERT(_outVariable != nullptr);
+}
 
 void SubqueryEndNode::toVelocyPackHelper(VPackBuilder& nodes, unsigned flags,
                                          std::unordered_set<ExecutionNode const*>& seen) const {
@@ -90,8 +101,9 @@ std::unique_ptr<ExecutionBlock> SubqueryEndNode::createBlock(
   auto const vpackOptions = trx->transactionContextPtr()->getVPackOptions();
   SubqueryEndExecutorInfos infos(inputRegisters, outputRegisters,
                                  getRegisterPlan()->nrRegs[previousNode->getDepth()],
-                                 getRegisterPlan()->nrRegs[getDepth()], getRegsToClear(),
-                                 calcRegsToKeep(), vpackOptions, inReg, outReg);
+                                 getRegisterPlan()->nrRegs[getDepth()],
+                                 getRegsToClear(), calcRegsToKeep(), vpackOptions,
+                                 inReg, outReg, isModificationSubquery());
 
   return std::make_unique<ExecutionBlockImpl<SubqueryEndExecutor>>(&engine, this,
                                                                    std::move(infos));
@@ -108,7 +120,8 @@ ExecutionNode* SubqueryEndNode::clone(ExecutionPlan* plan, bool withDependencies
       inVariable = plan->getAst()->variables()->createVariable(inVariable);
     }
   }
-  auto c = std::make_unique<SubqueryEndNode>(plan, _id, inVariable, outVariable);
+  auto c = std::make_unique<SubqueryEndNode>(plan, _id, inVariable, outVariable,
+                                             isModificationSubquery());
 
   return cloneHelper(std::move(c), withDependencies, withProperties);
 }
@@ -134,7 +147,8 @@ bool SubqueryEndNode::isEqualTo(ExecutionNode const& other) const {
     SubqueryEndNode const& p = dynamic_cast<SubqueryEndNode const&>(other);
     TRI_ASSERT(p._outVariable != nullptr);
     if (!CompareVariables(_outVariable, p._outVariable) ||
-        !CompareVariables(_inVariable, p._inVariable)) {
+        !CompareVariables(_inVariable, p._inVariable) ||
+        _isModificationSubquery != p._isModificationSubquery) {
       // One of the variables does not match
       return false;
     }
@@ -144,3 +158,7 @@ bool SubqueryEndNode::isEqualTo(ExecutionNode const& other) const {
     return false;
   }
 }
+
+bool SubqueryEndNode::isModificationSubquery() const {
+  return _isModificationSubquery;
+}
diff --git a/arangod/Aql/SubqueryEndExecutionNode.h b/arangod/Aql/SubqueryEndExecutionNode.h
@@ -37,11 +37,7 @@ class SubqueryEndNode : public ExecutionNode {
   SubqueryEndNode(ExecutionPlan*, arangodb::velocypack::Slice const& base);
 
   SubqueryEndNode(ExecutionPlan* plan, size_t id, Variable const* inVariable,
-                  Variable const* outVariable)
-      : ExecutionNode(plan, id), _inVariable(inVariable), _outVariable(outVariable) {
-    // _inVariable might be nullptr
-    TRI_ASSERT(_outVariable != nullptr);
-  }
+                  Variable const* outVariable, bool isModificationSubquery);
 
   CostEstimate estimateCost() const override final;
 
@@ -74,11 +70,12 @@ class SubqueryEndNode : public ExecutionNode {
   }
 
   void replaceOutVariable(Variable const* var);
+  bool isModificationSubquery() const;
 
  private:
   Variable const* _inVariable;
-
   Variable const* _outVariable;
+  bool _isModificationSubquery;
 };
 
 }  // namespace aql

diff --git a/arangod/Aql/SubqueryEndExecutor.cpp b/arangod/Aql/SubqueryEndExecutor.cpp
@@ -34,6 +34,8 @@
 #include <memory>
 #include <utility>
 
+#include "Logger/LogMacros.h"
+
 using namespace arangodb;
 using namespace arangodb::aql;
 
@@ -42,14 +44,15 @@ SubqueryEndExecutorInfos::SubqueryEndExecutorInfos(
     std::shared_ptr<std::unordered_set<RegisterId>> writeableOutputRegisters,
     RegisterId nrInputRegisters, RegisterId nrOutputRegisters,
     std::unordered_set<RegisterId> const& registersToClear,
-    std::unordered_set<RegisterId> registersToKeep,
-    velocypack::Options const* const options, RegisterId inReg, RegisterId outReg)
+    std::unordered_set<RegisterId> registersToKeep, velocypack::Options const* const options,
+    RegisterId inReg, RegisterId outReg, bool isModificationSubquery)
     : ExecutorInfos(std::move(readableInputRegisters),
                     std::move(writeableOutputRegisters), nrInputRegisters,
                     nrOutputRegisters, registersToClear, std::move(registersToKeep)),
       _vpackOptions(options),
       _outReg(outReg),
-      _inReg(inReg) {}
+      _inReg(inReg),
+      _isModificationSubquery(isModificationSubquery) {}
 
 SubqueryEndExecutorInfos::~SubqueryEndExecutorInfos() = default;
 
@@ -69,6 +72,10 @@ RegisterId SubqueryEndExecutorInfos::getInputRegister() const noexcept {
   return _inReg;
 }
 
+bool SubqueryEndExecutorInfos::isModificationSubquery() const noexcept {
+  return _isModificationSubquery;
+}
+
 SubqueryEndExecutor::SubqueryEndExecutor(Fetcher& fetcher, SubqueryEndExecutorInfos& infos)
     : _fetcher(fetcher), _infos(infos), _accumulator(_infos.vpackOptions()) {}
 
@@ -80,7 +87,7 @@ std::pair<ExecutionState, NoStats> SubqueryEndExecutor::produceRows(OutputAqlIte
 
 auto SubqueryEndExecutor::produceRows(AqlItemBlockInputRange& input, OutputAqlItemRow& output)
     -> std::tuple<ExecutorState, Stats, AqlCall> {
-  ExecutorState state;
+  ExecutorState state{ExecutorState::HASMORE};
   InputAqlItemRow inputRow = InputAqlItemRow{CreateInvalidInputRowHint()};
 
   while (input.hasDataRow()) {
@@ -94,7 +101,7 @@ auto SubqueryEndExecutor::produceRows(AqlItemBlockInputRange& input, OutputAqlIt
       _accumulator.addValue(inputRow.getValue(_infos.getInputRegister()));
     }
   }
-  return {state, NoStats{}, AqlCall{}};
+  return {input.upstreamState(), NoStats{}, AqlCall{}};
 }
 
 auto SubqueryEndExecutor::skipRowsRange(AqlItemBlockInputRange& input, AqlCall& call)
@@ -105,18 +112,11 @@ auto SubqueryEndExecutor::skipRowsRange(AqlItemBlockInputRange& input, AqlCall&
   while (input.hasDataRow()) {
     std::tie(state, inputRow) = input.nextDataRow();
     TRI_ASSERT(inputRow.isInitialized());
-
-    // We got a data row, put it into the accumulator,
-    // if we're getting data through an input register.
-    // If not, we just "accumulate" an empty output.
-    // if (_infos.usesInputRegister()) {
-    //   _accumulator.addValue(inputRow.getValue(_infos.getInputRegister()));
-    // }
   }
   // This is correct since the SubqueryEndExecutor produces one output out
   // of the accumulation of all the (relevant) inputs
   call.didSkip(1);
-  return {state, 1, AqlCall{}};
+  return {input.upstreamState(), 1, AqlCall{}};
 }
 
 auto SubqueryEndExecutor::stealValue(AqlValue& result) -> AqlValueGuard {
@@ -125,6 +125,10 @@ auto SubqueryEndExecutor::stealValue(AqlValue& result) -> AqlValueGuard {
   return _accumulator.stealValue(result);
 }
 
+auto SubqueryEndExecutor::isModificationSubquery() const noexcept -> bool {
+  return _infos.isModificationSubquery();
+}
+
 void SubqueryEndExecutor::Accumulator::reset() {
   _buffer = std::make_unique<arangodb::velocypack::Buffer<uint8_t>>();
   _builder = std::make_unique<VPackBuilder>(*_buffer);

diff --git a/arangod/Aql/SubqueryEndExecutor.h b/arangod/Aql/SubqueryEndExecutor.h
@@ -50,7 +50,7 @@ class SubqueryEndExecutorInfos : public ExecutorInfos {
                            std::unordered_set<RegisterId> const& registersToClear,
                            std::unordered_set<RegisterId> registersToKeep,
                            velocypack::Options const* options, RegisterId inReg,
-                           RegisterId outReg);
+                           RegisterId outReg, bool isModificationSubquery);
 
   SubqueryEndExecutorInfos() = delete;
   SubqueryEndExecutorInfos(SubqueryEndExecutorInfos&&) noexcept = default;
@@ -61,11 +61,13 @@ class SubqueryEndExecutorInfos : public ExecutorInfos {
   [[nodiscard]] RegisterId getOutputRegister() const noexcept;
   [[nodiscard]] bool usesInputRegister() const noexcept;
   [[nodiscard]] RegisterId getInputRegister() const noexcept;
+  [[nodiscard]] bool isModificationSubquery() const noexcept;
 
  private:
   velocypack::Options const* _vpackOptions;
   RegisterId const _outReg;
   RegisterId const _inReg;
+  bool const _isModificationSubquery;
 };
 
 class SubqueryEndExecutor {
@@ -98,6 +100,7 @@ class SubqueryEndExecutor {
   [[nodiscard]] auto skipRowsRange(AqlItemBlockInputRange& input, AqlCall& call)
       -> std::tuple<ExecutorState, size_t, AqlCall>;
   [[nodiscard]] auto stealValue(AqlValue& result) -> AqlValueGuard;
+  [[nodiscard]] auto isModificationSubquery() const noexcept -> bool;
 
  private:
   enum class State {

diff --git a/arangod/Aql/SubqueryStartExecutionNode.h b/arangod/Aql/SubqueryStartExecutionNode.h
@@ -57,10 +57,13 @@ class SubqueryStartNode : public ExecutionNode {
 
   bool isEqualTo(ExecutionNode const& other) const override final;
 
+  bool isModificationNode();
+
  private:
   /// @brief This is only required for Explain output.
   ///        it has no practical usage other then to print this information during explain.
   Variable const* _subqueryOutVariable;
+  bool _isModificationSubquery;
 };
 
 }  // namespace aql