only sync RocksDB WAL in MMFiles engine when there is a reason to do so (#2863)

arangod/MMFiles/MMFilesTransactionState.cpp

@@ -478,8 +478,22 @@ int MMFilesTransactionState::writeCommitMarker() {
     TRI_IF_FAILURE("TransactionWriteCommitMarkerNoRocksSync") { return TRI_ERROR_NO_ERROR; }
 
     if (_options.waitForSync) {
-      // also sync RocksDB WAL
-      MMFilesPersistentIndexFeature::syncWal();
+      // also sync RocksDB WAL if required
+      bool hasPersistentIndex = false;
+      allCollections([&hasPersistentIndex](TransactionCollection* collection) {
+        auto c = static_cast<MMFilesTransactionCollection*>(collection);
+        if (c->canAccess(AccessMode::Type::WRITE) && 
+            c->collection()->getPhysical()->hasIndexOfType(arangodb::Index::TRI_IDX_TYPE_PERSISTENT_INDEX)) {
+          hasPersistentIndex = true;
+          // abort iteration
+          return false;
+        }
+
+        return true;
+      });
+      if (hasPersistentIndex) {
+        MMFilesPersistentIndexFeature::syncWal();
+      }
     }
     
     TRI_IF_FAILURE("TransactionWriteCommitMarkerThrow") { 

arangod/StorageEngine/TransactionState.cpp

@@ -198,6 +198,16 @@ int TransactionState::addCollection(TRI_voc_cid_t cid,
 Result TransactionState::ensureCollections(int nestingLevel) {
   return useCollections(nestingLevel);
 }
+  
+/// @brief run a callback on all collections
+void TransactionState::allCollections(std::function<bool(TransactionCollection*)> const& cb) {
+  for (auto& trxCollection : _collections) {
+    if (!cb(trxCollection)) {
+      // abort early
+      return;
+    }
+  }
+}
 
 /// @brief use all participating collections of a transaction
 Result TransactionState::useCollections(int nestingLevel) {

arangod/StorageEngine/TransactionState.h

@@ -123,6 +123,9 @@ class TransactionState {
   /// @brief use all participating collections of a transaction
   Result useCollections(int nestingLevel);
 
+  /// @brief run a callback on all collections
+  void allCollections(std::function<bool(TransactionCollection*)> const& cb);
+
   /// @brief release collection locks for a transaction
   int unuseCollections(int nestingLevel);