1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#pragma once
#include <vespa/searchcore/proton/reprocessing/reprocessingrunner.h>
#include <vespa/searchcore/proton/bucketdb/bucketdbhandler.h>
#include <vespa/searchcore/proton/common/hw_info.h>
#include <vespa/searchcore/proton/persistenceengine/i_document_retriever.h>
#include <vespa/searchlib/common/serialnum.h>
#include <vespa/vespalib/util/varholder.h>
#include <vespa/vespalib/util/idestructorcallback.h>
#include <mutex>
#include <optional>
namespace vespalib {
class Clock;
class Executor;
class ThreadStackExecutorBase;
}
namespace search {
namespace attribute { class Interlock; }
namespace common { class FileHeaderContext; }
namespace transactionlog { class SyncProxy; }
}
namespace searchcorespi {
class IFlushTarget;
namespace index { struct IThreadingService; }
}
namespace proton {
class DocTypeName;
class DocumentDBConfig;
class DocumentDBReconfig;
class FeedHandler;
class HwInfo;
class IDocumentRetriever;
class IDocumentSubDB;
class IDocumentSubDBOwner;
class IFeedView;
class IGetSerialNum;
class MaintenanceController;
class ReconfigParams;
class RemoveDocumentsOperation;
struct DocumentDBTaggedMetrics;
struct IBucketStateCalculator;
struct IDocumentDBReferenceResolver;
struct MetricsWireService;
namespace matching { class QueryLimiter; }
namespace initializer { class InitializerTask; }
namespace index { struct IndexConfig; }
class DocumentSubDBCollection {
public:
using SubDBVector = std::vector<IDocumentSubDB *>;
using const_iterator = SubDBVector::const_iterator;
using SerialNum = search::SerialNum;
using OnDone = std::shared_ptr<vespalib::IDestructorCallback>;
private:
using IFeedViewSP = std::shared_ptr<IFeedView>;
using IBucketStateCalculatorSP = std::shared_ptr<IBucketStateCalculator>;
using IFlushTargetList = std::vector<std::shared_ptr<searchcorespi::IFlushTarget>>;
SubDBVector _subDBs;
IDocumentSubDBOwner &_owner;
IBucketStateCalculatorSP _calc;
const uint32_t _readySubDbId;
const uint32_t _remSubDbId;
const uint32_t _notReadySubDbId;
using RetrieversSP = std::shared_ptr<std::vector<std::shared_ptr<IDocumentRetriever>> >;
vespalib::VarHolder<RetrieversSP> _retrievers;
ReprocessingRunner _reprocessingRunner;
std::shared_ptr<bucketdb::BucketDBOwner> _bucketDB;
std::unique_ptr<bucketdb::BucketDBHandler> _bucketDBHandler;
HwInfo _hwInfo;
public:
DocumentSubDBCollection(
IDocumentSubDBOwner &owner,
search::transactionlog::SyncProxy &tlSyncer,
const IGetSerialNum &getSerialNum,
const DocTypeName &docTypeName,
searchcorespi::index::IThreadingService &writeService,
vespalib::Executor &warmupExecutor,
const search::common::FileHeaderContext &fileHeaderContext,
std::shared_ptr<search::attribute::Interlock> attribute_interlock,
MetricsWireService &metricsWireService,
DocumentDBTaggedMetrics &metrics,
matching::QueryLimiter & queryLimiter,
const vespalib::Clock &clock,
std::mutex &configMutex,
const vespalib::string &baseDir,
const HwInfo &hwInfo);
~DocumentSubDBCollection();
void setBucketStateCalculator(const IBucketStateCalculatorSP &calc, OnDone onDone);
void createRetrievers();
void maintenanceSync(MaintenanceController &mc);
// Internally synchronized
RetrieversSP getRetrievers(IDocumentRetriever::ReadConsistency consistency);
IDocumentSubDB *getReadySubDB() { return _subDBs[_readySubDbId]; }
const IDocumentSubDB *getReadySubDB() const { return _subDBs[_readySubDbId]; }
IDocumentSubDB *getRemSubDB() { return _subDBs[_remSubDbId]; }
const IDocumentSubDB *getRemSubDB() const { return _subDBs[_remSubDbId]; }
IDocumentSubDB *getNotReadySubDB() { return _subDBs[_notReadySubDbId]; }
const IDocumentSubDB *getNotReadySubDB() const { return _subDBs[_notReadySubDbId]; }
const_iterator begin() const { return _subDBs.begin(); }
const_iterator end() const { return _subDBs.end(); }
bucketdb::BucketDBOwner &getBucketDB() { return *_bucketDB; }
const bucketdb::BucketDBOwner &getBucketDB() const { return *_bucketDB; }
bucketdb::IBucketDBHandler &getBucketDBHandler() {
return *_bucketDBHandler;
}
bucketdb::IBucketCreateNotifier &getBucketCreateNotifier() {
return _bucketDBHandler->getBucketCreateNotifier();
}
std::shared_ptr<initializer::InitializerTask>
createInitializer(const DocumentDBConfig &configSnapshot, SerialNum configSerialNum,
const index::IndexConfig & indexCfg);
void initViews(const DocumentDBConfig &configSnapshot);
void clearViews();
void onReplayDone();
void onReprocessDone(SerialNum serialNum);
SerialNum getOldestFlushedSerial();
SerialNum getNewestFlushedSerial();
void pruneRemovedFields(SerialNum serialNum);
std::unique_ptr<DocumentDBReconfig> prepare_reconfig(const DocumentDBConfig& new_config_snapshot, const ReconfigParams& reconfig_params, std::optional<SerialNum> serial_num);
void complete_prepare_reconfig(DocumentDBReconfig& prepared_reconfig, SerialNum serial_num);
void applyConfig(const DocumentDBConfig &newConfigSnapshot, const DocumentDBConfig &oldConfigSnapshot,
SerialNum serialNum, const ReconfigParams ¶ms, IDocumentDBReferenceResolver &resolver, const DocumentDBReconfig& prepared_reconfig);
IFeedViewSP getFeedView();
IFlushTargetList getFlushTargets();
ReprocessingRunner &getReprocessingRunner() { return _reprocessingRunner; }
double getReprocessingProgress() const;
void close();
void tearDownReferences(IDocumentDBReferenceResolver &resolver);
void validateDocStore(FeedHandler & feedHandler, SerialNum serialNum);
};
} // namespace proton
|
