Refactor disabling of intermediate commits in case of UPSERTs

arangodb · jsteemann · Aug 31, 2021 · Jun 8, 2021 · Jun 8, 2021 · Jun 8, 2021
commit b6f1e5822cbc6883c13e2a1373f802be66f9c46f
diff --git a/arangod/Aql/AqlTransaction.cpp b/arangod/Aql/AqlTransaction.cpp
@@ -55,7 +55,9 @@ AqlTransaction::AqlTransaction(
     std::shared_ptr<transaction::Context> const& transactionContext,
     transaction::Options const& options)
     : transaction::Methods(transactionContext, options) {
-  addHint(transaction::Hints::Hint::INTERMEDIATE_COMMITS);
+  if (options.isIntermediateCommitEnabled()) {
+    addHint(transaction::Hints::Hint::INTERMEDIATE_COMMITS);
+  }
 }
 
   /// protected so we can create different subclasses
@@ -64,7 +66,9 @@ AqlTransaction::AqlTransaction(
     aql::Collections const& collections,
     transaction::Options const& options)
     : transaction::Methods(transactionContext, options) { 
-  addHint(transaction::Hints::Hint::INTERMEDIATE_COMMITS);
+  if (options.isIntermediateCommitEnabled()) {
+    addHint(transaction::Hints::Hint::INTERMEDIATE_COMMITS);
+  }
 
   collections.visit([this](std::string const&, aql::Collection& collection) {
     Result res = processCollection(collection);

diff --git a/arangod/Aql/Ast.cpp b/arangod/Aql/Ast.cpp
@@ -562,6 +562,8 @@ AstNode* Ast::createNodeUpsert(AstNodeType type, AstNode const* docVariable,
   node->addMember(createNodeReference(Variable::NAME_OLD));
   node->addMember(createNodeVariable(TRI_CHAR_LENGTH_PAIR(Variable::NAME_NEW), false));
 
+  this->setContainsUpsertNode();
+
   return node;
 }
 
@@ -4184,6 +4186,12 @@ void Ast::setContainsModificationNode() noexcept {
   _containsModificationNode = true;
 }
 
+bool Ast::containsUpsertNode() const noexcept { return _containsUpsertNode; }
+
+void Ast::setContainsUpsertNode() noexcept {
+  _containsUpsertNode = true;
+}
+
 void Ast::setContainsParallelNode() noexcept {
 #ifdef USE_ENTERPRISE
   _containsParallelNode = true;

diff --git a/arangod/Aql/Ast.h b/arangod/Aql/Ast.h
@@ -129,6 +129,8 @@ class Ast {
   /// @brief whether or not query contains any modification operations
   bool containsModificationNode() const noexcept;
   void setContainsModificationNode() noexcept;
+  bool containsUpsertNode() const noexcept;
+  void setContainsUpsertNode() noexcept;
   void setContainsParallelNode() noexcept;
   bool willUseV8() const noexcept;
   void setWillUseV8() noexcept;
@@ -647,6 +649,8 @@ class Ast {
   /// @brief contains INSERT / UPDATE / REPLACE / REMOVE
   bool _containsModificationNode;
 
+  bool _containsUpsertNode{false};
+
   /// @brief contains a parallel traversal
   bool _containsParallelNode;
 

diff --git a/arangod/Aql/ExecutionPlan.h b/arangod/Aql/ExecutionPlan.h
@@ -173,10 +173,6 @@ class ExecutionPlan {
   void enableAsyncPrefetching() noexcept { _isAsyncPrefetchEnabled = true; }
 
   bool isAsyncPrefetchEnabled() const noexcept { return _isAsyncPrefetchEnabled; }
-
-  void disableIntermediateCommits() noexcept { _isIntermediateCommitAllowed = false; }
-
-  bool isIntermediateCommitAllowed() const noexcept { return _isIntermediateCommitAllowed; }
 
   /// @brief get the node where variable with id <id> is introduced . . .
   ExecutionNode* getVarSetBy(VariableId id) const {
@@ -420,8 +416,6 @@ class ExecutionPlan {
   /// @brief flag to indicate whether the postprocessing step to enable async
   /// prefetching on the node level should be executed.
   bool _isAsyncPrefetchEnabled{false};
-
-  bool _isIntermediateCommitAllowed{true};
 
   /// @brief current nesting level while building the plan
   int _nestingLevel;

diff --git a/arangod/Aql/OptimizerRules.cpp b/arangod/Aql/OptimizerRules.cpp
@@ -7878,13 +7878,6 @@ void arangodb::aql::enableReadOwnWritesForUpsertSubquery(ExecutionPlan& plan) {
   ::arangodb::containers::SmallVector<ExecutionNode*> nodes{a};
   plan.findNodesOfType(nodes, EN::UPSERT, true);
 
-  if (!nodes.empty()) {
-    // UPSERTs and intermediate commits do not play nice together, because the
-    // intermediate commit invalidates the read-own-write iterator required by
-    // the subquery, so we simply disable them.
-    plan.disableIntermediateCommits();
-  }
-
   for (auto n : nodes) {
     auto node = ExecutionNode::castTo<UpsertNode const*>(n);
     auto inputVar = node->inDocVariable();

diff --git a/arangod/Aql/Query.cpp b/arangod/Aql/Query.cpp
@@ -330,6 +330,15 @@ std::unique_ptr<ExecutionPlan> Query::preparePlan() {
   // put in bind parameters
   parser.ast()->injectBindParameters(_bindParameters, this->resolver());
 
+  if (parser.ast()->containsUpsertNode()) {
+    // UPSERTs and intermediate commits do not play nice together, because the
+    // intermediate commit invalidates the read-own-write iterator required by
+    // the subquery. Setting intermediateCommitSize and intermediateCommitCount
+    // to UINT64_MAX allows us to disable intermediate commits.
+    _queryOptions.transactionOptions.intermediateCommitSize = UINT64_MAX;
+    _queryOptions.transactionOptions.intermediateCommitCount = UINT64_MAX;
+  }
+
   TRI_ASSERT(_trx == nullptr);
   // needs to be created after the AST collected all collections
   std::unordered_set<std::string> inaccessibleCollections;
@@ -376,10 +385,6 @@ std::unique_ptr<ExecutionPlan> Query::preparePlan() {
 
   TRI_ASSERT(plan != nullptr);
 
-  if (!plan->isIntermediateCommitAllowed()) {
-    _trx->state()->disableIntermediateCommits();
-  }
-
   // return the V8 context if we are in one
   exitV8Context();
 

diff --git a/arangod/RocksDBEngine/Methods/RocksDBTrxMethods.cpp b/arangod/RocksDBEngine/Methods/RocksDBTrxMethods.cpp
@@ -66,14 +66,6 @@ Result RocksDBTrxMethods::beginTransaction() {
      return result;
 }
 
-void RocksDBTrxMethods::disableIntermediateCommits() {
-  TRI_ASSERT(!hasIntermediateCommitsEnabled());
-  if (_iteratorReadSnapshot) {
-    _db->ReleaseSnapshot(_iteratorReadSnapshot);
-    _iteratorReadSnapshot = nullptr;
-  }
-}
-
 rocksdb::ReadOptions RocksDBTrxMethods::iteratorReadOptions() const {
   if (hasIntermediateCommitsEnabled()) {
     rocksdb::ReadOptions ro = _readOptions;

diff --git a/arangod/RocksDBEngine/Methods/RocksDBTrxMethods.h b/arangod/RocksDBEngine/Methods/RocksDBTrxMethods.h
@@ -40,8 +40,6 @@ class RocksDBTrxMethods : public RocksDBTrxBaseMethods {
 
   Result beginTransaction() override;
 
-  void disableIntermediateCommits() override;
-
   rocksdb::ReadOptions iteratorReadOptions() const override;
 
   void prepareOperation(DataSourceId cid, RevisionId rid, TRI_voc_document_operation_e operationType) override;

diff --git a/arangod/RocksDBEngine/RocksDBTransactionMethods.h b/arangod/RocksDBEngine/RocksDBTransactionMethods.h
@@ -53,8 +53,6 @@ class RocksDBTransactionMethods : public RocksDBMethods {
   virtual Result commitTransaction() = 0;
 
   virtual Result abortTransaction() = 0;
-
-  virtual void disableIntermediateCommits() {}
 
   // Only relevant for RocksDBTrxMethods
   virtual Result checkIntermediateCommit(bool& hasPerformedIntermediateCommit) { return {}; }

diff --git a/arangod/RocksDBEngine/RocksDBTransactionState.cpp b/arangod/RocksDBEngine/RocksDBTransactionState.cpp
@@ -302,13 +302,6 @@ Result RocksDBTransactionState::abortTransaction(transaction::Methods* activeTrx
   return result;
 }
 
-void RocksDBTransactionState::disableIntermediateCommits() {
-  TransactionState::disableIntermediateCommits();
-  if (_rocksMethods) {
-    _rocksMethods->disableIntermediateCommits();
-  }
-}
-
 void RocksDBTransactionState::beginQuery(bool isModificationQuery) {
   auto* trxMethods = dynamic_cast<RocksDBTrxMethods*>(_rocksMethods.get());
   if (trxMethods) {

diff --git a/arangod/RocksDBEngine/RocksDBTransactionState.h b/arangod/RocksDBEngine/RocksDBTransactionState.h
@@ -77,8 +77,6 @@ class RocksDBTransactionState final : public TransactionState {
   /// @brief abort a transaction
   Result abortTransaction(transaction::Methods* trx) override;
 
-  void disableIntermediateCommits() override;
-
   /// @returns tick of last operation in a transaction
   /// @note the value is guaranteed to be valid only after
   ///       transaction is committed

diff --git a/arangod/StorageEngine/TransactionState.h b/arangod/StorageEngine/TransactionState.h
@@ -177,8 +177,6 @@ class TransactionState {
   /// @brief whether or not a specific hint is set for the transaction
   bool hasHint(transaction::Hints::Hint hint) const { return _hints.has(hint); }
 
-  virtual void disableIntermediateCommits() { _hints.unset(transaction::Hints::Hint::INTERMEDIATE_COMMITS); }
-
   /// @brief begin a transaction
   virtual arangodb::Result beginTransaction(transaction::Hints hints) = 0;
 

diff --git a/arangod/Transaction/Options.cpp b/arangod/Transaction/Options.cpp
@@ -88,6 +88,10 @@ void Options::setLimits(uint64_t maxTransactionSize, uint64_t intermediateCommit
   defaultIntermediateCommitCount = intermediateCommitCount;
 }
 
+bool Options::isIntermediateCommitEnabled() const noexcept {
+  return intermediateCommitSize != UINT64_MAX || intermediateCommitCount != UINT64_MAX;
+}
+
 void Options::fromVelocyPack(arangodb::velocypack::Slice const& slice) {
   VPackSlice value;
 

diff --git a/arangod/Transaction/Options.h b/arangod/Transaction/Options.h
@@ -57,6 +57,8 @@ struct Options {
   static void adjustIntermediateCommitCount(Options& options);
 #endif
 
+  bool isIntermediateCommitEnabled() const noexcept;
+
   static constexpr double defaultLockTimeout = 900.0;
   static uint64_t defaultMaxTransactionSize;
   static uint64_t defaultIntermediateCommitSize;