[ES-2763] [COR-51] Fix crash on nested subqueries with modification

arangod/Aql/AqlCallStack.cpp

@@ -184,17 +184,21 @@ auto AqlCallStack::needToSkipSubquery() const noexcept -> bool {
                      });
 }
 
-auto AqlCallStack::shadowRowDepthToSkip() const -> size_t {
-  TRI_ASSERT(needToCountSubquery());
-  size_t const n = _operations.size();
+// TODO: is there a better name?
+auto AqlCallStack::shadowRowDepthToSkip() const -> std::optional<size_t> {
+  auto const n = _operations.size();
+
   for (size_t i = 0; i < n; ++i) {
     auto& call = _operations[i];
+    if (!call.hasMoreCalls()) {
+      return n - i - 1;
+    }
     auto const& nextCall = call.peekNextCall();
     if (nextCall.needSkipMore() || nextCall.getLimit() == 0) {
       return n - i - 1;
     }
   }
-  return 0;
+  return std::nullopt;
 }
 
 auto AqlCallStack::modifyCallAtDepth(size_t depth) -> AqlCall& {

arangod/Aql/AqlCallStack.h

@@ -123,14 +123,13 @@ class AqlCallStack {
 
   auto needToSkipSubquery() const noexcept -> bool;
   /**
-   * @brief This is only valid if needToCountSubquery is true.
-   *        It will resolve to the heighest subquery level
-   *        (outermost) that needs to be skipped.
-   *
+   * @brief This will resolve to the highest subquery level
+   *        (outermost) that needs to be skipped or std::nullopt
+   *        if no stack entry has skip or limit.
    *
    * @return size_t Depth of the subquery that asks to be skipped.
    */
-  auto shadowRowDepthToSkip() const -> size_t;
+  auto shadowRowDepthToSkip() const -> std::optional<size_t>;
 
   /**
    * @brief Get a reference to the call at the given shadowRowDepth

arangod/Aql/ExecutionBlockImpl.h

@@ -326,14 +326,6 @@ class ExecutionBlockImpl final : public ExecutionBlock {
   // as soon as we reach a place where there is no skip
   // ordered in the outer shadow rows, this call
   // will fall back to shadowRowForwarding.
-  // We need to make this method a template to prevent it from being implicitly
-  // instantiated for explicit ExecutionBlockImpl instantiations, because that
-  // would cause the static assert in the implementation to fail for executors
-  // that don't have side effects.
-  template<class E = Executor>
-  [[nodiscard]] auto sideEffectShadowRowForwarding(AqlCallStack& stack,
-                                                   SkipResult& skipResult)
-      -> ExecState;
 
   void initOnce();
 
@@ -349,18 +341,32 @@ class ExecutionBlockImpl final : public ExecutionBlock {
 
   auto countShadowRowProduced(AqlCallStack& stack, size_t depth) -> void;
 
+  auto forwardShadowRow(AqlCallStack& stack,
+                        std::unique_ptr<OutputAqlItemRow>& _outputItemRow,
+                        ShadowAqlItemRow& shadowRow) -> void;
+
+  enum class SideEffectSkipResult {
+    FORWARD_SHADOW_ROW,
+    DROP_SHADOW_ROW,
+    RETURN_DONE
+  };
+  auto sideEffectSkipHandling(AqlCallStack& stack,
+                              std::unique_ptr<OutputAqlItemRow>& _outputItemRow,
+                              ShadowAqlItemRow& shadowRow, SkipResult& skipped)
+      -> SideEffectSkipResult;
+
  private:
   /**
    * @brief The PrefetchTask is used to asynchronously prefetch the next batch
    * from upstream. Each block holds only a single instance (if any), so each
    * block can have max one pending async prefetch task. This instance is
-   * created on demand when the first async request is spawned, later tasks can
-   * reuse that instance.
-   * The async task is queued on the global scheduler so it can be picked up by
-   * some worker thread. However, sometimes the original thread might that
-   * created the task might be faster, in which case we don't want to wait
-   * until a worker has picked up the task. Instead, any thread that wants to
-   * process the task has to _claim_ it. This is managed via the task's `state`.
+   * created on demand when the first async request is spawned, later tasks
+   * can reuse that instance. The async task is queued on the global scheduler
+   * so it can be picked up by some worker thread. However, sometimes the
+   * original thread might that created the task might be faster, in which
+   * case we don't want to wait until a worker has picked up the task.
+   * Instead, any thread that wants to process the task has to _claim_ it.
+   * This is managed via the task's `state`.
    *
    * Before the task is queued on the scheduler, `state` is set to `Pending`.
    * When a thread wants to process the task, it must call `tryClaim` which