<commit_before>// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.

#include <vespa/fastos/fastos.h>
#include <vespa/log/log.h>
LOG_SETUP(".seach.docstore");

#include "cachestats.h"
#include "documentstore.h"
#include <vespa/vespalib/objects/nbostream.h>
#include <vespa/vespalib/util/atomic.h>

namespace search
{

namespace {

class DocumentVisitorAdapter : public IBufferVisitor
{
public:
    DocumentVisitorAdapter(const document::DocumentTypeRepo & repo, IDocumentVisitor & visitor) :
        _repo(repo),
        _visitor(visitor)
    { }
    void visit(uint32_t lid, vespalib::ConstBufferRef buf) override;
private:
    const document::DocumentTypeRepo & _repo;
    IDocumentVisitor & _visitor;
};

void
DocumentVisitorAdapter::visit(uint32_t lid, vespalib::ConstBufferRef buf) {
    if (buf.size() > 0) {
        vespalib::nbostream is(buf.c_str(), buf.size());
        document::Document::UP doc(new document::Document(_repo, is));
        _visitor.visit(lid, std::move(doc));
    }
}

}

using vespalib::nbostream;

void
DocumentStore::Value::set(vespalib::DataBuffer && buf,
                          ssize_t len,
                          const document::CompressionConfig & compression)
{
    //Underlying buffer must be identical to allow swap.
    vespalib::DataBuffer compressed(buf.getData(), 0u);
    document::CompressionConfig::Type type =
        document::compress(compression,
                           vespalib::ConstBufferRef(buf.getData(), len),
                           compressed, true);
    _compressedSize = compressed.getDataLen();
    if (buf.getData() == compressed.getData()) {
        // Uncompressed so we can just steal the underlying buffer.
        buf.stealBuffer().swap(_buf);
    } else {
        compressed.stealBuffer().swap(_buf);
    }
    assert(((type == document::CompressionConfig::NONE) &&
            (len == ssize_t(_compressedSize))) ||
           ((type != document::CompressionConfig::NONE) &&
            (len > ssize_t(_compressedSize))));
    setCompression(type, len);
}


document::Document::UP
DocumentStore::Value::deserializeDocument(const document::DocumentTypeRepo & repo)
{
    vespalib::DataBuffer uncompressed((char *)_buf.get(), (size_t)0);
    document::decompress(getCompression(),
                         getUncompressedSize(),
                         vespalib::ConstBufferRef(*this, size()),
                         uncompressed, true);
    vespalib::nbostream is(uncompressed.getData(), uncompressed.getDataLen());
    return document::Document::UP(new document::Document(repo, is));
}


void DocumentStore::BackingStore::visit(const LidVector & lids, const document::DocumentTypeRepo &repo, IDocumentVisitor & visitor) const {
    DocumentVisitorAdapter adapter(repo, visitor);
    _backingStore.read(lids, adapter);
}

bool
DocumentStore::BackingStore::read(DocumentIdT key, Value & value) const {
    bool found(false);
    vespalib::DataBuffer buf(4096);
    ssize_t len = _backingStore.read(key, buf);
    if (len > 0) {
        value.set(std::move(buf), len, _compression);
        found = true;
    }
    return found;
}

DocumentStore::DocumentStore(const Config & config, IDataStore & store)
    : IDocumentStore(),
      _config(config),
      _backingStore(store),
      _store(_backingStore, config.getCompression()),
      _cache(new Cache(_store, config.getMaxCacheBytes())),
      _visitCache(new VisitCache(store, config.getMaxCacheBytes(), config.getCompression())),
      _uncached_lookups(0)
{
    _cache->reserveElements(config.getInitialCacheEntries());
}

DocumentStore::~DocumentStore()
{
}


void
DocumentStore::visit(const LidVector & lids, const document::DocumentTypeRepo &repo, IDocumentVisitor & visitor) const
{
    if (useCache() && _config.allowVisitCaching() && visitor.allowVisitCaching()) {
        docstore::BlobSet blobSet = _visitCache->read(lids).getBlobSet();
        DocumentVisitorAdapter adapter(repo, visitor);
        for (DocumentIdT lid : lids) {
            adapter.visit(lid, blobSet.get(lid));
        }
    } else {
        _store.visit(lids, repo, visitor);
    }
}

document::Document::UP
DocumentStore::read(DocumentIdT lid, const document::DocumentTypeRepo &repo) const
{
    document::Document::UP retval;
    Value value;
    if (useCache()) {
        value = _cache->read(lid);
    } else {
        vespalib::Atomic::add(&_uncached_lookups, 1UL);
        _store.read(lid, value);
    }
    if ( ! value.empty() ) {
        retval = value.deserializeDocument(repo);
    }
    return retval;
}

void
DocumentStore::write(uint64_t syncToken, const document::Document& doc, DocumentIdT lid)
{
    nbostream stream(12345);
    doc.serialize(stream);
    _backingStore.write(syncToken, lid, stream.peek(), stream.size());
    if (useCache()) {
        _cache->invalidate(lid);
        _visitCache->invalidate(lid);
    }
}

void
DocumentStore::remove(uint64_t syncToken, DocumentIdT lid)
{
    _backingStore.remove(syncToken, lid);
    if (useCache()) {
        _cache->invalidate(lid);
    }
}

void
DocumentStore::compact(uint64_t syncToken)
{
    (void) syncToken;
    // Most implementations does not offer compact.
}

void
DocumentStore::flush(uint64_t syncToken)
{
    _backingStore.flush(syncToken);
}

uint64_t
DocumentStore::initFlush(uint64_t syncToken)
{
    return _backingStore.initFlush(syncToken);
}

uint64_t
DocumentStore::lastSyncToken() const
{
    return _backingStore.lastSyncToken();
}<commit_msg>Must invalidate on remove too ...<commit_after>// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.

#include <vespa/fastos/fastos.h>
#include <vespa/log/log.h>
LOG_SETUP(".seach.docstore");

#include "cachestats.h"
#include "documentstore.h"
#include <vespa/vespalib/objects/nbostream.h>
#include <vespa/vespalib/util/atomic.h>

namespace search
{

namespace {

class DocumentVisitorAdapter : public IBufferVisitor
{
public:
    DocumentVisitorAdapter(const document::DocumentTypeRepo & repo, IDocumentVisitor & visitor) :
        _repo(repo),
        _visitor(visitor)
    { }
    void visit(uint32_t lid, vespalib::ConstBufferRef buf) override;
private:
    const document::DocumentTypeRepo & _repo;
    IDocumentVisitor & _visitor;
};

void
DocumentVisitorAdapter::visit(uint32_t lid, vespalib::ConstBufferRef buf) {
    if (buf.size() > 0) {
        vespalib::nbostream is(buf.c_str(), buf.size());
        document::Document::UP doc(new document::Document(_repo, is));
        _visitor.visit(lid, std::move(doc));
    }
}

}

using vespalib::nbostream;

void
DocumentStore::Value::set(vespalib::DataBuffer && buf,
                          ssize_t len,
                          const document::CompressionConfig & compression)
{
    //Underlying buffer must be identical to allow swap.
    vespalib::DataBuffer compressed(buf.getData(), 0u);
    document::CompressionConfig::Type type =
        document::compress(compression,
                           vespalib::ConstBufferRef(buf.getData(), len),
                           compressed, true);
    _compressedSize = compressed.getDataLen();
    if (buf.getData() == compressed.getData()) {
        // Uncompressed so we can just steal the underlying buffer.
        buf.stealBuffer().swap(_buf);
    } else {
        compressed.stealBuffer().swap(_buf);
    }
    assert(((type == document::CompressionConfig::NONE) &&
            (len == ssize_t(_compressedSize))) ||
           ((type != document::CompressionConfig::NONE) &&
            (len > ssize_t(_compressedSize))));
    setCompression(type, len);
}


document::Document::UP
DocumentStore::Value::deserializeDocument(const document::DocumentTypeRepo & repo)
{
    vespalib::DataBuffer uncompressed((char *)_buf.get(), (size_t)0);
    document::decompress(getCompression(),
                         getUncompressedSize(),
                         vespalib::ConstBufferRef(*this, size()),
                         uncompressed, true);
    vespalib::nbostream is(uncompressed.getData(), uncompressed.getDataLen());
    return document::Document::UP(new document::Document(repo, is));
}


void DocumentStore::BackingStore::visit(const LidVector & lids, const document::DocumentTypeRepo &repo, IDocumentVisitor & visitor) const {
    DocumentVisitorAdapter adapter(repo, visitor);
    _backingStore.read(lids, adapter);
}

bool
DocumentStore::BackingStore::read(DocumentIdT key, Value & value) const {
    bool found(false);
    vespalib::DataBuffer buf(4096);
    ssize_t len = _backingStore.read(key, buf);
    if (len > 0) {
        value.set(std::move(buf), len, _compression);
        found = true;
    }
    return found;
}

DocumentStore::DocumentStore(const Config & config, IDataStore & store)
    : IDocumentStore(),
      _config(config),
      _backingStore(store),
      _store(_backingStore, config.getCompression()),
      _cache(new Cache(_store, config.getMaxCacheBytes())),
      _visitCache(new VisitCache(store, config.getMaxCacheBytes(), config.getCompression())),
      _uncached_lookups(0)
{
    _cache->reserveElements(config.getInitialCacheEntries());
}

DocumentStore::~DocumentStore()
{
}


void
DocumentStore::visit(const LidVector & lids, const document::DocumentTypeRepo &repo, IDocumentVisitor & visitor) const
{
    if (useCache() && _config.allowVisitCaching() && visitor.allowVisitCaching()) {
        docstore::BlobSet blobSet = _visitCache->read(lids).getBlobSet();
        DocumentVisitorAdapter adapter(repo, visitor);
        for (DocumentIdT lid : lids) {
            adapter.visit(lid, blobSet.get(lid));
        }
    } else {
        _store.visit(lids, repo, visitor);
    }
}

document::Document::UP
DocumentStore::read(DocumentIdT lid, const document::DocumentTypeRepo &repo) const
{
    document::Document::UP retval;
    Value value;
    if (useCache()) {
        value = _cache->read(lid);
    } else {
        vespalib::Atomic::add(&_uncached_lookups, 1UL);
        _store.read(lid, value);
    }
    if ( ! value.empty() ) {
        retval = value.deserializeDocument(repo);
    }
    return retval;
}

void
DocumentStore::write(uint64_t syncToken, const document::Document& doc, DocumentIdT lid)
{
    nbostream stream(12345);
    doc.serialize(stream);
    _backingStore.write(syncToken, lid, stream.peek(), stream.size());
    if (useCache()) {
        _cache->invalidate(lid);
        _visitCache->invalidate(lid);
    }
}

void
DocumentStore::remove(uint64_t syncToken, DocumentIdT lid)
{
    _backingStore.remove(syncToken, lid);
    if (useCache()) {
        _cache->invalidate(lid);
        _visitCache->invalidate(lid);
    }
}

void
DocumentStore::compact(uint64_t syncToken)
{
    (void) syncToken;
    // Most implementations does not offer compact.
}

void
DocumentStore::flush(uint64_t syncToken)
{
    _backingStore.flush(syncToken);
}

uint64_t
DocumentStore::initFlush(uint64_t syncToken)
{
    return _backingStore.initFlush(syncToken);
}

uint64_t
DocumentStore::lastSyncToken() const
{
    return _backingStore.lastSyncToken();
}
{'fraction_non_alphanumeric': 0.12932900432900432, 'fraction_numerical': 0.007034632034632035, 'mean_word_length': 3.610810810810811, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 30, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2649205', 'n_tokens_mistral': 3449, 'n_tokens_neox': 3377, 'n_words': 650}