arangodb · mpoeter · Jun 20, 2024 · Jun 6, 2024 · Jun 11, 2024 · Jun 11, 2024
diff --git a/CHANGELOG b/CHANGELOG
@@ -1,6 +1,9 @@
 devel
 -----
 
+* Document modification operations (insert, update, etc) during an AQL query on
+  DBServers no longer block threads while waiting for replication.
+
 * Fixed a bug where accessing a collection right after creation can sometimes
   fail.
 

diff --git a/arangod/Aql/AqlItemBlock.h b/arangod/Aql/AqlItemBlock.h
@@ -71,10 +71,6 @@ class AqlItemBlock {
   AqlItemBlock(AqlItemBlock const&) = delete;
   AqlItemBlock& operator=(AqlItemBlock const&) = delete;
 
-  /// @brief create the block
-  AqlItemBlock(AqlItemBlockManager&, size_t numRows,
-               RegisterCount numRegisters);
-
   void initFromSlice(arangodb::velocypack::Slice);
 
   /// @brief auxiliary struct to track how often the same AqlValue is
@@ -113,8 +109,13 @@ class AqlItemBlock {
   using ShadowRowIterator = std::vector<uint32_t>::const_iterator;
 
  protected:
+  /// @brief create the block
+  /// Should only ever be called by AqlItemBlockManager, so it's protected
+  AqlItemBlock(AqlItemBlockManager&, size_t numRows,
+               RegisterCount numRegisters);
+
   /// @brief destroy the block
-  /// Should only ever be deleted by AqlItemManager::returnBlock, so the
+  /// Should only ever be deleted by AqlItemBlockManager::returnBlock, so the
   /// destructor is protected.
   ~AqlItemBlock();
 
@@ -407,11 +408,12 @@ class AqlItemBlock {
     explicit OwnershipChecker(std::atomic<std::thread::id>& v) : _v(v) {
       auto old =
           _v.exchange(std::this_thread::get_id(), std::memory_order_relaxed);
-      TRI_ASSERT(old == std::thread::id());
+      TRI_ASSERT(old == std::thread::id()) << "old=" << old;
     }
     ~OwnershipChecker() {
       auto old = _v.exchange(std::thread::id(), std::memory_order_relaxed);
-      TRI_ASSERT(old == std::this_thread::get_id());
+      TRI_ASSERT(old == std::this_thread::get_id())
+          << "old=" << old << ", this=" << std::this_thread::get_id();
     }
     std::atomic<std::thread::id>& _v;
   };

diff --git a/arangod/Aql/AqlItemBlockManager.cpp b/arangod/Aql/AqlItemBlockManager.cpp
@@ -38,7 +38,13 @@ AqlItemBlockManager::AqlItemBlockManager(
     : _resourceMonitor(resourceMonitor) {}
 
 /// @brief destroy the manager
-AqlItemBlockManager::~AqlItemBlockManager() { delete _constValueBlock; }
+AqlItemBlockManager::~AqlItemBlockManager() {
+#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+  TRI_ASSERT(_leasedBlocks == 0) << "leased blocks left: " << _leasedBlocks;
+#endif
+
+  delete _constValueBlock;
+}
 
 void AqlItemBlockManager::initializeConstValueBlock(RegisterCount nrRegs) {
   TRI_ASSERT(_constValueBlock == nullptr);
@@ -89,13 +95,24 @@ SharedAqlItemBlockPtr AqlItemBlockManager::requestBlock(
   TRI_ASSERT(block->getRefCount() == 0);
   TRI_ASSERT(block->hasShadowRows() == false);
 
+#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+  ++_leasedBlocks;
+#endif
+
   return SharedAqlItemBlockPtr{block};
 }
 
 /// @brief return a block to the manager
 void AqlItemBlockManager::returnBlock(AqlItemBlock*& block) noexcept {
   TRI_ASSERT(block != nullptr);
   TRI_ASSERT(block->getRefCount() == 0);
+#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+  auto leasedBlocks = _leasedBlocks.load();
+  do {
+    TRI_ASSERT(leasedBlocks > 0);
+  } while (
+      !_leasedBlocks.compare_exchange_weak(leasedBlocks, leasedBlocks - 1));
+#endif
 
   size_t const targetSize = block->capacity();
   uint32_t const i = Bucket::getId(targetSize);

diff --git a/arangod/Aql/AqlItemBlockManager.h b/arangod/Aql/AqlItemBlockManager.h
@@ -89,6 +89,9 @@ class AqlItemBlockManager {
 
   static constexpr uint32_t numBuckets = 12;
   static constexpr size_t numBlocksPerBucket = 7;
+#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
+  std::atomic<size_t> _leasedBlocks = 0;
+#endif
 
   struct Bucket {
     std::array<AqlItemBlock*, numBlocksPerBucket> blocks;

diff --git a/arangod/Aql/ExecutionNode/GatherNode.cpp b/arangod/Aql/ExecutionNode/GatherNode.cpp
@@ -59,18 +59,6 @@ constexpr std::string_view kParallelismParallel("parallel");
 constexpr std::string_view kParallelismSerial("serial");
 constexpr std::string_view kParallelismUndefined("undefined");
 
-constexpr std::string_view toString(GatherNode::Parallelism value) noexcept {
-  switch (value) {
-    case GatherNode::Parallelism::Parallel:
-      return kParallelismParallel;
-    case GatherNode::Parallelism::Serial:
-      return kParallelismSerial;
-    case GatherNode::Parallelism::Undefined:
-    default:
-      return kParallelismUndefined;
-  }
-}
-
 constexpr GatherNode::Parallelism parallelismFromString(
     std::string_view value) noexcept {
   if (value == kParallelismParallel) {
@@ -100,7 +88,28 @@ bool toSortMode(std::string_view str, GatherNode::SortMode& mode) noexcept {
   return true;
 }
 
-std::string_view toString(GatherNode::SortMode mode) noexcept {
+}  // namespace
+
+auto arangodb::aql::toString(GatherNode::Parallelism value) noexcept
+    -> std::string_view {
+  switch (value) {
+    case GatherNode::Parallelism::Parallel:
+      return kParallelismParallel;
+    case GatherNode::Parallelism::Serial:
+      return kParallelismSerial;
+    case GatherNode::Parallelism::Undefined:
+      return kParallelismUndefined;
+    default:
+      LOG_TOPIC("c9367", FATAL, Logger::AQL)
+          << "Invalid value for parallelism: "
+          << static_cast<std::underlying_type_t<GatherNode::Parallelism>>(
+                 value);
+      FATAL_ERROR_ABORT();
+  }
+}
+
+auto arangodb::aql::toString(GatherNode::SortMode mode) noexcept
+    -> std::string_view {
   switch (mode) {
     case GatherNode::SortMode::MinElement:
       return kSortModeMinElement;
@@ -114,8 +123,6 @@ std::string_view toString(GatherNode::SortMode mode) noexcept {
   }
 }
 
-}  // namespace
-
 /*static*/ Collection const* GatherNode::findCollection(
     GatherNode const& root) noexcept {
   ExecutionNode const* node = root.getFirstDependency();

diff --git a/arangod/Aql/ExecutionNode/GatherNode.h b/arangod/Aql/ExecutionNode/GatherNode.h
@@ -147,5 +147,8 @@ class GatherNode final : public ExecutionNode {
   size_t _limit;
 };
 
+auto toString(GatherNode::SortMode mode) noexcept -> std::string_view;
+auto toString(GatherNode::Parallelism) noexcept -> std::string_view;
+
 }  // namespace aql
 }  // namespace arangodb
diff --git a/arangod/Aql/Executor/ModificationExecutor.h b/arangod/Aql/Executor/ModificationExecutor.h
@@ -170,7 +170,7 @@ class ModificationExecutor {
   using Stats = ModificationStats;
 
   ModificationExecutor(FetcherType&, Infos&);
-  ~ModificationExecutor() = default;
+  ~ModificationExecutor();
 
   [[nodiscard]] auto produceRows(typename FetcherType::DataRange& input,
                                  OutputAqlItemRow& output)
@@ -206,4 +206,12 @@ class ModificationExecutor {
   Stats _stats{};
 };
 
+template<typename FetcherType, typename ModifierType>
+ModificationExecutor<FetcherType, ModifierType>::~ModificationExecutor() {
+  // Clear all InputAqlItemRows the modifier still holds, and with it the
+  // SharedAqlItemBlockPtrs. This is so the referenced AqlItemBlocks can be
+  // returned to the AqlItemBlockManager now, while the latter still exists.
+  _modifier->clearRows();
+}
+
 }  // namespace arangodb::aql
diff --git a/arangod/Aql/QuerySnippet.cpp b/arangod/Aql/QuerySnippet.cpp
@@ -23,7 +23,6 @@
 
 #include "QuerySnippet.h"
 
-#include "Aql/Collection.h"
 #include "Aql/ExecutionNode/CollectionAccessingNode.h"
 #include "Aql/ExecutionNode/DistributeConsumerNode.h"
 #include "Aql/ExecutionNode/DistributeNode.h"
@@ -40,6 +39,8 @@
 #include "Basics/Exceptions.h"
 #include "Basics/StringUtils.h"
 #include "Cluster/ServerState.h"
+#include "Logger/LogLevel.h"
+#include "Logger/LogMacros.h"
 
 #ifdef USE_ENTERPRISE
 #include "Enterprise/Aql/LocalGraphNode.h"
@@ -470,6 +471,15 @@ void QuerySnippet::serializeIntoBuilder(
     // it needs to expose its input register by all means
     internalGather->setVarsUsedLater(_nodes.front()->getVarsUsedLaterStack());
     internalGather->setRegsToClear({});
+    // No DBServer-internal parallelism (yet)
+    auto const parallelism = internalGather->parallelism();
+    LOG_TOPIC_IF("dd0f1", DEBUG, Logger::QUERIES,
+                 parallelism != GatherNode::Parallelism::Serial)
+        << "Overriding parallelism of " << toString(parallelism) << " with "
+        << toString(GatherNode::Parallelism::Serial)
+        << " on the DBServer's gather node (belonging to " << _sinkNode->id()
+        << ")";
+    internalGather->setParallelism(GatherNode::Parallelism::Serial);
     auto const reservedId = ExecutionNodeId::InternalNode;
     nodeAliases.try_emplace(internalGather->id(), reservedId);
 
@@ -520,7 +530,7 @@ void QuerySnippet::serializeIntoBuilder(
         }
       } else {
         // In this case we actually do not care for the real value, we just need
-        // to ensure that every client get's exactly one copy.
+        // to ensure that every client gets exactly one copy.
         for (size_t i = 0; i < numberOfShardsToPermutate; i++) {
           distIds.emplace_back(StringUtils::itoa(i));
         }

diff --git a/arangod/Aql/SimpleModifier.cpp b/arangod/Aql/SimpleModifier.cpp
@@ -186,13 +186,6 @@ ExecutionState SimpleModifier<ModifierCompletion, Enable>::transact(
 
   auto result = _completion.transact(trx, _accumulator.closeAndGetContents());
 
-  // we are currently waiting here for the `result` future to get
-  // ready before we continue. this makes the AQL modification
-  // operations blocking as in previous versions of ArangoDB.
-  // TODO: fix this and make it truly non-blocking (requires to
-  // fix some lifecycle issues for AQL queries first).
-  result.wait();
-
   if (result.isReady()) {
     _results = std::move(result.waitAndGet());
     return ExecutionState::DONE;
@@ -360,6 +353,11 @@ bool SimpleModifier<ModifierCompletion,
                     _results);
 }
 
+template<typename ModifierCompletion, typename Enable>
+void SimpleModifier<ModifierCompletion, Enable>::clearRows() noexcept {
+  _operations.clear();
+}
+
 template class ::arangodb::aql::SimpleModifier<InsertModifierCompletion>;
 template class ::arangodb::aql::SimpleModifier<RemoveModifierCompletion>;
 template class ::arangodb::aql::SimpleModifier<UpdateReplaceModifierCompletion>;
diff --git a/arangod/Aql/SimpleModifier.h b/arangod/Aql/SimpleModifier.h
@@ -150,6 +150,11 @@ class SimpleModifier : public std::enable_shared_from_this<
   bool hasResultOrException() const noexcept;
   bool hasNeitherResultNorOperationPending() const noexcept;
 
+  // Destroy all InputAqlItemRows, and with it SharedAqlItemBlockPtrs, this
+  // holds. This is necessary to ensure the lifetime of the AqlItemBlocks is
+  // shorter than of the AqlItemBlockManager, to which they are returned.
+  void clearRows() noexcept;
+
  private:
   [[nodiscard]] bool resultAvailable() const;
   [[nodiscard]] VPackArrayIterator getResultsIterator() const;

diff --git a/arangod/Aql/UpsertModifier.cpp b/arangod/Aql/UpsertModifier.cpp
@@ -368,3 +368,5 @@ bool UpsertModifier::hasResultOrException() const noexcept {
 bool UpsertModifier::hasNeitherResultNorOperationPending() const noexcept {
   return resultState() == ModificationExecutorResultState::NoResult;
 }
+
+void UpsertModifier::clearRows() noexcept { _operations.clear(); }
diff --git a/arangod/Aql/UpsertModifier.h b/arangod/Aql/UpsertModifier.h
@@ -116,6 +116,11 @@ class UpsertModifier {
   bool hasResultOrException() const noexcept;
   bool hasNeitherResultNorOperationPending() const noexcept;
 
+  // Destroy all InputAqlItemRows, and with it SharedAqlItemBlockPtrs, this
+  // holds. This is necessary to ensure the lifetime of the AqlItemBlocks is
+  // shorter than of the AqlItemBlockManager, to which they are returned.
+  void clearRows() noexcept;
+
  private:
   bool resultAvailable() const;
   VPackArrayIterator getUpdateResultsIterator() const;

diff --git a/arangod/Transaction/Helpers.cpp b/arangod/Transaction/Helpers.cpp
@@ -905,9 +905,12 @@ BuilderLeaser::BuilderLeaser(Context* transactionContext)
 BuilderLeaser::BuilderLeaser(Methods* trx)
     : BuilderLeaser{trx->transactionContextPtr()} {}
 
-BuilderLeaser::~BuilderLeaser() {
+BuilderLeaser::~BuilderLeaser() { clear(); }
+
+void BuilderLeaser::clear() {
   if (_builder != nullptr) {
     _transactionContext->returnBuilder(_builder);
+    _builder = nullptr;
   }
 }
 

diff --git a/arangod/Transaction/Helpers.h b/arangod/Transaction/Helpers.h
@@ -189,6 +189,8 @@ class BuilderLeaser {
   velocypack::Builder* get() const noexcept { return _builder; }
   velocypack::Builder* steal() { return std::exchange(_builder, nullptr); }
 
+  void clear();
+
  private:
   Context* _transactionContext;
   velocypack::Builder* _builder;