properly handle removals during recovery (#9528) (#9529)

3rdParty/iresearch/core/index/index_writer.cpp

@@ -722,6 +722,7 @@ index_writer::documents_context::~documents_context() NOEXCEPT {
 }
 
 void index_writer::documents_context::reset() NOEXCEPT {
+  tick_ = 0; // reset tick
   auto& ctx = segment_.ctx();
 
   if (!ctx) {

3rdParty/iresearch/core/index/index_writer.hpp

@@ -190,7 +190,9 @@ class IRESEARCH_API index_writer:
     documents_context(documents_context&& other) NOEXCEPT
       : segment_(std::move(other.segment_)),
         segment_use_count_(std::move(other.segment_use_count_)),
+        tick_(other.tick_),
         writer_(other.writer_) {
+      other.tick_ = 0;
       other.segment_use_count_ = 0;
     }
 

3rdParty/iresearch/core/index/segment_writer.cpp

@@ -328,6 +328,7 @@ void segment_writer::flush(index_meta::index_segment_t& segment) {
 
 void segment_writer::reset() NOEXCEPT {
   initialized_ = false;
+  tick_ = 0;
   dir_.clear_tracked();
   docs_context_.clear();
   docs_mask_.clear();

arangod/IResearch/IResearchLink.cpp

@@ -1074,6 +1074,13 @@ Result IResearchLink::initDataStore(InitCallback const& initCallback, bool sorte
           "failed to get last committed tick while initializing link '" + std::to_string(id()) + "'"
         };
       }
+
+      LOG_TOPIC("7e028", TRACE, iresearch::TOPIC)
+        << "successfully opened existing data store data store reader for link '" + std::to_string(id())
+        << "', docs count '" << _dataStore._reader->docs_count()
+        << "', live docs count '" << _dataStore._reader->live_docs_count()
+        << "', recovery tick '" << _dataStore._recoveryTick << "'";
+
     } catch (irs::index_not_found const&) {
       // NOOP
     }

arangod/IResearch/IResearchRocksDBRecoveryHelper.cpp

@@ -205,54 +205,56 @@ void IResearchRocksDBRecoveryHelper::PutCF(
     const rocksdb::Slice& key,
     const rocksdb::Slice& value,
     rocksdb::SequenceNumber /*tick*/) {
-  if (column_family_id == _documentCF) {
+  if (column_family_id != _documentCF) {
-    auto coll = lookupCollection(*_dbFeature, *_engine, RocksDBKey::objectId(key));
+    return;
+  }
 
-    if (coll == nullptr) {
+  auto coll = lookupCollection(*_dbFeature, *_engine, RocksDBKey::objectId(key));
-      return;
+
+  if (coll == nullptr) {
+    return;
+  }
+
+  auto const links = lookupLinks(*coll);
+
+  if (links.empty()) {
+    return;
+  }
+
+  auto docId = RocksDBKey::documentId(key);
+  auto doc = RocksDBValue::data(value);
+
+  transaction::StandaloneContext ctx(coll->vocbase());
+
+  SingleCollectionTransaction trx(
+    std::shared_ptr<transaction::Context>(
+      std::shared_ptr<transaction::Context>(),
+      &ctx), // aliasing ctor
+    *coll, arangodb::AccessMode::Type::WRITE);
+
+  trx.begin();
+
+  for (std::shared_ptr<arangodb::Index> const& link : links) {
+    IndexId indexId(coll->vocbase().id(), coll->id(), link->id());
+
+    // optimization: avoid insertion of recovered documents twice,
+    //               first insertion done during index creation
+    if (!link || _recoveredIndexes.find(indexId) != _recoveredIndexes.end()) {
+      continue;  // index was already populated when it was created
     }
 
-    auto const links = lookupLinks(*coll);
-
-    if (links.empty()) {
-      return;
-    }
-
-    auto docId = RocksDBKey::documentId(key);
-    auto doc = RocksDBValue::data(value);
-
-    transaction::StandaloneContext ctx(coll->vocbase());
-
-    SingleCollectionTransaction trx(
-      std::shared_ptr<transaction::Context>(
-        std::shared_ptr<transaction::Context>(),
-        &ctx), // aliasing ctor
-      *coll, arangodb::AccessMode::Type::WRITE);
-
-    trx.begin();
-
-    for (std::shared_ptr<arangodb::Index> const& link : links) {
-      IndexId indexId(coll->vocbase().id(), coll->id(), link->id());
-
-      // optimization: avoid insertion of recovered documents twice,
-      //               first insertion done during index creation
-      if (!link || _recoveredIndexes.find(indexId) != _recoveredIndexes.end()) {
-        continue;  // index was already populated when it was created
-      }
-
 #ifdef ARANGODB_ENABLE_MAINTAINER_MODE
     IResearchLink& impl = dynamic_cast<IResearchRocksDBLink&>(*link);
 #else
     IResearchLink& impl = static_cast<IResearchRocksDBLink&>(*link);
 #endif
 
-      impl.insert(trx, docId, doc, arangodb::Index::OperationMode::internal);
+    impl.insert(trx, docId, doc, arangodb::Index::OperationMode::internal);
-    }
-
-    trx.commit();
-
-    return;
   }
+
+  trx.commit();
+
+  return;
 }
 
 // common implementation for DeleteCF / SingleDeleteCF
@@ -260,9 +262,10 @@ void IResearchRocksDBRecoveryHelper::handleDeleteCF(
     uint32_t column_family_id,
     const rocksdb::Slice& key,
     rocksdb::SequenceNumber /*tick*/) {
-  if (column_family_id == _documentCF) {
+  if (column_family_id != _documentCF) {
     return;
   }
+
   auto coll = lookupCollection(*_dbFeature, *_engine, RocksDBKey::objectId(key));
 
   if (coll == nullptr) {