arangodb · jsteemann · Jul 19, 2021 · Jul 21, 2021 · Jul 21, 2021 · Jul 22, 2021
diff --git a/CHANGELOG b/CHANGELOG
@@ -1,6 +1,10 @@
 v3.8.1 (XXXX-XX-XX)
 -------------------
 
+* Make AQL modification operations in a cluster asynchronous. This allows to
+  free the thread for other work until both the write and synchronous
+  replication are complete.
+
 * When creating Pregel memory-mapped files, create them with O_TMPFILE attribute
   on Linux so that files are guaranteed to vanish even if a process dies.
 

diff --git a/arangod/Aql/AqlCall.h b/arangod/Aql/AqlCall.h
@@ -192,7 +192,7 @@ struct AqlCall {
           TRI_ASSERT(n <= i);
           i -= n;
         },
-        [](auto) {},
+        [](Infinity) {},
     };
     std::visit(minus, softLimit);
     std::visit(minus, hardLimit);

diff --git a/arangod/Aql/AqlExecuteResult.cpp b/arangod/Aql/AqlExecuteResult.cpp
@@ -35,6 +35,7 @@
 
 #include <map>
 #include <string_view>
+#include <utility>
 
 using namespace arangodb;
 using namespace arangodb::aql;
@@ -49,7 +50,7 @@ auto getStringView(velocypack::Slice slice) -> std::string_view {
 
 AqlExecuteResult::AqlExecuteResult(ExecutionState state, SkipResult skipped,
                                    SharedAqlItemBlockPtr&& block)
-    : _state(state), _skipped(skipped), _block(std::move(block)) {
+    : _state(state), _skipped(std::move(skipped)), _block(std::move(block)) {
   // Make sure we only produce a valid response
   // The block should have checked as well.
   // We must return skipped and/or data when reporting HASMORE

diff --git a/arangod/Aql/AqlItemBlock.cpp b/arangod/Aql/AqlItemBlock.cpp
@@ -1169,7 +1169,7 @@ size_t AqlItemBlock::maxModifiedEntries() const noexcept { return _numRegisters
 
 size_t AqlItemBlock::capacity() const noexcept { return _data.capacity(); }
 
-bool AqlItemBlock::isShadowRow(size_t row) const {
+bool AqlItemBlock::isShadowRow(size_t row) const noexcept {
   return _shadowRows.is(row);
 }
 

diff --git a/arangod/Aql/AqlItemBlock.h b/arangod/Aql/AqlItemBlock.h
@@ -296,7 +296,7 @@ class AqlItemBlock {
 
   /// @brief test if the given row is a shadow row and conveys subquery
   /// information only. It should not be handed to any non-subquery executor.
-  bool isShadowRow(size_t row) const;
+  bool isShadowRow(size_t row) const noexcept;
 
   /// @brief get the ShadowRowDepth 
   /// Does only work if this row is a shadow row

diff --git a/arangod/Aql/AqlItemBlockInputMatrix.cpp b/arangod/Aql/AqlItemBlockInputMatrix.cpp
@@ -42,8 +42,8 @@ AqlItemBlockInputMatrix::AqlItemBlockInputMatrix(ExecutorState state)
    }
 
 // only used for block passthrough
-AqlItemBlockInputMatrix::AqlItemBlockInputMatrix(arangodb::aql::SharedAqlItemBlockPtr const& block)
-    : _block{block}, _aqlItemMatrix{nullptr} {
+AqlItemBlockInputMatrix::AqlItemBlockInputMatrix(arangodb::aql::SharedAqlItemBlockPtr block)
+    : _block{std::move(block)}, _aqlItemMatrix{nullptr} {
   TRI_ASSERT(_aqlItemMatrix == nullptr);
   TRI_ASSERT(!hasDataRow());
 }
@@ -64,7 +64,7 @@ AqlItemBlockInputRange& AqlItemBlockInputMatrix::getInputRange() {
   if (_lastRange.hasDataRow()) {
     return _lastRange;
   }
-  // Need initialze lastRange
+  // Need initialize lastRange
   if (_aqlItemMatrix->numberOfBlocks() == 0) {
     _lastRange = {AqlItemBlockInputRange{upstreamState()}};
   } else {
@@ -83,7 +83,6 @@ SharedAqlItemBlockPtr AqlItemBlockInputMatrix::getBlock() const noexcept {
 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.
@@ -109,7 +108,7 @@ bool AqlItemBlockInputMatrix::hasValidRow() const noexcept {
 
 bool AqlItemBlockInputMatrix::hasDataRow() const noexcept {
   return _aqlItemMatrix != nullptr && !hasShadowRow() &&
-         ((_aqlItemMatrix->stoppedOnShadowRow()) ||
+         (_aqlItemMatrix->stoppedOnShadowRow() ||
           (_aqlItemMatrix->size() > 0 && _finalState == ExecutorState::DONE));
 }
 

diff --git a/arangod/Aql/AqlItemBlockInputMatrix.h b/arangod/Aql/AqlItemBlockInputMatrix.h
@@ -21,8 +21,7 @@
 /// @author Tobias Gödderz
 ////////////////////////////////////////////////////////////////////////////////
 
-#ifndef ARANGOD_AQL_AQLITEMBLOCKMATRIXITERATOR_H
-#define ARANGOD_AQL_AQLITEMBLOCKMATRIXITERATOR_H
+#pragma once
 
 #include "Aql/AqlItemBlockInputRange.h"
 #include "Aql/AqlItemMatrix.h"
@@ -38,7 +37,7 @@ class AqlItemBlockInputMatrix {
  public:
   explicit AqlItemBlockInputMatrix(ExecutorState state);
 
-  AqlItemBlockInputMatrix(arangodb::aql::SharedAqlItemBlockPtr const&);
+  explicit AqlItemBlockInputMatrix(arangodb::aql::SharedAqlItemBlockPtr);
 
   AqlItemBlockInputMatrix(ExecutorState state, AqlItemMatrix* aqlItemMatrix);
 
@@ -63,7 +62,6 @@ class AqlItemBlockInputMatrix {
 
   size_t skipAllShadowRowsOfDepth(size_t depth);
 
-
      // Will return HASMORE if we were able to increase the row index.
   // Otherwise will return DONE.
   ExecutorState incrBlockIndex();
@@ -101,5 +99,3 @@ class AqlItemBlockInputMatrix {
 };
 
 }  // namespace arangodb::aql
-
-#endif  // ARANGOD_AQL_AQLITEMBLOCKINPUTITERATOR_H
diff --git a/arangod/Aql/AqlItemMatrix.cpp b/arangod/Aql/AqlItemMatrix.cpp
@@ -245,3 +245,94 @@ AqlItemMatrix::AqlItemMatrix(RegisterCount nrRegs)
   clear();
   return {skipped, ShadowAqlItemRow{CreateInvalidShadowRowHint()}};
 }
+
+AqlItemMatrix::RowIterator AqlItemMatrix::begin() const {
+  if (size() > 0) {
+    return {this, 0, _startIndexInFirstBlock};
+  } else {
+    return end();
+  }
+}
+
+AqlItemMatrix::RowIterator AqlItemMatrix::end() const {
+  return {this, this->numberOfBlocks(), 0};
+}
+
+AqlItemMatrix::RowIterator::RowIterator(AqlItemMatrix const* matrix, size_t blockIndex, size_t rowIndex)
+    : _matrix(matrix), _blockIndex(blockIndex), _rowIndex(rowIndex) {}
+
+AqlItemMatrix::RowIterator::value_type AqlItemMatrix::RowIterator::next() noexcept {
+  auto& it = *this;
+  auto ret = *it;
+  ++it;
+  return ret;
+}
+
+auto AqlItemMatrix::RowIterator::isInitialized() const noexcept -> bool {
+  return _matrix != nullptr;
+}
+
+auto AqlItemMatrix::RowIterator::hasMore() const noexcept -> bool {
+  // _blockIndex == _matrix->size() => _rowIndex == 0
+  TRI_ASSERT((_matrix != nullptr && _blockIndex < _matrix->numberOfBlocks()) || _rowIndex == 0);
+  // If _blockIndex is valid, _rowIndex must be, too.
+  return ADB_LIKELY(_matrix != nullptr) && _blockIndex < _matrix->numberOfBlocks();
+}
+
+AqlItemMatrix::RowIterator::value_type AqlItemMatrix::RowIterator::operator*() const noexcept {
+  return {_matrix->getBlock(_blockIndex).first, _rowIndex};
+}
+
+AqlItemMatrix::RowIterator& AqlItemMatrix::RowIterator::operator++() noexcept {
+  // Assume ++ is only called on a valid and dereferenceable iterator
+  TRI_ASSERT(_matrix != nullptr);
+  TRI_ASSERT(_blockIndex < _matrix->numberOfBlocks());
+  auto const* block = _matrix->getBlockRef(_blockIndex).first;
+  TRI_ASSERT(_rowIndex < block->numRows());
+  TRI_ASSERT(!block->isShadowRow(_rowIndex));
+
+  // Increase the row index
+  ++_rowIndex;
+  if (_rowIndex >= block->numRows()) {
+    // If the row index is invalid, move to the next block.
+    // If the block index is now invalid, this is equal to the "end()"
+    // iterator.
+    ++_blockIndex;
+    _rowIndex = 0;
+  }
+
+  if (_blockIndex < _matrix->numberOfBlocks()) {
+    block = _matrix->getBlockRef(_blockIndex).first;
+    if (block->isShadowRow(_rowIndex)) {
+      // If we're at a shadow row, this must be the last block.
+      TRI_ASSERT(_blockIndex + 1 == _matrix->numberOfBlocks());
+      // This makes this equal to the "end()" iterator.
+      ++_blockIndex;
+      _rowIndex = 0;
+    }
+  }
+
+  return *this;
+}
+
+auto AqlItemMatrix::RowIterator::operator++(int) & noexcept -> AqlItemMatrix::RowIterator {
+  auto tmp = *this;
+  ++(*this);
+  return tmp;
+}
+
+AqlItemMatrix::RowIterator::operator bool() const noexcept {
+  return hasMore();
+}
+
+bool aql::operator==(AqlItemMatrix::RowIterator const& a,
+                     AqlItemMatrix::RowIterator const& b) {
+  return ADB_LIKELY(a._matrix == b._matrix) &&
+         (ADB_UNLIKELY(a._matrix == nullptr /* => b._matrix == nullptr */) ||
+          (ADB_LIKELY(a._blockIndex == b._blockIndex) && a._rowIndex == b._rowIndex));
+}
+
+bool aql::operator!=(AqlItemMatrix::RowIterator const& a,
+                     AqlItemMatrix::RowIterator const& b) {
+  return !(a == b);
+}
diff --git a/arangod/Aql/AqlItemMatrix.h b/arangod/Aql/AqlItemMatrix.h
@@ -21,17 +21,15 @@
 /// @author Michael Hackstein
 ////////////////////////////////////////////////////////////////////////////////
 
-#ifndef ARANGOD_AQL_AQL_ITEM_MATRIX_H
-#define ARANGOD_AQL_AQL_ITEM_MATRIX_H 1
+#pragma once
 
 #include "Aql/ShadowAqlItemRow.h"
 
 #include <cstdint>
 #include <utility>
 #include <vector>
 
-namespace arangodb {
-namespace aql {
+namespace arangodb::aql {
 
 class InputAqlItemRow;
 class SharedAqlItemBlockPtr;
@@ -120,6 +118,49 @@ class AqlItemMatrix {
   [[nodiscard]] auto skipAllShadowRowsOfDepth(size_t depth)
       -> std::tuple<size_t, ShadowAqlItemRow>;
 
+  class RowIterator {
+   public:
+    using value_type        = InputAqlItemRow;
+
+    RowIterator() = default;
+    RowIterator(AqlItemMatrix const* matrix, size_t blockIndex, size_t rowIndex);
+
+    // Returns the current value, and move the iterator to the next value
+    value_type next() noexcept;
+
+    auto isInitialized() const noexcept -> bool;
+
+    // Returns whether the current value is valid, i.e. whether next() may be
+    // called
+    auto hasMore() const noexcept -> bool;
+
+    value_type operator*() const noexcept;
+
+    // This can't be implemented, as we can only create the InputAqlItemRow
+    // on-the-fly.
+    // pointer operator->();
+
+    // Prefix increment
+    RowIterator& operator++() noexcept;
+
+    // Postfix increment.
+    auto operator++(int) & noexcept -> RowIterator;
+
+    explicit operator bool() const noexcept;
+
+    friend bool operator==(RowIterator const& a, RowIterator const& b);
+    friend bool operator!=(RowIterator const& a, RowIterator const& b);
+
+   private:
+    AqlItemMatrix const* _matrix{};
+    std::size_t _blockIndex{};
+    // Invariant: _rowIndex is valid iff _blockIndex is valid.
+    std::size_t _rowIndex{};
+  };
+
+  [[nodiscard]] RowIterator begin() const;
+  [[nodiscard]] RowIterator end() const;
+
  private:
   std::vector<SharedAqlItemBlockPtr> _blocks;
 
@@ -130,7 +171,7 @@ class AqlItemMatrix {
   size_t _stopIndexInLastBlock;
 };
 
-}  // namespace aql
-}  // namespace arangodb
+bool operator==(AqlItemMatrix::RowIterator const& a, AqlItemMatrix::RowIterator const& b);
+bool operator!=(AqlItemMatrix::RowIterator const& a, AqlItemMatrix::RowIterator const& b);
 
-#endif
+}  // namespace arangodb::aql
diff --git a/arangod/Aql/ClusterNodes.cpp b/arangod/Aql/ClusterNodes.cpp
@@ -612,6 +612,7 @@ std::unique_ptr<ExecutionBlock> SingleRemoteOperationNode::createBlock(
                                            std::move(writableOutputRegisters));
 
   auto executorInfos = SingleRemoteModificationInfos(
+      &engine,
       in, outputNew, outputOld, out, _plan->getAst()->query(), std::move(options),
       collection(), ConsultAqlWriteFilter(_options.consultAqlWriteFilter),
       IgnoreErrors(_options.ignoreErrors),