Keep document live during attribute population until attribute write threads

no longer uses fields retrieved from the document.

13 files changed, 48 insertions, 32 deletions

diff --git a/searchcore/src/tests/proton/attribute/attribute_populator/attribute_populator_test.cpp b/searchcore/src/tests/proton/attribute/attribute_populator/attribute_populator_test.cpp
index 7255e899621..ed71477d853 100644
--- a/searchcore/src/tests/proton/attribute/attribute_populator/attribute_populator_test.cpp
+++ b/searchcore/src/tests/proton/attribute/attribute_populator/attribute_populator_test.cpp
@@ -44,7 +44,7 @@ struct DocContext
           _builder(_schema)
     {
     }
-    Document::UP create(uint32_t id, int64_t fieldValue) {
+    std::shared_ptr<Document> create(uint32_t id, int64_t fieldValue) {
         vespalib::string docId =
                 vespalib::make_string("id:searchdocument:searchdocument::%u", id);
         return _builder.startDocument(docId).
@@ -87,12 +87,12 @@ TEST_F("require that reprocess with document populates attribute", Fixture)
     AttributeGuard::UP attr = f.getAttr();
     EXPECT_EQUAL(1u, attr->get()->getNumDocs());
 
-    f._pop->handleExisting(5, *f._ctx.create(0, 33));
+    f._pop->handleExisting(5, f._ctx.create(0, 33));
     EXPECT_EQUAL(6u, attr->get()->getNumDocs());
     EXPECT_EQUAL(33, attr->get()->getInt(5));
     EXPECT_EQUAL(1u, attr->get()->getStatus().getLastSyncToken());
 
-    f._pop->handleExisting(6, *f._ctx.create(1, 44));
+    f._pop->handleExisting(6, f._ctx.create(1, 44));
     EXPECT_EQUAL(7u, attr->get()->getNumDocs());
     EXPECT_EQUAL(44, attr->get()->getInt(6));
     EXPECT_EQUAL(2u, attr->get()->getStatus().getLastSyncToken());
diff --git a/searchcore/src/tests/proton/attribute/document_field_populator/document_field_populator_test.cpp b/searchcore/src/tests/proton/attribute/document_field_populator/document_field_populator_test.cpp
index 24de9d3bd47..a684f24f20d 100644
--- a/searchcore/src/tests/proton/attribute/document_field_populator/document_field_populator_test.cpp
+++ b/searchcore/src/tests/proton/attribute/document_field_populator/document_field_populator_test.cpp
@@ -71,8 +71,8 @@ TEST_F("require that document field is populated based on attribute content", Fi
     // NOTE: DocumentFieldRetriever (used by DocumentFieldPopulator) is fully tested
     // with all data types in searchcore/src/tests/proton/server/documentretriever_test.cpp.
     {
-        Document::UP doc = f._ctx.create(1);
-        f._pop.handleExisting(1, *doc);
+        std::shared_ptr<Document> doc = f._ctx.create(1);
+        f._pop.handleExisting(1, doc);
         EXPECT_EQUAL(100, doc->getValue("a1")->getAsInt());
     }
 }
diff --git a/searchcore/src/tests/proton/reprocessing/document_reprocessing_handler/document_reprocessing_handler_test.cpp b/searchcore/src/tests/proton/reprocessing/document_reprocessing_handler/document_reprocessing_handler_test.cpp
index 2454a797749..17f53661916 100644
--- a/searchcore/src/tests/proton/reprocessing/document_reprocessing_handler/document_reprocessing_handler_test.cpp
+++ b/searchcore/src/tests/proton/reprocessing/document_reprocessing_handler/document_reprocessing_handler_test.cpp
@@ -10,22 +10,22 @@ using namespace document;
 using namespace proton;
 using namespace search::index;
 
-template <typename ReprocessingType, typename DocumentType>
+template <typename ReprocessingType>
 struct MyProcessor : public ReprocessingType
 {
-    typedef std::shared_ptr<MyProcessor<ReprocessingType, DocumentType> > SP;
+    typedef std::shared_ptr<MyProcessor<ReprocessingType> > SP;
     uint32_t _lid;
     DocumentId _docId;
 
     MyProcessor() : _lid(0), _docId() {}
-    virtual void handleExisting(uint32_t lid, DocumentType doc) override {
+    virtual void handleExisting(uint32_t lid, const std::shared_ptr<Document> &doc) override {
         _lid = lid;
-        _docId = doc.getId();
+        _docId = doc->getId();
     }
 };
 
-typedef MyProcessor<IReprocessingReader, const Document &> MyReader;
-typedef MyProcessor<IReprocessingRewriter, Document &> MyRewriter;
+typedef MyProcessor<IReprocessingReader> MyReader;
+typedef MyProcessor<IReprocessingRewriter> MyRewriter;
 
 const vespalib::string DOC_ID = "id:test:searchdocument::0";
 
@@ -35,7 +35,7 @@ struct FixtureBase
     DocBuilder _docBuilder;
     FixtureBase(uint32_t docIdLimit);
     ~FixtureBase();
-    Document::UP createDoc() {
+    std::shared_ptr<Document> createDoc() {
         return _docBuilder.startDocument(DOC_ID).endDocument();
     }
 };
@@ -84,7 +84,7 @@ struct RewriterFixture : public FixtureBase
 
 TEST_F("require that handler propagates visit of existing document to readers", ReaderFixture)
 {
-    f._handler.visit(23u, *f.createDoc());
+    f._handler.visit(23u, f.createDoc());
     EXPECT_EQUAL(23u, f._reader1->_lid);
     EXPECT_EQUAL(DOC_ID, f._reader1->_docId.toString());
     EXPECT_EQUAL(23u, f._reader2->_lid);
@@ -93,7 +93,7 @@ TEST_F("require that handler propagates visit of existing document to readers",
 
 TEST_F("require that handler propagates visit of existing document to rewriters", RewriterFixture)
 {
-    f._handler.getRewriteVisitor().visit(23u, *f.createDoc());
+    f._handler.getRewriteVisitor().visit(23u, f.createDoc());
     EXPECT_EQUAL(23u, f._rewriter1->_lid);
     EXPECT_EQUAL(DOC_ID, f._rewriter1->_docId.toString());
     EXPECT_EQUAL(23u, f._rewriter2->_lid);
@@ -103,7 +103,7 @@ TEST_F("require that handler propagates visit of existing document to rewriters"
 TEST_F("require that handler skips out of range visit to readers",
        ReaderFixture(10))
 {
-    f._handler.visit(23u, *f.createDoc());
+    f._handler.visit(23u, f.createDoc());
     EXPECT_EQUAL(0u, f._reader1->_lid);
     EXPECT_EQUAL(DocumentId().toString(), f._reader1->_docId.toString());
     EXPECT_EQUAL(0u, f._reader2->_lid);
@@ -113,7 +113,7 @@ TEST_F("require that handler skips out of range visit to readers",
 TEST_F("require that handler skips out of range visit to rewriters",
        RewriterFixture(10))
 {
-    f._handler.getRewriteVisitor().visit(23u, *f.createDoc());
+    f._handler.getRewriteVisitor().visit(23u, f.createDoc());
     EXPECT_EQUAL(0u, f._rewriter1->_lid);
     EXPECT_EQUAL(DocumentId().toString(), f._rewriter1->_docId.toString());
     EXPECT_EQUAL(0u, f._rewriter2->_lid);
diff --git a/searchcore/src/vespa/searchcore/proton/attribute/attribute_populator.cpp b/searchcore/src/vespa/searchcore/proton/attribute/attribute_populator.cpp
index bd072ddca9c..91e935fe7c0 100644
--- a/searchcore/src/vespa/searchcore/proton/attribute/attribute_populator.cpp
+++ b/searchcore/src/vespa/searchcore/proton/attribute/attribute_populator.cpp
@@ -12,6 +12,21 @@ using search::IDestructorCallback;
 
 namespace proton {
 
+namespace {
+
+class PopulateDoneContext : public IDestructorCallback
+{
+    std::shared_ptr<document::Document> _doc;
+public:
+    PopulateDoneContext(const std::shared_ptr<document::Document> &doc)
+        : _doc(doc)
+    {
+    }
+    ~PopulateDoneContext() override = default;
+};
+
+}
+
 search::SerialNum
 AttributePopulator::nextSerialNum()
 {
@@ -55,10 +70,11 @@ AttributePopulator::~AttributePopulator()
 }
 
 void
-AttributePopulator::handleExisting(uint32_t lid, const document::Document &doc)
+AttributePopulator::handleExisting(uint32_t lid, const std::shared_ptr<document::Document> &doc)
 {
     search::SerialNum serialNum(nextSerialNum());
-    _writer.put(serialNum, doc, lid, true, std::shared_ptr<IDestructorCallback>());
+    auto populateDoneContext = std::make_shared<PopulateDoneContext>(doc);
+    _writer.put(serialNum, *doc, lid, true, populateDoneContext);
 }
 
 void
diff --git a/searchcore/src/vespa/searchcore/proton/attribute/attribute_populator.h b/searchcore/src/vespa/searchcore/proton/attribute/attribute_populator.h
index f11bf5aab47..eae3329fe5a 100644
--- a/searchcore/src/vespa/searchcore/proton/attribute/attribute_populator.h
+++ b/searchcore/src/vespa/searchcore/proton/attribute/attribute_populator.h
@@ -35,7 +35,7 @@ public:
     const IAttributeWriter &getWriter() const { return _writer; }
 
     // Implements IReprocessingReader
-    virtual void handleExisting(uint32_t lid, const document::Document &doc) override;
+    virtual void handleExisting(uint32_t lid, const std::shared_ptr<document::Document> &doc) override;
     virtual void done() override;
 };
 
diff --git a/searchcore/src/vespa/searchcore/proton/attribute/document_field_populator.cpp b/searchcore/src/vespa/searchcore/proton/attribute/document_field_populator.cpp
index 9564c241e26..b530cffed4f 100644
--- a/searchcore/src/vespa/searchcore/proton/attribute/document_field_populator.cpp
+++ b/searchcore/src/vespa/searchcore/proton/attribute/document_field_populator.cpp
@@ -45,9 +45,9 @@ DocumentFieldPopulator::~DocumentFieldPopulator()
 }
 
 void
-DocumentFieldPopulator::handleExisting(uint32_t lid, Document &doc)
+DocumentFieldPopulator::handleExisting(uint32_t lid, const std::shared_ptr<Document> &doc)
 {
-    DocumentFieldRetriever::populate(lid, doc, _fieldName, *_attr, false);
+    DocumentFieldRetriever::populate(lid, *doc, _fieldName, *_attr, false);
     ++_documentsPopulated;
 }
 
diff --git a/searchcore/src/vespa/searchcore/proton/attribute/document_field_populator.h b/searchcore/src/vespa/searchcore/proton/attribute/document_field_populator.h
index e6efbd9c275..29975b286e6 100644
--- a/searchcore/src/vespa/searchcore/proton/attribute/document_field_populator.h
+++ b/searchcore/src/vespa/searchcore/proton/attribute/document_field_populator.h
@@ -31,7 +31,7 @@ public:
     }
 
     // Implements IReprocessingRewriter
-    virtual void handleExisting(uint32_t lid, document::Document &doc) override;
+    virtual void handleExisting(uint32_t lid, const std::shared_ptr<document::Document> &doc) override;
 };
 
 } // namespace proton
diff --git a/searchcore/src/vespa/searchcore/proton/reprocessing/document_reprocessing_handler.cpp b/searchcore/src/vespa/searchcore/proton/reprocessing/document_reprocessing_handler.cpp
index 73f7840699b..885f577ab4e 100644
--- a/searchcore/src/vespa/searchcore/proton/reprocessing/document_reprocessing_handler.cpp
+++ b/searchcore/src/vespa/searchcore/proton/reprocessing/document_reprocessing_handler.cpp
@@ -8,7 +8,7 @@ LOG_SETUP(".proton.reprocessing.document_reprocessing_handler");
 namespace proton {
 
 void
-DocumentReprocessingHandler::rewriteVisit(uint32_t lid, document::Document &doc)
+DocumentReprocessingHandler::rewriteVisit(uint32_t lid, const std::shared_ptr<document::Document> &doc)
 {
     if (lid == 0 || lid >= _docIdLimit)
         return;
@@ -30,7 +30,7 @@ DocumentReprocessingHandler::DocumentReprocessingHandler(uint32_t docIdLimit)
 DocumentReprocessingHandler::~DocumentReprocessingHandler() {}
 
 void
-DocumentReprocessingHandler::visit(uint32_t lid, const document::Document &doc)
+DocumentReprocessingHandler::visit(uint32_t lid, const std::shared_ptr<document::Document> &doc)
 {
     if (lid == 0 || lid >= _docIdLimit)
         return;
diff --git a/searchcore/src/vespa/searchcore/proton/reprocessing/document_reprocessing_handler.h b/searchcore/src/vespa/searchcore/proton/reprocessing/document_reprocessing_handler.h
index b6c60ec6512..87963ed968f 100644
--- a/searchcore/src/vespa/searchcore/proton/reprocessing/document_reprocessing_handler.h
+++ b/searchcore/src/vespa/searchcore/proton/reprocessing/document_reprocessing_handler.h
@@ -25,7 +25,7 @@ private:
     public:
         RewriteVisitor(DocumentReprocessingHandler &handler) : _handler(handler) {}
         // Implements search::IDocumentStoreRewriteVisitor
-        virtual void visit(uint32_t lid, document::Document &doc) override {
+        virtual void visit(uint32_t lid, const std::shared_ptr<document::Document> &doc) override {
             _handler.rewriteVisit(lid, doc);
         }
     };
@@ -35,7 +35,7 @@ private:
     RewriteVisitor _rewriteVisitor;
     uint32_t       _docIdLimit;
 
-    void rewriteVisit(uint32_t lid, document::Document &doc);
+    void rewriteVisit(uint32_t lid, const std::shared_ptr<document::Document> &doc);
 
 public:
     DocumentReprocessingHandler(uint32_t docIdLimit);
@@ -67,7 +67,7 @@ public:
     }
 
     // Implements search::IDocumentStoreReadVisitor
-    virtual void visit(uint32_t lid, const document::Document &doc) override;
+    virtual void visit(uint32_t lid, const std::shared_ptr<document::Document> &doc) override;
 
     virtual void visit(uint32_t lid) override;
 
diff --git a/searchcore/src/vespa/searchcore/proton/reprocessing/i_reprocessing_reader.h b/searchcore/src/vespa/searchcore/proton/reprocessing/i_reprocessing_reader.h
index cc083e291cc..a05fa327082 100644
--- a/searchcore/src/vespa/searchcore/proton/reprocessing/i_reprocessing_reader.h
+++ b/searchcore/src/vespa/searchcore/proton/reprocessing/i_reprocessing_reader.h
@@ -19,7 +19,7 @@ struct IReprocessingReader
     /**
      * Handle the given existing document.
      */
-    virtual void handleExisting(uint32_t lid, const document::Document &doc) = 0;
+    virtual void handleExisting(uint32_t lid, const std::shared_ptr<document::Document> &doc) = 0;
 
     // signals that there are no more documents
     virtual void done() {}
diff --git a/searchcore/src/vespa/searchcore/proton/reprocessing/i_reprocessing_rewriter.h b/searchcore/src/vespa/searchcore/proton/reprocessing/i_reprocessing_rewriter.h
index ca09fb61ad1..85e692f6e0b 100644
--- a/searchcore/src/vespa/searchcore/proton/reprocessing/i_reprocessing_rewriter.h
+++ b/searchcore/src/vespa/searchcore/proton/reprocessing/i_reprocessing_rewriter.h
@@ -19,7 +19,7 @@ struct IReprocessingRewriter
     /**
      * Handle and rewrite the given existing document.
      */
-    virtual void handleExisting(uint32_t lid, document::Document &doc) = 0;
+    virtual void handleExisting(uint32_t lid, const std::shared_ptr<document::Document> &doc) = 0;
 };
 
 } // namespace proton
diff --git a/searchcore/src/vespa/searchcore/proton/server/docstorevalidator.cpp b/searchcore/src/vespa/searchcore/proton/server/docstorevalidator.cpp
index df540298ae7..1c3a5c94733 100644
--- a/searchcore/src/vespa/searchcore/proton/server/docstorevalidator.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/docstorevalidator.cpp
@@ -23,7 +23,7 @@ DocStoreValidator::DocStoreValidator(IDocumentMetaStore &dms)
 
 
 void
-DocStoreValidator::visit(uint32_t lid, const document::Document &doc)
+DocStoreValidator::visit(uint32_t lid, const std::shared_ptr<document::Document> &doc)
 {
     if (lid == 0 || lid >= _docIdLimit)
         return;
@@ -32,7 +32,7 @@ DocStoreValidator::visit(uint32_t lid, const document::Document &doc)
         _orphans->setBit(lid);
         return;
     }
-    const document::DocumentId &docId(doc.getId());
+    const document::DocumentId &docId(doc->getId());
     const document::GlobalId &gid = docId.getGlobalId();
     const RawDocumentMetaData &meta = _dms.getRawMetaData(lid);
     const document::GlobalId &dmsGid = meta.getGid();
diff --git a/searchcore/src/vespa/searchcore/proton/server/docstorevalidator.h b/searchcore/src/vespa/searchcore/proton/server/docstorevalidator.h
index 60b059a2bf6..fba57d4f718 100644
--- a/searchcore/src/vespa/searchcore/proton/server/docstorevalidator.h
+++ b/searchcore/src/vespa/searchcore/proton/server/docstorevalidator.h
@@ -21,7 +21,7 @@ class DocStoreValidator : public search::IDocumentStoreReadVisitor
 public:
     DocStoreValidator(IDocumentMetaStore &dms);
 
-    virtual void visit(uint32_t lid, const document::Document &doc) override;
+    virtual void visit(uint32_t lid, const std::shared_ptr<document::Document> &doc) override;
     virtual void visit(uint32_t lid) override;
 
     void visitDone();