diff options
author | Rémi Nicole <nicole@kolabsystems.com> | 2018-08-22 14:16:59 +0200 |
---|---|---|
committer | Christian Mollekopf <chrigi_1@fastmail.fm> | 2018-08-22 14:28:51 +0200 |
commit | 46313049ac01a3007ef60bdc937442945355a38d (patch) | |
tree | 56ce0cd679367a60ba3a706ac4d207bc9cc82230 /common | |
parent | af91a18748b91f4a4fc0d83247561371d376bec5 (diff) | |
download | sink-46313049ac01a3007ef60bdc937442945355a38d.tar.gz sink-46313049ac01a3007ef60bdc937442945355a38d.zip |
Separate UIDs and Revisions in main databases
Summary:
- Change revision type from `qint64` to `size_t` for LMDB in a couple of places (LMDB supports `unsigned int` or `size_t` which are `long unsigned int` on my machine)
- Better support for database flags (duplicate, integer keys, integer values for now but is extensible)
- Main databases' keys are now revisions
- Some databases switched to integer keys databases:
- Main databases
- the revision to uid mapping database
- the revision to entity type mapping database
- Refactor the entity type's `typeDatabases` method (if in the future we need to change the main databases' flags again)
- New uid to revision mapping database (`uidsToRevisions`):
- Stores all revisions (not uid to latest revision) because we need it for cleaning old revisions
- Flags are: duplicates + integer values (so findLatest finds the latest revision for the given uid)
~~Problems to fix before merging:~~
All Fixed!
- ~~Sometimes Sink can't read what has just been written to the database (maybe because of transactions race conditions)~~
- ~~Most of the times, this results in Sink not able to find the uid for a given revision by reading the `revisions` database~~
- ~~`pipelinetest`'s `testModifyWithConflict` fails because the local changes are overridden~~
~~The first problem prevents me from running benchmarks~~
Reviewers: cmollekopf
Tags: #sink
Differential Revision: https://phabricator.kde.org/D14974
Diffstat (limited to 'common')
-rw-r--r-- | common/changereplay.cpp | 7 | ||||
-rw-r--r-- | common/domain/typeimplementations.cpp | 20 | ||||
-rw-r--r-- | common/domain/typeimplementations_p.h | 10 | ||||
-rw-r--r-- | common/index.cpp | 8 | ||||
-rw-r--r-- | common/mail/threadindexer.cpp | 4 | ||||
-rw-r--r-- | common/storage.h | 40 | ||||
-rw-r--r-- | common/storage/entitystore.cpp | 160 | ||||
-rw-r--r-- | common/storage/key.cpp | 5 | ||||
-rw-r--r-- | common/storage/key.h | 5 | ||||
-rw-r--r-- | common/storage_common.cpp | 105 | ||||
-rw-r--r-- | common/storage_lmdb.cpp | 136 | ||||
-rw-r--r-- | common/utils.cpp | 10 | ||||
-rw-r--r-- | common/utils.h | 14 |
13 files changed, 344 insertions, 180 deletions
diff --git a/common/changereplay.cpp b/common/changereplay.cpp index d7f46dc..96162b8 100644 --- a/common/changereplay.cpp +++ b/common/changereplay.cpp | |||
@@ -116,16 +116,15 @@ KAsync::Job<void> ChangeReplay::replayNextRevision() | |||
116 | } else { | 116 | } else { |
117 | // TODO: should not use internal representations | 117 | // TODO: should not use internal representations |
118 | const auto key = Storage::Key(Storage::Identifier::fromDisplayByteArray(uid), revision); | 118 | const auto key = Storage::Key(Storage::Identifier::fromDisplayByteArray(uid), revision); |
119 | const auto internalKey = key.toInternalByteArray(); | ||
120 | const auto displayKey = key.toDisplayByteArray(); | 119 | const auto displayKey = key.toDisplayByteArray(); |
121 | QByteArray entityBuffer; | 120 | QByteArray entityBuffer; |
122 | DataStore::mainDatabase(mMainStoreTransaction, type) | 121 | DataStore::mainDatabase(mMainStoreTransaction, type) |
123 | .scan(internalKey, | 122 | .scan(revision, |
124 | [&entityBuffer](const QByteArray &key, const QByteArray &value) -> bool { | 123 | [&entityBuffer](const size_t, const QByteArray &value) -> bool { |
125 | entityBuffer = value; | 124 | entityBuffer = value; |
126 | return false; | 125 | return false; |
127 | }, | 126 | }, |
128 | [this, key](const DataStore::Error &) { SinkErrorCtx(mLogCtx) << "Failed to read the entity buffer " << key; }); | 127 | [this, key](const DataStore::Error &e) { SinkErrorCtx(mLogCtx) << "Failed to read the entity buffer " << key << "error:" << e; }); |
129 | 128 | ||
130 | if (entityBuffer.isEmpty()) { | 129 | if (entityBuffer.isEmpty()) { |
131 | SinkErrorCtx(mLogCtx) << "Failed to replay change " << key; | 130 | SinkErrorCtx(mLogCtx) << "Failed to replay change " << key; |
diff --git a/common/domain/typeimplementations.cpp b/common/domain/typeimplementations.cpp index aedf889..6e14501 100644 --- a/common/domain/typeimplementations.cpp +++ b/common/domain/typeimplementations.cpp | |||
@@ -80,7 +80,11 @@ typedef IndexConfig<Calendar, | |||
80 | ValueIndex<Calendar::Name> | 80 | ValueIndex<Calendar::Name> |
81 | > CalendarIndexConfig; | 81 | > CalendarIndexConfig; |
82 | 82 | ||
83 | 83 | template <typename EntityType, typename EntityIndexConfig> | |
84 | QMap<QByteArray, int> defaultTypeDatabases() | ||
85 | { | ||
86 | return merge(QMap<QByteArray, int>{{QByteArray{EntityType::name} + ".main", Storage::IntegerKeys}}, EntityIndexConfig::databases()); | ||
87 | } | ||
84 | 88 | ||
85 | void TypeImplementation<Mail>::configure(TypeIndex &index) | 89 | void TypeImplementation<Mail>::configure(TypeIndex &index) |
86 | { | 90 | { |
@@ -89,7 +93,7 @@ void TypeImplementation<Mail>::configure(TypeIndex &index) | |||
89 | 93 | ||
90 | QMap<QByteArray, int> TypeImplementation<Mail>::typeDatabases() | 94 | QMap<QByteArray, int> TypeImplementation<Mail>::typeDatabases() |
91 | { | 95 | { |
92 | return merge(QMap<QByteArray, int>{{QByteArray{Mail::name} + ".main", 0}}, MailIndexConfig::databases()); | 96 | return defaultTypeDatabases<Mail, MailIndexConfig>(); |
93 | } | 97 | } |
94 | 98 | ||
95 | void TypeImplementation<Mail>::configure(IndexPropertyMapper &indexPropertyMapper) | 99 | void TypeImplementation<Mail>::configure(IndexPropertyMapper &indexPropertyMapper) |
@@ -132,7 +136,7 @@ void TypeImplementation<Folder>::configure(TypeIndex &index) | |||
132 | 136 | ||
133 | QMap<QByteArray, int> TypeImplementation<Folder>::typeDatabases() | 137 | QMap<QByteArray, int> TypeImplementation<Folder>::typeDatabases() |
134 | { | 138 | { |
135 | return merge(QMap<QByteArray, int>{{QByteArray{Folder::name} + ".main", 0}}, FolderIndexConfig::databases()); | 139 | return defaultTypeDatabases<Folder, FolderIndexConfig>(); |
136 | } | 140 | } |
137 | 141 | ||
138 | void TypeImplementation<Folder>::configure(PropertyMapper &propertyMapper) | 142 | void TypeImplementation<Folder>::configure(PropertyMapper &propertyMapper) |
@@ -157,7 +161,7 @@ void TypeImplementation<Contact>::configure(TypeIndex &index) | |||
157 | 161 | ||
158 | QMap<QByteArray, int> TypeImplementation<Contact>::typeDatabases() | 162 | QMap<QByteArray, int> TypeImplementation<Contact>::typeDatabases() |
159 | { | 163 | { |
160 | return merge(QMap<QByteArray, int>{{QByteArray{Contact::name} + ".main", 0}}, ContactIndexConfig::databases()); | 164 | return defaultTypeDatabases<Contact, ContactIndexConfig>(); |
161 | } | 165 | } |
162 | 166 | ||
163 | void TypeImplementation<Contact>::configure(PropertyMapper &propertyMapper) | 167 | void TypeImplementation<Contact>::configure(PropertyMapper &propertyMapper) |
@@ -185,7 +189,7 @@ void TypeImplementation<Addressbook>::configure(TypeIndex &index) | |||
185 | 189 | ||
186 | QMap<QByteArray, int> TypeImplementation<Addressbook>::typeDatabases() | 190 | QMap<QByteArray, int> TypeImplementation<Addressbook>::typeDatabases() |
187 | { | 191 | { |
188 | return merge(QMap<QByteArray, int>{{QByteArray{Addressbook::name} + ".main", 0}}, AddressbookIndexConfig::databases()); | 192 | return defaultTypeDatabases<Addressbook, AddressbookIndexConfig>(); |
189 | } | 193 | } |
190 | 194 | ||
191 | void TypeImplementation<Addressbook>::configure(PropertyMapper &propertyMapper) | 195 | void TypeImplementation<Addressbook>::configure(PropertyMapper &propertyMapper) |
@@ -207,7 +211,7 @@ void TypeImplementation<Event>::configure(TypeIndex &index) | |||
207 | 211 | ||
208 | QMap<QByteArray, int> TypeImplementation<Event>::typeDatabases() | 212 | QMap<QByteArray, int> TypeImplementation<Event>::typeDatabases() |
209 | { | 213 | { |
210 | return merge(QMap<QByteArray, int>{{QByteArray{Event::name} + ".main", 0}}, EventIndexConfig::databases()); | 214 | return defaultTypeDatabases<Event, EventIndexConfig>(); |
211 | } | 215 | } |
212 | 216 | ||
213 | void TypeImplementation<Event>::configure(PropertyMapper &propertyMapper) | 217 | void TypeImplementation<Event>::configure(PropertyMapper &propertyMapper) |
@@ -235,7 +239,7 @@ void TypeImplementation<Todo>::configure(TypeIndex &index) | |||
235 | 239 | ||
236 | QMap<QByteArray, int> TypeImplementation<Todo>::typeDatabases() | 240 | QMap<QByteArray, int> TypeImplementation<Todo>::typeDatabases() |
237 | { | 241 | { |
238 | return merge(QMap<QByteArray, int>{{QByteArray{Todo::name} + ".main", 0}}, TodoIndexConfig::databases()); | 242 | return defaultTypeDatabases<Todo, TodoIndexConfig>(); |
239 | } | 243 | } |
240 | 244 | ||
241 | void TypeImplementation<Todo>::configure(PropertyMapper &propertyMapper) | 245 | void TypeImplementation<Todo>::configure(PropertyMapper &propertyMapper) |
@@ -266,7 +270,7 @@ void TypeImplementation<Calendar>::configure(TypeIndex &index) | |||
266 | 270 | ||
267 | QMap<QByteArray, int> TypeImplementation<Calendar>::typeDatabases() | 271 | QMap<QByteArray, int> TypeImplementation<Calendar>::typeDatabases() |
268 | { | 272 | { |
269 | return merge(QMap<QByteArray, int>{{QByteArray{Calendar::name} + ".main", 0}}, CalendarIndexConfig::databases()); | 273 | return defaultTypeDatabases<Calendar, CalendarIndexConfig>(); |
270 | } | 274 | } |
271 | 275 | ||
272 | void TypeImplementation<Calendar>::configure(PropertyMapper &propertyMapper) | 276 | void TypeImplementation<Calendar>::configure(PropertyMapper &propertyMapper) |
diff --git a/common/domain/typeimplementations_p.h b/common/domain/typeimplementations_p.h index 51af113..bfdea77 100644 --- a/common/domain/typeimplementations_p.h +++ b/common/domain/typeimplementations_p.h | |||
@@ -57,7 +57,7 @@ public: | |||
57 | template <typename EntityType> | 57 | template <typename EntityType> |
58 | static QMap<QByteArray, int> databases() | 58 | static QMap<QByteArray, int> databases() |
59 | { | 59 | { |
60 | return {{QByteArray{EntityType::name} +".index." + Property::name, 1}}; | 60 | return {{QByteArray{EntityType::name} +".index." + Property::name, Sink::Storage::AllowDuplicates}}; |
61 | } | 61 | } |
62 | }; | 62 | }; |
63 | 63 | ||
@@ -74,7 +74,7 @@ public: | |||
74 | template <typename EntityType> | 74 | template <typename EntityType> |
75 | static QMap<QByteArray, int> databases() | 75 | static QMap<QByteArray, int> databases() |
76 | { | 76 | { |
77 | return {{QByteArray{EntityType::name} +".index." + Property::name + ".sort." + SortProperty::name, 1}}; | 77 | return {{QByteArray{EntityType::name} +".index." + Property::name + ".sort." + SortProperty::name, Sink::Storage::AllowDuplicates}}; |
78 | } | 78 | } |
79 | }; | 79 | }; |
80 | 80 | ||
@@ -90,7 +90,7 @@ public: | |||
90 | template <typename EntityType> | 90 | template <typename EntityType> |
91 | static QMap<QByteArray, int> databases() | 91 | static QMap<QByteArray, int> databases() |
92 | { | 92 | { |
93 | return {{QByteArray{EntityType::name} +".index." + SortProperty::name + ".sorted", 1}}; | 93 | return {{QByteArray{EntityType::name} +".index." + SortProperty::name + ".sorted", Sink::Storage::AllowDuplicates}}; |
94 | } | 94 | } |
95 | }; | 95 | }; |
96 | 96 | ||
@@ -106,7 +106,7 @@ public: | |||
106 | template <typename EntityType> | 106 | template <typename EntityType> |
107 | static QMap<QByteArray, int> databases() | 107 | static QMap<QByteArray, int> databases() |
108 | { | 108 | { |
109 | return {{QByteArray{EntityType::name} +".index." + Property::name + SecondaryProperty::name, 1}}; | 109 | return {{QByteArray{EntityType::name} +".index." + Property::name + SecondaryProperty::name, Sink::Storage::AllowDuplicates}}; |
110 | } | 110 | } |
111 | }; | 111 | }; |
112 | 112 | ||
@@ -142,7 +142,7 @@ public: | |||
142 | template <typename EntityType> | 142 | template <typename EntityType> |
143 | static QMap<QByteArray, int> databases() | 143 | static QMap<QByteArray, int> databases() |
144 | { | 144 | { |
145 | return {{QByteArray{EntityType::name} +".index." + RangeBeginProperty::name + ".range." + RangeEndProperty::name, 1}}; | 145 | return {{QByteArray{EntityType::name} +".index." + RangeBeginProperty::name + ".range." + RangeEndProperty::name, Sink::Storage::AllowDuplicates}}; |
146 | } | 146 | } |
147 | }; | 147 | }; |
148 | 148 | ||
diff --git a/common/index.cpp b/common/index.cpp index 238a745..bf8fcfc 100644 --- a/common/index.cpp +++ b/common/index.cpp | |||
@@ -6,7 +6,7 @@ using Sink::Storage::Identifier; | |||
6 | 6 | ||
7 | Index::Index(const QString &storageRoot, const QString &dbName, const QString &indexName, Sink::Storage::DataStore::AccessMode mode) | 7 | Index::Index(const QString &storageRoot, const QString &dbName, const QString &indexName, Sink::Storage::DataStore::AccessMode mode) |
8 | : mTransaction(Sink::Storage::DataStore(storageRoot, dbName, mode).createTransaction(mode)), | 8 | : mTransaction(Sink::Storage::DataStore(storageRoot, dbName, mode).createTransaction(mode)), |
9 | mDb(mTransaction.openDatabase(indexName.toLatin1(), std::function<void(const Sink::Storage::DataStore::Error &)>(), true)), | 9 | mDb(mTransaction.openDatabase(indexName.toLatin1(), std::function<void(const Sink::Storage::DataStore::Error &)>(), Sink::Storage::AllowDuplicates)), |
10 | mName(indexName), | 10 | mName(indexName), |
11 | mLogCtx("index." + indexName.toLatin1()) | 11 | mLogCtx("index." + indexName.toLatin1()) |
12 | { | 12 | { |
@@ -14,7 +14,7 @@ Index::Index(const QString &storageRoot, const QString &dbName, const QString &i | |||
14 | 14 | ||
15 | Index::Index(const QString &storageRoot, const QString &name, Sink::Storage::DataStore::AccessMode mode) | 15 | Index::Index(const QString &storageRoot, const QString &name, Sink::Storage::DataStore::AccessMode mode) |
16 | : mTransaction(Sink::Storage::DataStore(storageRoot, name, mode).createTransaction(mode)), | 16 | : mTransaction(Sink::Storage::DataStore(storageRoot, name, mode).createTransaction(mode)), |
17 | mDb(mTransaction.openDatabase(name.toLatin1(), std::function<void(const Sink::Storage::DataStore::Error &)>(), true)), | 17 | mDb(mTransaction.openDatabase(name.toLatin1(), std::function<void(const Sink::Storage::DataStore::Error &)>(), Sink::Storage::AllowDuplicates)), |
18 | mName(name), | 18 | mName(name), |
19 | mLogCtx("index." + name.toLatin1()) | 19 | mLogCtx("index." + name.toLatin1()) |
20 | { | 20 | { |
@@ -22,14 +22,14 @@ Index::Index(const QString &storageRoot, const QString &name, Sink::Storage::Dat | |||
22 | 22 | ||
23 | Index::Index(const QString &storageRoot, const Sink::Storage::DbLayout &layout, Sink::Storage::DataStore::AccessMode mode) | 23 | Index::Index(const QString &storageRoot, const Sink::Storage::DbLayout &layout, Sink::Storage::DataStore::AccessMode mode) |
24 | : mTransaction(Sink::Storage::DataStore(storageRoot, layout, mode).createTransaction(mode)), | 24 | : mTransaction(Sink::Storage::DataStore(storageRoot, layout, mode).createTransaction(mode)), |
25 | mDb(mTransaction.openDatabase(layout.name, std::function<void(const Sink::Storage::DataStore::Error &)>(), true)), | 25 | mDb(mTransaction.openDatabase(layout.name, std::function<void(const Sink::Storage::DataStore::Error &)>(), Sink::Storage::AllowDuplicates)), |
26 | mName(layout.name), | 26 | mName(layout.name), |
27 | mLogCtx("index." + layout.name) | 27 | mLogCtx("index." + layout.name) |
28 | { | 28 | { |
29 | } | 29 | } |
30 | 30 | ||
31 | Index::Index(const QByteArray &name, Sink::Storage::DataStore::Transaction &transaction) | 31 | Index::Index(const QByteArray &name, Sink::Storage::DataStore::Transaction &transaction) |
32 | : mDb(transaction.openDatabase(name, std::function<void(const Sink::Storage::DataStore::Error &)>(), true)), mName(name), | 32 | : mDb(transaction.openDatabase(name, std::function<void(const Sink::Storage::DataStore::Error &)>(), Sink::Storage::AllowDuplicates)), mName(name), |
33 | mLogCtx("index." + name) | 33 | mLogCtx("index." + name) |
34 | { | 34 | { |
35 | } | 35 | } |
diff --git a/common/mail/threadindexer.cpp b/common/mail/threadindexer.cpp index c1d1aa8..b9de266 100644 --- a/common/mail/threadindexer.cpp +++ b/common/mail/threadindexer.cpp | |||
@@ -118,7 +118,7 @@ void ThreadIndexer::remove(const ApplicationDomain::ApplicationDomainType &entit | |||
118 | 118 | ||
119 | QMap<QByteArray, int> ThreadIndexer::databases() | 119 | QMap<QByteArray, int> ThreadIndexer::databases() |
120 | { | 120 | { |
121 | return {{"mail.index.messageIdthreadId", 1}, | 121 | return {{"mail.index.messageIdthreadId", Sink::Storage::AllowDuplicates}, |
122 | {"mail.index.threadIdmessageId", 1}}; | 122 | {"mail.index.threadIdmessageId", Sink::Storage::AllowDuplicates}}; |
123 | } | 123 | } |
124 | 124 | ||
diff --git a/common/storage.h b/common/storage.h index 8904148..ac6509d 100644 --- a/common/storage.h +++ b/common/storage.h | |||
@@ -32,6 +32,11 @@ | |||
32 | namespace Sink { | 32 | namespace Sink { |
33 | namespace Storage { | 33 | namespace Storage { |
34 | 34 | ||
35 | extern int AllowDuplicates; | ||
36 | extern int IntegerKeys; | ||
37 | // Only useful with AllowDuplicates | ||
38 | extern int IntegerValues; | ||
39 | |||
35 | struct SINK_EXPORT DbLayout { | 40 | struct SINK_EXPORT DbLayout { |
36 | typedef QMap<QByteArray, int> Databases; | 41 | typedef QMap<QByteArray, int> Databases; |
37 | DbLayout(); | 42 | DbLayout(); |
@@ -80,15 +85,23 @@ public: | |||
80 | */ | 85 | */ |
81 | bool write(const QByteArray &key, const QByteArray &value, const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>()); | 86 | bool write(const QByteArray &key, const QByteArray &value, const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>()); |
82 | 87 | ||
88 | // of QByteArray for keys | ||
89 | bool write(const size_t key, const QByteArray &value, const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>()); | ||
90 | |||
83 | /** | 91 | /** |
84 | * Remove a key | 92 | * Remove a key |
85 | */ | 93 | */ |
86 | void remove(const QByteArray &key, const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>()); | 94 | void remove(const QByteArray &key, const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>()); |
95 | |||
96 | void remove(const size_t key, const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>()); | ||
97 | |||
87 | /** | 98 | /** |
88 | * Remove a key-value pair | 99 | * Remove a key-value pair |
89 | */ | 100 | */ |
90 | void remove(const QByteArray &key, const QByteArray &value, const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>()); | 101 | void remove(const QByteArray &key, const QByteArray &value, const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>()); |
91 | 102 | ||
103 | void remove(const size_t key, const QByteArray &value, const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>()); | ||
104 | |||
92 | /** | 105 | /** |
93 | * Read values with a given key. | 106 | * Read values with a given key. |
94 | * | 107 | * |
@@ -101,6 +114,9 @@ public: | |||
101 | int scan(const QByteArray &key, const std::function<bool(const QByteArray &key, const QByteArray &value)> &resultHandler, | 114 | int scan(const QByteArray &key, const std::function<bool(const QByteArray &key, const QByteArray &value)> &resultHandler, |
102 | const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>(), bool findSubstringKeys = false, bool skipInternalKeys = true) const; | 115 | const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>(), bool findSubstringKeys = false, bool skipInternalKeys = true) const; |
103 | 116 | ||
117 | int scan(const size_t key, const std::function<bool(size_t key, const QByteArray &value)> &resultHandler, | ||
118 | const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>(), bool skipInternalKeys = true) const; | ||
119 | |||
104 | /** | 120 | /** |
105 | * Finds the last value in a series matched by prefix. | 121 | * Finds the last value in a series matched by prefix. |
106 | * | 122 | * |
@@ -110,6 +126,9 @@ public: | |||
110 | void findLatest(const QByteArray &uid, const std::function<void(const QByteArray &key, const QByteArray &value)> &resultHandler, | 126 | void findLatest(const QByteArray &uid, const std::function<void(const QByteArray &key, const QByteArray &value)> &resultHandler, |
111 | const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>()) const; | 127 | const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>()) const; |
112 | 128 | ||
129 | void findLatest(size_t key, const std::function<void(size_t key, const QByteArray &value)> &resultHandler, | ||
130 | const std::function<void(const DataStore::Error &error)> &errorHandler = std::function<void(const DataStore::Error &error)>()) const; | ||
131 | |||
113 | /** | 132 | /** |
114 | * Finds all the keys and values whose keys are in a given range | 133 | * Finds all the keys and values whose keys are in a given range |
115 | * (inclusive). | 134 | * (inclusive). |
@@ -119,6 +138,10 @@ public: | |||
119 | const std::function<void(const DataStore::Error &error)> &errorHandler = | 138 | const std::function<void(const DataStore::Error &error)> &errorHandler = |
120 | std::function<void(const DataStore::Error &error)>()) const; | 139 | std::function<void(const DataStore::Error &error)>()) const; |
121 | 140 | ||
141 | int findAllInRange(const size_t lowerBound, const size_t upperBound, | ||
142 | const std::function<void(size_t key, const QByteArray &value)> &resultHandler, | ||
143 | const std::function<void(const DataStore::Error &error)> &errorHandler = {}) const; | ||
144 | |||
122 | /** | 145 | /** |
123 | * Returns true if the database contains the substring key. | 146 | * Returns true if the database contains the substring key. |
124 | */ | 147 | */ |
@@ -163,8 +186,9 @@ public: | |||
163 | 186 | ||
164 | QList<QByteArray> getDatabaseNames() const; | 187 | QList<QByteArray> getDatabaseNames() const; |
165 | 188 | ||
166 | NamedDatabase openDatabase(const QByteArray &name = {"default"}, | 189 | NamedDatabase openDatabase(const QByteArray &name = { "default" }, |
167 | const std::function<void(const DataStore::Error &error)> &errorHandler = {}, bool allowDuplicates = false) const; | 190 | const std::function<void(const DataStore::Error &error)> &errorHandler = {}, |
191 | int flags = 0) const; | ||
168 | 192 | ||
169 | Transaction(Transaction &&other); | 193 | Transaction(Transaction &&other); |
170 | Transaction &operator=(Transaction &&other); | 194 | Transaction &operator=(Transaction &&other); |
@@ -224,13 +248,17 @@ public: | |||
224 | static qint64 cleanedUpRevision(const Transaction &); | 248 | static qint64 cleanedUpRevision(const Transaction &); |
225 | static void setCleanedUpRevision(Transaction &, qint64 revision); | 249 | static void setCleanedUpRevision(Transaction &, qint64 revision); |
226 | 250 | ||
227 | static QByteArray getUidFromRevision(const Transaction &, qint64 revision); | 251 | static QByteArray getUidFromRevision(const Transaction &, size_t revision); |
228 | static QByteArray getTypeFromRevision(const Transaction &, qint64 revision); | 252 | static size_t getLatestRevisionFromUid(Transaction &, const QByteArray &uid); |
229 | static void recordRevision(Transaction &, qint64 revision, const QByteArray &uid, const QByteArray &type); | 253 | static QList<size_t> getRevisionsUntilFromUid(DataStore::Transaction &, const QByteArray &uid, size_t lastRevision); |
230 | static void removeRevision(Transaction &, qint64 revision); | 254 | static QList<size_t> getRevisionsFromUid(DataStore::Transaction &, const QByteArray &uid); |
255 | static QByteArray getTypeFromRevision(const Transaction &, size_t revision); | ||
256 | static void recordRevision(Transaction &, size_t revision, const QByteArray &uid, const QByteArray &type); | ||
257 | static void removeRevision(Transaction &, size_t revision); | ||
231 | static void recordUid(DataStore::Transaction &transaction, const QByteArray &uid, const QByteArray &type); | 258 | static void recordUid(DataStore::Transaction &transaction, const QByteArray &uid, const QByteArray &type); |
232 | static void removeUid(DataStore::Transaction &transaction, const QByteArray &uid, const QByteArray &type); | 259 | static void removeUid(DataStore::Transaction &transaction, const QByteArray &uid, const QByteArray &type); |
233 | static void getUids(const QByteArray &type, const Transaction &, const std::function<void(const QByteArray &uid)> &); | 260 | static void getUids(const QByteArray &type, const Transaction &, const std::function<void(const QByteArray &uid)> &); |
261 | static bool hasUid(const QByteArray &type, const Transaction &, const QByteArray &uid); | ||
234 | 262 | ||
235 | bool exists() const; | 263 | bool exists() const; |
236 | static bool exists(const QString &storageRoot, const QString &name); | 264 | static bool exists(const QString &storageRoot, const QString &name); |
diff --git a/common/storage/entitystore.cpp b/common/storage/entitystore.cpp index 276ee6a..454e25a 100644 --- a/common/storage/entitystore.cpp +++ b/common/storage/entitystore.cpp | |||
@@ -38,8 +38,9 @@ using namespace Sink::Storage; | |||
38 | 38 | ||
39 | static QMap<QByteArray, int> baseDbs() | 39 | static QMap<QByteArray, int> baseDbs() |
40 | { | 40 | { |
41 | return {{"revisionType", 0}, | 41 | return {{"revisionType", Storage::IntegerKeys}, |
42 | {"revisions", 0}, | 42 | {"revisions", Storage::IntegerKeys}, |
43 | {"uidsToRevisions", Storage::AllowDuplicates | Storage::IntegerValues}, | ||
43 | {"uids", 0}, | 44 | {"uids", 0}, |
44 | {"default", 0}, | 45 | {"default", 0}, |
45 | {"__flagtable", 0}}; | 46 | {"__flagtable", 0}}; |
@@ -242,12 +243,13 @@ bool EntityStore::add(const QByteArray &type, ApplicationDomainType entity, bool | |||
242 | const auto key = Key(identifier, newRevision); | 243 | const auto key = Key(identifier, newRevision); |
243 | 244 | ||
244 | DataStore::mainDatabase(d->transaction, type) | 245 | DataStore::mainDatabase(d->transaction, type) |
245 | .write(key.toInternalByteArray(), BufferUtils::extractBuffer(fbb), | 246 | .write(newRevision, BufferUtils::extractBuffer(fbb), |
246 | [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to write entity" << entity.identifier() << newRevision; }); | 247 | [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to write entity" << entity.identifier() << newRevision; }); |
248 | |||
247 | DataStore::setMaxRevision(d->transaction, newRevision); | 249 | DataStore::setMaxRevision(d->transaction, newRevision); |
248 | DataStore::recordRevision(d->transaction, newRevision, entity.identifier(), type); | 250 | DataStore::recordRevision(d->transaction, newRevision, entity.identifier(), type); |
249 | DataStore::recordUid(d->transaction, entity.identifier(), type); | 251 | DataStore::recordUid(d->transaction, entity.identifier(), type); |
250 | SinkTraceCtx(d->logCtx) << "Wrote entity: " << entity.identifier() << type << newRevision; | 252 | SinkTraceCtx(d->logCtx) << "Wrote entity: " << key << "of type:" << type; |
251 | return true; | 253 | return true; |
252 | } | 254 | } |
253 | 255 | ||
@@ -319,8 +321,9 @@ bool EntityStore::modify(const QByteArray &type, const ApplicationDomainType &cu | |||
319 | const auto key = Key(identifier, newRevision); | 321 | const auto key = Key(identifier, newRevision); |
320 | 322 | ||
321 | DataStore::mainDatabase(d->transaction, type) | 323 | DataStore::mainDatabase(d->transaction, type) |
322 | .write(key.toInternalByteArray(), BufferUtils::extractBuffer(fbb), | 324 | .write(newRevision, BufferUtils::extractBuffer(fbb), |
323 | [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to write entity" << newEntity.identifier() << newRevision; }); | 325 | [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to write entity" << newEntity.identifier() << newRevision; }); |
326 | |||
324 | DataStore::setMaxRevision(d->transaction, newRevision); | 327 | DataStore::setMaxRevision(d->transaction, newRevision); |
325 | DataStore::recordRevision(d->transaction, newRevision, newEntity.identifier(), type); | 328 | DataStore::recordRevision(d->transaction, newRevision, newEntity.identifier(), type); |
326 | SinkTraceCtx(d->logCtx) << "Wrote modified entity: " << newEntity.identifier() << type << newRevision; | 329 | SinkTraceCtx(d->logCtx) << "Wrote modified entity: " << newEntity.identifier() << type << newRevision; |
@@ -356,8 +359,9 @@ bool EntityStore::remove(const QByteArray &type, const ApplicationDomainType &cu | |||
356 | const auto key = Key(identifier, newRevision); | 359 | const auto key = Key(identifier, newRevision); |
357 | 360 | ||
358 | DataStore::mainDatabase(d->transaction, type) | 361 | DataStore::mainDatabase(d->transaction, type) |
359 | .write(key.toInternalByteArray(), BufferUtils::extractBuffer(fbb), | 362 | .write(newRevision, BufferUtils::extractBuffer(fbb), |
360 | [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to write entity" << uid << newRevision; }); | 363 | [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to write entity" << uid << newRevision; }); |
364 | |||
361 | DataStore::setMaxRevision(d->transaction, newRevision); | 365 | DataStore::setMaxRevision(d->transaction, newRevision); |
362 | DataStore::recordRevision(d->transaction, newRevision, uid, type); | 366 | DataStore::recordRevision(d->transaction, newRevision, uid, type); |
363 | DataStore::removeUid(d->transaction, uid, type); | 367 | DataStore::removeUid(d->transaction, uid, type); |
@@ -375,30 +379,33 @@ void EntityStore::cleanupEntityRevisionsUntil(qint64 revision) | |||
375 | } | 379 | } |
376 | SinkTraceCtx(d->logCtx) << "Cleaning up revision " << revision << uid << bufferType; | 380 | SinkTraceCtx(d->logCtx) << "Cleaning up revision " << revision << uid << bufferType; |
377 | const auto internalUid = Identifier::fromDisplayByteArray(uid).toInternalByteArray(); | 381 | const auto internalUid = Identifier::fromDisplayByteArray(uid).toInternalByteArray(); |
378 | DataStore::mainDatabase(d->transaction, bufferType) | ||
379 | .scan(internalUid, | ||
380 | [&](const QByteArray &key, const QByteArray &data) -> bool { | ||
381 | EntityBuffer buffer(const_cast<const char *>(data.data()), data.size()); | ||
382 | if (!buffer.isValid()) { | ||
383 | SinkWarningCtx(d->logCtx) << "Read invalid buffer from disk"; | ||
384 | } else { | ||
385 | const auto metadata = flatbuffers::GetRoot<Metadata>(buffer.metadataBuffer()); | ||
386 | const qint64 rev = metadata->revision(); | ||
387 | const auto isRemoval = metadata->operation() == Operation_Removal; | ||
388 | // Remove old revisions, and the current if the entity has already been removed | ||
389 | if (rev < revision || isRemoval) { | ||
390 | DataStore::removeRevision(d->transaction, rev); | ||
391 | DataStore::mainDatabase(d->transaction, bufferType).remove(key); | ||
392 | } | ||
393 | //Don't cleanup more than specified | ||
394 | if (rev >= revision) { | ||
395 | return false; | ||
396 | } | ||
397 | } | ||
398 | 382 | ||
399 | return true; | 383 | // Remove old revisions |
400 | }, | 384 | const auto revisionsToRemove = DataStore::getRevisionsUntilFromUid(d->transaction, uid, revision); |
401 | [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Error while reading: " << error.message; }, true); | 385 | |
386 | for (const auto &revisionToRemove : revisionsToRemove) { | ||
387 | DataStore::removeRevision(d->transaction, revisionToRemove); | ||
388 | DataStore::mainDatabase(d->transaction, bufferType).remove(revisionToRemove); | ||
389 | } | ||
390 | |||
391 | // And remove the specified revision only if marked for removal | ||
392 | DataStore::mainDatabase(d->transaction, bufferType).scan(revision, [&](size_t, const QByteArray &data) { | ||
393 | EntityBuffer buffer(const_cast<const char *>(data.data()), data.size()); | ||
394 | if (!buffer.isValid()) { | ||
395 | SinkWarningCtx(d->logCtx) << "Read invalid buffer from disk"; | ||
396 | return false; | ||
397 | } | ||
398 | |||
399 | const auto metadata = flatbuffers::GetRoot<Metadata>(buffer.metadataBuffer()); | ||
400 | const qint64 rev = metadata->revision(); | ||
401 | if (metadata->operation() == Operation_Removal) { | ||
402 | DataStore::removeRevision(d->transaction, revision); | ||
403 | DataStore::mainDatabase(d->transaction, bufferType).remove(revision); | ||
404 | } | ||
405 | |||
406 | return false; | ||
407 | }); | ||
408 | |||
402 | DataStore::setCleanedUpRevision(d->transaction, revision); | 409 | DataStore::setCleanedUpRevision(d->transaction, revision); |
403 | } | 410 | } |
404 | 411 | ||
@@ -433,20 +440,12 @@ QVector<Identifier> EntityStore::fullScan(const QByteArray &type) | |||
433 | SinkTraceCtx(d->logCtx) << "Database is not existing: " << type; | 440 | SinkTraceCtx(d->logCtx) << "Database is not existing: " << type; |
434 | return {}; | 441 | return {}; |
435 | } | 442 | } |
436 | //The scan can return duplicate results if we have multiple revisions, so we use a set to deduplicate. | 443 | |
437 | QSet<Identifier> keys; | 444 | QSet<Identifier> keys; |
438 | DataStore::mainDatabase(d->getTransaction(), type) | 445 | |
439 | .scan(QByteArray(), | 446 | DataStore::getUids(type, d->getTransaction(), [&keys] (const QByteArray &uid) { |
440 | [&](const QByteArray &key, const QByteArray &value) -> bool { | 447 | keys << Identifier::fromDisplayByteArray(uid); |
441 | const auto uid = Sink::Storage::Key::fromInternalByteArray(key).identifier(); | 448 | }); |
442 | if (keys.contains(uid)) { | ||
443 | //Not something that should persist if the replay works, so we keep a message for now. | ||
444 | SinkTraceCtx(d->logCtx) << "Multiple revisions for uid: " << Sink::Storage::Key::fromInternalByteArray(key) << ". This is normal if changereplay has not completed yet."; | ||
445 | } | ||
446 | keys << uid; | ||
447 | return true; | ||
448 | }, | ||
449 | [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Error during fullScan query: " << error.message; }); | ||
450 | 449 | ||
451 | SinkTraceCtx(d->logCtx) << "Full scan retrieved " << keys.size() << " results."; | 450 | SinkTraceCtx(d->logCtx) << "Full scan retrieved " << keys.size() << " results."; |
452 | return keys.toList().toVector(); | 451 | return keys.toList().toVector(); |
@@ -492,12 +491,12 @@ void EntityStore::indexLookup(const QByteArray &type, const QByteArray &property | |||
492 | void EntityStore::readLatest(const QByteArray &type, const Identifier &id, const std::function<void(const QByteArray &uid, const EntityBuffer &entity)> callback) | 491 | void EntityStore::readLatest(const QByteArray &type, const Identifier &id, const std::function<void(const QByteArray &uid, const EntityBuffer &entity)> callback) |
493 | { | 492 | { |
494 | Q_ASSERT(d); | 493 | Q_ASSERT(d); |
495 | const auto internalKey = id.toInternalByteArray(); | 494 | const size_t revision = DataStore::getLatestRevisionFromUid(d->getTransaction(), id.toDisplayByteArray()); |
496 | auto db = DataStore::mainDatabase(d->getTransaction(), type); | 495 | auto db = DataStore::mainDatabase(d->getTransaction(), type); |
497 | db.findLatest(internalKey, | 496 | db.scan(revision, |
498 | [=](const QByteArray &key, const QByteArray &value) { | 497 | [=](size_t, const QByteArray &value) { |
499 | const auto uid = Sink::Storage::Key::fromInternalByteArray(key).identifier().toDisplayByteArray(); | 498 | callback(id.toDisplayByteArray(), Sink::EntityBuffer(value.data(), value.size())); |
500 | callback(uid, Sink::EntityBuffer(value.data(), value.size())); | 499 | return false; |
501 | }, | 500 | }, |
502 | [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Error during readLatest query: " << error.message << id; }); | 501 | [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Error during readLatest query: " << error.message << id; }); |
503 | } | 502 | } |
@@ -546,9 +545,9 @@ void EntityStore::readEntity(const QByteArray &type, const QByteArray &displayKe | |||
546 | { | 545 | { |
547 | const auto key = Key::fromDisplayByteArray(displayKey); | 546 | const auto key = Key::fromDisplayByteArray(displayKey); |
548 | auto db = DataStore::mainDatabase(d->getTransaction(), type); | 547 | auto db = DataStore::mainDatabase(d->getTransaction(), type); |
549 | db.scan(key.toInternalByteArray(), | 548 | db.scan(key.revision().toSizeT(), |
550 | [=](const QByteArray &key, const QByteArray &value) -> bool { | 549 | [=](size_t rev, const QByteArray &value) -> bool { |
551 | const auto uid = Sink::Storage::Key::fromInternalByteArray(key).identifier().toDisplayByteArray(); | 550 | const auto uid = DataStore::getUidFromRevision(d->transaction, rev); |
552 | callback(uid, Sink::EntityBuffer(value.data(), value.size())); | 551 | callback(uid, Sink::EntityBuffer(value.data(), value.size())); |
553 | return false; | 552 | return false; |
554 | }, | 553 | }, |
@@ -604,18 +603,8 @@ void EntityStore::readRevisions(qint64 baseRevision, const QByteArray &expectedT | |||
604 | 603 | ||
605 | void EntityStore::readPrevious(const QByteArray &type, const Identifier &id, qint64 revision, const std::function<void(const QByteArray &uid, const EntityBuffer &entity)> callback) | 604 | void EntityStore::readPrevious(const QByteArray &type, const Identifier &id, qint64 revision, const std::function<void(const QByteArray &uid, const EntityBuffer &entity)> callback) |
606 | { | 605 | { |
607 | auto db = DataStore::mainDatabase(d->getTransaction(), type); | 606 | const auto previousRevisions = DataStore::getRevisionsUntilFromUid(d->getTransaction(), id.toDisplayByteArray(), revision); |
608 | qint64 latestRevision = 0; | 607 | const size_t latestRevision = previousRevisions[previousRevisions.size() - 1]; |
609 | const auto internalUid = id.toInternalByteArray(); | ||
610 | db.scan(internalUid, | ||
611 | [&latestRevision, revision](const QByteArray &key, const QByteArray &) -> bool { | ||
612 | const auto foundRevision = Key::fromInternalByteArray(key).revision().toQint64(); | ||
613 | if (foundRevision < revision && foundRevision > latestRevision) { | ||
614 | latestRevision = foundRevision; | ||
615 | } | ||
616 | return true; | ||
617 | }, | ||
618 | [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to read current value from storage: " << error.message; }, true); | ||
619 | const auto key = Key(id, latestRevision); | 608 | const auto key = Key(id, latestRevision); |
620 | readEntity(type, key.toDisplayByteArray(), callback); | 609 | readEntity(type, key.toDisplayByteArray(), callback); |
621 | } | 610 | } |
@@ -641,21 +630,20 @@ void EntityStore::readAllUids(const QByteArray &type, const std::function<void(c | |||
641 | DataStore::getUids(type, d->getTransaction(), callback); | 630 | DataStore::getUids(type, d->getTransaction(), callback); |
642 | } | 631 | } |
643 | 632 | ||
644 | bool EntityStore::contains(const QByteArray &type, const QByteArray &uid) | 633 | bool EntityStore::contains(const QByteArray & /* type */, const QByteArray &uid) |
645 | { | 634 | { |
646 | Q_ASSERT(!uid.isEmpty()); | 635 | Q_ASSERT(!uid.isEmpty()); |
647 | const auto internalUid = Identifier::fromDisplayByteArray(uid).toInternalByteArray(); | 636 | return !DataStore::getRevisionsFromUid(d->getTransaction(), uid).isEmpty(); |
648 | return DataStore::mainDatabase(d->getTransaction(), type).contains(internalUid); | ||
649 | } | 637 | } |
650 | 638 | ||
651 | bool EntityStore::exists(const QByteArray &type, const QByteArray &uid) | 639 | bool EntityStore::exists(const QByteArray &type, const QByteArray &uid) |
652 | { | 640 | { |
653 | bool found = false; | 641 | bool found = false; |
654 | bool alreadyRemoved = false; | 642 | bool alreadyRemoved = false; |
655 | const auto internalUid = Identifier::fromDisplayByteArray(uid).toInternalByteArray(); | 643 | const size_t revision = DataStore::getLatestRevisionFromUid(d->getTransaction(), uid); |
656 | DataStore::mainDatabase(d->transaction, type) | 644 | DataStore::mainDatabase(d->transaction, type) |
657 | .findLatest(internalUid, | 645 | .scan(revision, |
658 | [&found, &alreadyRemoved](const QByteArray &key, const QByteArray &data) { | 646 | [&found, &alreadyRemoved](size_t, const QByteArray &data) { |
659 | auto entity = GetEntity(data.data()); | 647 | auto entity = GetEntity(data.data()); |
660 | if (entity && entity->metadata()) { | 648 | if (entity && entity->metadata()) { |
661 | auto metadata = GetMetadata(entity->metadata()->Data()); | 649 | auto metadata = GetMetadata(entity->metadata()->Data()); |
@@ -664,6 +652,7 @@ bool EntityStore::exists(const QByteArray &type, const QByteArray &uid) | |||
664 | alreadyRemoved = true; | 652 | alreadyRemoved = true; |
665 | } | 653 | } |
666 | } | 654 | } |
655 | return true; | ||
667 | }, | 656 | }, |
668 | [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to read old revision from storage: " << error.message; }); | 657 | [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to read old revision from storage: " << error.message; }); |
669 | if (!found) { | 658 | if (!found) { |
@@ -677,23 +666,32 @@ bool EntityStore::exists(const QByteArray &type, const QByteArray &uid) | |||
677 | return true; | 666 | return true; |
678 | } | 667 | } |
679 | 668 | ||
680 | void EntityStore::readRevisions(const QByteArray &type, const QByteArray &uid, qint64 startingRevision, const std::function<void(const QByteArray &uid, qint64 revision, const EntityBuffer &entity)> callback) | 669 | void EntityStore::readRevisions(const QByteArray &type, const QByteArray &uid, qint64 startingRevision, |
670 | const std::function<void(const QByteArray &uid, qint64 revision, const EntityBuffer &entity)> callback) | ||
681 | { | 671 | { |
682 | Q_ASSERT(d); | 672 | Q_ASSERT(d); |
683 | Q_ASSERT(!uid.isEmpty()); | 673 | Q_ASSERT(!uid.isEmpty()); |
684 | const auto internalUid = Identifier::fromDisplayByteArray(uid).toInternalByteArray(); | ||
685 | DataStore::mainDatabase(d->transaction, type) | ||
686 | .scan(internalUid, | ||
687 | [&](const QByteArray &key, const QByteArray &value) -> bool { | ||
688 | const auto parsedKey = Key::fromInternalByteArray(key); | ||
689 | const auto revision = parsedKey.revision().toQint64(); | ||
690 | if (revision >= startingRevision) { | ||
691 | callback(parsedKey.identifier().toDisplayByteArray(), revision, Sink::EntityBuffer(value.data(), value.size())); | ||
692 | } | ||
693 | return true; | ||
694 | }, | ||
695 | [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Error while reading: " << error.message; }, true); | ||
696 | 674 | ||
675 | const auto revisions = DataStore::getRevisionsFromUid(d->transaction, uid); | ||
676 | |||
677 | const auto db = DataStore::mainDatabase(d->transaction, type); | ||
678 | |||
679 | for (const auto revision : revisions) { | ||
680 | if (revision < startingRevision) { | ||
681 | continue; | ||
682 | } | ||
683 | |||
684 | db.scan(revision, | ||
685 | [&](size_t rev, const QByteArray &value) { | ||
686 | Q_ASSERT(rev == revision); | ||
687 | callback(uid, revision, Sink::EntityBuffer(value.data(), value.size())); | ||
688 | return false; | ||
689 | }, | ||
690 | [&](const DataStore::Error &error) { | ||
691 | SinkWarningCtx(d->logCtx) << "Error while reading: " << error.message; | ||
692 | }, | ||
693 | true); | ||
694 | } | ||
697 | } | 695 | } |
698 | 696 | ||
699 | qint64 EntityStore::maxRevision() | 697 | qint64 EntityStore::maxRevision() |
diff --git a/common/storage/key.cpp b/common/storage/key.cpp index 2327061..a6567ea 100644 --- a/common/storage/key.cpp +++ b/common/storage/key.cpp | |||
@@ -155,6 +155,11 @@ qint64 Revision::toQint64() const | |||
155 | return rev; | 155 | return rev; |
156 | } | 156 | } |
157 | 157 | ||
158 | size_t Revision::toSizeT() const | ||
159 | { | ||
160 | return rev; | ||
161 | } | ||
162 | |||
158 | bool Revision::isValidInternal(const QByteArray &bytes) | 163 | bool Revision::isValidInternal(const QByteArray &bytes) |
159 | { | 164 | { |
160 | if (bytes.size() != Revision::INTERNAL_REPR_SIZE) { | 165 | if (bytes.size() != Revision::INTERNAL_REPR_SIZE) { |
diff --git a/common/storage/key.h b/common/storage/key.h index baabe38..da90ddd 100644 --- a/common/storage/key.h +++ b/common/storage/key.h | |||
@@ -67,7 +67,7 @@ public: | |||
67 | static const constexpr size_t INTERNAL_REPR_SIZE = 19; | 67 | static const constexpr size_t INTERNAL_REPR_SIZE = 19; |
68 | static const constexpr size_t DISPLAY_REPR_SIZE = 19; | 68 | static const constexpr size_t DISPLAY_REPR_SIZE = 19; |
69 | 69 | ||
70 | Revision(qint64 rev) : rev(rev) {} | 70 | Revision(size_t rev) : rev(rev) {} |
71 | 71 | ||
72 | QByteArray toInternalByteArray() const; | 72 | QByteArray toInternalByteArray() const; |
73 | static Revision fromInternalByteArray(const QByteArray &bytes); | 73 | static Revision fromInternalByteArray(const QByteArray &bytes); |
@@ -75,6 +75,7 @@ public: | |||
75 | QByteArray toDisplayByteArray() const; | 75 | QByteArray toDisplayByteArray() const; |
76 | static Revision fromDisplayByteArray(const QByteArray &bytes); | 76 | static Revision fromDisplayByteArray(const QByteArray &bytes); |
77 | qint64 toQint64() const; | 77 | qint64 toQint64() const; |
78 | size_t toSizeT() const; | ||
78 | 79 | ||
79 | static bool isValidInternal(const QByteArray &); | 80 | static bool isValidInternal(const QByteArray &); |
80 | static bool isValidDisplay(const QByteArray &); | 81 | static bool isValidDisplay(const QByteArray &); |
@@ -84,7 +85,7 @@ public: | |||
84 | bool operator!=(const Revision &other) const; | 85 | bool operator!=(const Revision &other) const; |
85 | 86 | ||
86 | private: | 87 | private: |
87 | qint64 rev; | 88 | size_t rev; |
88 | }; | 89 | }; |
89 | 90 | ||
90 | class Key | 91 | class Key |
diff --git a/common/storage_common.cpp b/common/storage_common.cpp index 264f223..7c794c3 100644 --- a/common/storage_common.cpp +++ b/common/storage_common.cpp | |||
@@ -117,26 +117,69 @@ qint64 DataStore::cleanedUpRevision(const DataStore::Transaction &transaction) | |||
117 | return r; | 117 | return r; |
118 | } | 118 | } |
119 | 119 | ||
120 | QByteArray DataStore::getUidFromRevision(const DataStore::Transaction &transaction, qint64 revision) | 120 | QByteArray DataStore::getUidFromRevision(const DataStore::Transaction &transaction, size_t revision) |
121 | { | 121 | { |
122 | QByteArray uid; | 122 | QByteArray uid; |
123 | transaction.openDatabase("revisions") | 123 | transaction |
124 | .scan(QByteArray::number(revision), | 124 | .openDatabase("revisions", /* errorHandler = */ {}, IntegerKeys) |
125 | [&](const QByteArray &, const QByteArray &value) -> bool { | 125 | .scan(revision, |
126 | uid = QByteArray{value.constData(), value.size()}; | 126 | [&](const size_t, const QByteArray &value) -> bool { |
127 | uid = QByteArray{ value.constData(), value.size() }; | ||
127 | return false; | 128 | return false; |
128 | }, | 129 | }, |
129 | [revision](const Error &error) { SinkWarning() << "Couldn't find uid for revision: " << revision << error.message; }); | 130 | [revision](const Error &error) { |
131 | SinkWarning() << "Couldn't find uid for revision: " << revision << error.message; | ||
132 | }); | ||
130 | Q_ASSERT(!uid.isEmpty()); | 133 | Q_ASSERT(!uid.isEmpty()); |
131 | return uid; | 134 | return uid; |
132 | } | 135 | } |
133 | 136 | ||
134 | QByteArray DataStore::getTypeFromRevision(const DataStore::Transaction &transaction, qint64 revision) | 137 | size_t DataStore::getLatestRevisionFromUid(DataStore::Transaction &t, const QByteArray &uid) |
138 | { | ||
139 | size_t revision; | ||
140 | t.openDatabase("uidsToRevisions", {}, AllowDuplicates | IntegerValues) | ||
141 | .findLatest(uid, [&revision](const QByteArray &key, const QByteArray &value) { | ||
142 | revision = byteArrayToSizeT(value); | ||
143 | }); | ||
144 | |||
145 | return revision; | ||
146 | } | ||
147 | |||
148 | QList<size_t> DataStore::getRevisionsUntilFromUid(DataStore::Transaction &t, const QByteArray &uid, size_t lastRevision) | ||
149 | { | ||
150 | QList<size_t> queriedRevisions; | ||
151 | t.openDatabase("uidsToRevisions", {}, AllowDuplicates | IntegerValues) | ||
152 | .scan(uid, [&queriedRevisions, lastRevision](const QByteArray &, const QByteArray &value) { | ||
153 | size_t currentRevision = byteArrayToSizeT(value); | ||
154 | if (currentRevision < lastRevision) { | ||
155 | queriedRevisions << currentRevision; | ||
156 | return true; | ||
157 | } | ||
158 | |||
159 | return false; | ||
160 | }); | ||
161 | |||
162 | return queriedRevisions; | ||
163 | } | ||
164 | |||
165 | QList<size_t> DataStore::getRevisionsFromUid(DataStore::Transaction &t, const QByteArray &uid) | ||
166 | { | ||
167 | QList<size_t> queriedRevisions; | ||
168 | t.openDatabase("uidsToRevisions", {}, AllowDuplicates | IntegerValues) | ||
169 | .scan(uid, [&queriedRevisions](const QByteArray &, const QByteArray &value) { | ||
170 | queriedRevisions << byteArrayToSizeT(value); | ||
171 | return true; | ||
172 | }); | ||
173 | |||
174 | return queriedRevisions; | ||
175 | } | ||
176 | |||
177 | QByteArray DataStore::getTypeFromRevision(const DataStore::Transaction &transaction, size_t revision) | ||
135 | { | 178 | { |
136 | QByteArray type; | 179 | QByteArray type; |
137 | transaction.openDatabase("revisionType") | 180 | transaction.openDatabase("revisionType", /* errorHandler = */ {}, IntegerKeys) |
138 | .scan(QByteArray::number(revision), | 181 | .scan(revision, |
139 | [&](const QByteArray &, const QByteArray &value) -> bool { | 182 | [&](const size_t, const QByteArray &value) -> bool { |
140 | type = QByteArray{value.constData(), value.size()}; | 183 | type = QByteArray{value.constData(), value.size()}; |
141 | return false; | 184 | return false; |
142 | }, | 185 | }, |
@@ -145,17 +188,31 @@ QByteArray DataStore::getTypeFromRevision(const DataStore::Transaction &transact | |||
145 | return type; | 188 | return type; |
146 | } | 189 | } |
147 | 190 | ||
148 | void DataStore::recordRevision(DataStore::Transaction &transaction, qint64 revision, const QByteArray &uid, const QByteArray &type) | 191 | void DataStore::recordRevision(DataStore::Transaction &transaction, size_t revision, |
192 | const QByteArray &uid, const QByteArray &type) | ||
149 | { | 193 | { |
150 | // TODO use integerkeys | 194 | transaction |
151 | transaction.openDatabase("revisions").write(QByteArray::number(revision), uid); | 195 | .openDatabase("revisions", /* errorHandler = */ {}, IntegerKeys) |
152 | transaction.openDatabase("revisionType").write(QByteArray::number(revision), type); | 196 | .write(revision, uid); |
197 | transaction.openDatabase("uidsToRevisions", /* errorHandler = */ {}, AllowDuplicates | IntegerValues) | ||
198 | .write(uid, sizeTToByteArray(revision)); | ||
199 | transaction | ||
200 | .openDatabase("revisionType", /* errorHandler = */ {}, IntegerKeys) | ||
201 | .write(revision, type); | ||
153 | } | 202 | } |
154 | 203 | ||
155 | void DataStore::removeRevision(DataStore::Transaction &transaction, qint64 revision) | 204 | void DataStore::removeRevision(DataStore::Transaction &transaction, size_t revision) |
156 | { | 205 | { |
157 | transaction.openDatabase("revisions").remove(QByteArray::number(revision)); | 206 | const QByteArray uid = getUidFromRevision(transaction, revision); |
158 | transaction.openDatabase("revisionType").remove(QByteArray::number(revision)); | 207 | |
208 | transaction | ||
209 | .openDatabase("revisions", /* errorHandler = */ {}, IntegerKeys) | ||
210 | .remove(revision); | ||
211 | transaction.openDatabase("uidsToRevisions", /* errorHandler = */ {}, AllowDuplicates | IntegerValues) | ||
212 | .remove(uid, sizeTToByteArray(revision)); | ||
213 | transaction | ||
214 | .openDatabase("revisionType", /* errorHandler = */ {}, IntegerKeys) | ||
215 | .remove(revision); | ||
159 | } | 216 | } |
160 | 217 | ||
161 | void DataStore::recordUid(DataStore::Transaction &transaction, const QByteArray &uid, const QByteArray &type) | 218 | void DataStore::recordUid(DataStore::Transaction &transaction, const QByteArray &uid, const QByteArray &type) |
@@ -176,6 +233,18 @@ void DataStore::getUids(const QByteArray &type, const Transaction &transaction, | |||
176 | }); | 233 | }); |
177 | } | 234 | } |
178 | 235 | ||
236 | bool DataStore::hasUid(const QByteArray &type, const Transaction &transaction, const QByteArray &uid) | ||
237 | { | ||
238 | bool hasTheUid = false; | ||
239 | transaction.openDatabase(type + "uids").scan(uid, [&](const QByteArray &key, const QByteArray &) { | ||
240 | Q_ASSERT(uid == key); | ||
241 | hasTheUid = true; | ||
242 | return false; | ||
243 | }); | ||
244 | |||
245 | return hasTheUid; | ||
246 | } | ||
247 | |||
179 | bool DataStore::isInternalKey(const char *key) | 248 | bool DataStore::isInternalKey(const char *key) |
180 | { | 249 | { |
181 | return key && strncmp(key, s_internalPrefix, s_internalPrefixSize) == 0; | 250 | return key && strncmp(key, s_internalPrefix, s_internalPrefixSize) == 0; |
@@ -207,7 +276,7 @@ DataStore::NamedDatabase DataStore::mainDatabase(const DataStore::Transaction &t | |||
207 | Q_ASSERT(false); | 276 | Q_ASSERT(false); |
208 | return {}; | 277 | return {}; |
209 | } | 278 | } |
210 | return t.openDatabase(type + ".main"); | 279 | return t.openDatabase(type + ".main", /* errorHandler= */ {}, IntegerKeys); |
211 | } | 280 | } |
212 | 281 | ||
213 | bool DataStore::NamedDatabase::contains(const QByteArray &uid) | 282 | bool DataStore::NamedDatabase::contains(const QByteArray &uid) |
diff --git a/common/storage_lmdb.cpp b/common/storage_lmdb.cpp index a007405..0458dae 100644 --- a/common/storage_lmdb.cpp +++ b/common/storage_lmdb.cpp | |||
@@ -48,6 +48,10 @@ static QMutex sCreateDbiLock; | |||
48 | static QHash<QString, MDB_env *> sEnvironments; | 48 | static QHash<QString, MDB_env *> sEnvironments; |
49 | static QHash<QString, MDB_dbi> sDbis; | 49 | static QHash<QString, MDB_dbi> sDbis; |
50 | 50 | ||
51 | int AllowDuplicates = MDB_DUPSORT; | ||
52 | int IntegerKeys = MDB_INTEGERKEY; | ||
53 | int IntegerValues = MDB_INTEGERDUP; | ||
54 | |||
51 | int getErrorCode(int e) | 55 | int getErrorCode(int e) |
52 | { | 56 | { |
53 | switch (e) { | 57 | switch (e) { |
@@ -101,14 +105,8 @@ static QList<QByteArray> getDatabaseNames(MDB_txn *transaction) | |||
101 | * and we always need to commit the transaction ASAP | 105 | * and we always need to commit the transaction ASAP |
102 | * We can only ever enter from one point per process. | 106 | * We can only ever enter from one point per process. |
103 | */ | 107 | */ |
104 | static bool createDbi(MDB_txn *transaction, const QByteArray &db, bool readOnly, bool allowDuplicates, MDB_dbi &dbi) | 108 | static bool createDbi(MDB_txn *transaction, const QByteArray &db, bool readOnly, int flags, MDB_dbi &dbi) |
105 | { | 109 | { |
106 | |||
107 | unsigned int flags = 0; | ||
108 | if (allowDuplicates) { | ||
109 | flags |= MDB_DUPSORT; | ||
110 | } | ||
111 | |||
112 | MDB_dbi flagtableDbi; | 110 | MDB_dbi flagtableDbi; |
113 | if (const int rc = mdb_dbi_open(transaction, "__flagtable", readOnly ? 0 : MDB_CREATE, &flagtableDbi)) { | 111 | if (const int rc = mdb_dbi_open(transaction, "__flagtable", readOnly ? 0 : MDB_CREATE, &flagtableDbi)) { |
114 | if (!readOnly) { | 112 | if (!readOnly) { |
@@ -130,6 +128,10 @@ static bool createDbi(MDB_txn *transaction, const QByteArray &db, bool readOnly, | |||
130 | } | 128 | } |
131 | } | 129 | } |
132 | 130 | ||
131 | if (flags & IntegerValues && !(flags & AllowDuplicates)) { | ||
132 | SinkWarning() << "Opening a database with integer values, but not duplicate keys"; | ||
133 | } | ||
134 | |||
133 | if (const int rc = mdb_dbi_open(transaction, db.constData(), flags, &dbi)) { | 135 | if (const int rc = mdb_dbi_open(transaction, db.constData(), flags, &dbi)) { |
134 | //Create the db if it is not existing already | 136 | //Create the db if it is not existing already |
135 | if (rc == MDB_NOTFOUND && !readOnly) { | 137 | if (rc == MDB_NOTFOUND && !readOnly) { |
@@ -165,7 +167,7 @@ static bool createDbi(MDB_txn *transaction, const QByteArray &db, bool readOnly, | |||
165 | //Store the flags without the create option | 167 | //Store the flags without the create option |
166 | const auto ba = QByteArray::number(flags); | 168 | const auto ba = QByteArray::number(flags); |
167 | value.mv_data = const_cast<void*>(static_cast<const void*>(ba.constData())); | 169 | value.mv_data = const_cast<void*>(static_cast<const void*>(ba.constData())); |
168 | value.mv_size = db.size(); | 170 | value.mv_size = ba.size(); |
169 | if (const int rc = mdb_put(transaction, flagtableDbi, &key, &value, MDB_NOOVERWRITE)) { | 171 | if (const int rc = mdb_put(transaction, flagtableDbi, &key, &value, MDB_NOOVERWRITE)) { |
170 | //We expect this to fail if we're only creating the dbi but not the db | 172 | //We expect this to fail if we're only creating the dbi but not the db |
171 | if (rc != MDB_KEYEXIST) { | 173 | if (rc != MDB_KEYEXIST) { |
@@ -175,7 +177,7 @@ static bool createDbi(MDB_txn *transaction, const QByteArray &db, bool readOnly, | |||
175 | } else { | 177 | } else { |
176 | //It's not an error if we only want to read | 178 | //It's not an error if we only want to read |
177 | if (!readOnly) { | 179 | if (!readOnly) { |
178 | SinkWarning() << "Failed to open db " << QByteArray(mdb_strerror(rc)); | 180 | SinkWarning() << "Failed to open db " << db << "error:" << QByteArray(mdb_strerror(rc)); |
179 | return true; | 181 | return true; |
180 | } | 182 | } |
181 | return false; | 183 | return false; |
@@ -187,8 +189,14 @@ static bool createDbi(MDB_txn *transaction, const QByteArray &db, bool readOnly, | |||
187 | class DataStore::NamedDatabase::Private | 189 | class DataStore::NamedDatabase::Private |
188 | { | 190 | { |
189 | public: | 191 | public: |
190 | Private(const QByteArray &_db, bool _allowDuplicates, const std::function<void(const DataStore::Error &error)> &_defaultErrorHandler, const QString &_name, MDB_txn *_txn) | 192 | Private(const QByteArray &_db, int _flags, |
191 | : db(_db), transaction(_txn), allowDuplicates(_allowDuplicates), defaultErrorHandler(_defaultErrorHandler), name(_name) | 193 | const std::function<void(const DataStore::Error &error)> &_defaultErrorHandler, |
194 | const QString &_name, MDB_txn *_txn) | ||
195 | : db(_db), | ||
196 | transaction(_txn), | ||
197 | flags(_flags), | ||
198 | defaultErrorHandler(_defaultErrorHandler), | ||
199 | name(_name) | ||
192 | { | 200 | { |
193 | } | 201 | } |
194 | 202 | ||
@@ -199,7 +207,7 @@ public: | |||
199 | QByteArray db; | 207 | QByteArray db; |
200 | MDB_txn *transaction; | 208 | MDB_txn *transaction; |
201 | MDB_dbi dbi; | 209 | MDB_dbi dbi; |
202 | bool allowDuplicates; | 210 | int flags; |
203 | std::function<void(const DataStore::Error &error)> defaultErrorHandler; | 211 | std::function<void(const DataStore::Error &error)> defaultErrorHandler; |
204 | QString name; | 212 | QString name; |
205 | bool createdNewDbi = false; | 213 | bool createdNewDbi = false; |
@@ -313,7 +321,7 @@ public: | |||
313 | } else { | 321 | } else { |
314 | dbiTransaction = transaction; | 322 | dbiTransaction = transaction; |
315 | } | 323 | } |
316 | if (createDbi(dbiTransaction, db, readOnly, allowDuplicates, dbi)) { | 324 | if (createDbi(dbiTransaction, db, readOnly, flags, dbi)) { |
317 | if (readOnly) { | 325 | if (readOnly) { |
318 | mdb_txn_commit(dbiTransaction); | 326 | mdb_txn_commit(dbiTransaction); |
319 | Q_ASSERT(!sDbis.contains(dbiName)); | 327 | Q_ASSERT(!sDbis.contains(dbiName)); |
@@ -371,6 +379,12 @@ DataStore::NamedDatabase::~NamedDatabase() | |||
371 | delete d; | 379 | delete d; |
372 | } | 380 | } |
373 | 381 | ||
382 | bool DataStore::NamedDatabase::write(const size_t key, const QByteArray &value, | ||
383 | const std::function<void(const DataStore::Error &error)> &errorHandler) | ||
384 | { | ||
385 | return write(sizeTToByteArray(key), value, errorHandler); | ||
386 | } | ||
387 | |||
374 | bool DataStore::NamedDatabase::write(const QByteArray &sKey, const QByteArray &sValue, const std::function<void(const DataStore::Error &error)> &errorHandler) | 388 | bool DataStore::NamedDatabase::write(const QByteArray &sKey, const QByteArray &sValue, const std::function<void(const DataStore::Error &error)> &errorHandler) |
375 | { | 389 | { |
376 | if (!d || !d->transaction) { | 390 | if (!d || !d->transaction) { |
@@ -407,11 +421,23 @@ bool DataStore::NamedDatabase::write(const QByteArray &sKey, const QByteArray &s | |||
407 | return !rc; | 421 | return !rc; |
408 | } | 422 | } |
409 | 423 | ||
424 | void DataStore::NamedDatabase::remove( | ||
425 | const size_t key, const std::function<void(const DataStore::Error &error)> &errorHandler) | ||
426 | { | ||
427 | return remove(sizeTToByteArray(key), errorHandler); | ||
428 | } | ||
429 | |||
410 | void DataStore::NamedDatabase::remove(const QByteArray &k, const std::function<void(const DataStore::Error &error)> &errorHandler) | 430 | void DataStore::NamedDatabase::remove(const QByteArray &k, const std::function<void(const DataStore::Error &error)> &errorHandler) |
411 | { | 431 | { |
412 | remove(k, QByteArray(), errorHandler); | 432 | remove(k, QByteArray(), errorHandler); |
413 | } | 433 | } |
414 | 434 | ||
435 | void DataStore::NamedDatabase::remove(const size_t key, const QByteArray &value, | ||
436 | const std::function<void(const DataStore::Error &error)> &errorHandler) | ||
437 | { | ||
438 | return remove(sizeTToByteArray(key), value, errorHandler); | ||
439 | } | ||
440 | |||
415 | void DataStore::NamedDatabase::remove(const QByteArray &k, const QByteArray &value, const std::function<void(const DataStore::Error &error)> &errorHandler) | 441 | void DataStore::NamedDatabase::remove(const QByteArray &k, const QByteArray &value, const std::function<void(const DataStore::Error &error)> &errorHandler) |
416 | { | 442 | { |
417 | if (!d || !d->transaction) { | 443 | if (!d || !d->transaction) { |
@@ -445,6 +471,17 @@ void DataStore::NamedDatabase::remove(const QByteArray &k, const QByteArray &val | |||
445 | } | 471 | } |
446 | } | 472 | } |
447 | 473 | ||
474 | int DataStore::NamedDatabase::scan(const size_t key, | ||
475 | const std::function<bool(size_t key, const QByteArray &value)> &resultHandler, | ||
476 | const std::function<void(const DataStore::Error &error)> &errorHandler, bool skipInternalKeys) const | ||
477 | { | ||
478 | return scan(sizeTToByteArray(key), | ||
479 | [&resultHandler](const QByteArray &key, const QByteArray &value) { | ||
480 | return resultHandler(byteArrayToSizeT(key), value); | ||
481 | }, | ||
482 | errorHandler, /* findSubstringKeys = */ false, skipInternalKeys); | ||
483 | } | ||
484 | |||
448 | int DataStore::NamedDatabase::scan(const QByteArray &k, const std::function<bool(const QByteArray &key, const QByteArray &value)> &resultHandler, | 485 | int DataStore::NamedDatabase::scan(const QByteArray &k, const std::function<bool(const QByteArray &key, const QByteArray &value)> &resultHandler, |
449 | const std::function<void(const DataStore::Error &error)> &errorHandler, bool findSubstringKeys, bool skipInternalKeys) const | 486 | const std::function<void(const DataStore::Error &error)> &errorHandler, bool findSubstringKeys, bool skipInternalKeys) const |
450 | { | 487 | { |
@@ -471,8 +508,10 @@ int DataStore::NamedDatabase::scan(const QByteArray &k, const std::function<bool | |||
471 | 508 | ||
472 | int numberOfRetrievedValues = 0; | 509 | int numberOfRetrievedValues = 0; |
473 | 510 | ||
474 | if (k.isEmpty() || d->allowDuplicates || findSubstringKeys) { | 511 | bool allowDuplicates = d->flags & AllowDuplicates; |
475 | MDB_cursor_op op = d->allowDuplicates ? MDB_SET : MDB_FIRST; | 512 | |
513 | if (k.isEmpty() || allowDuplicates || findSubstringKeys) { | ||
514 | MDB_cursor_op op = allowDuplicates ? MDB_SET : MDB_FIRST; | ||
476 | if (findSubstringKeys) { | 515 | if (findSubstringKeys) { |
477 | op = MDB_SET_RANGE; | 516 | op = MDB_SET_RANGE; |
478 | } | 517 | } |
@@ -490,7 +529,7 @@ int DataStore::NamedDatabase::scan(const QByteArray &k, const std::function<bool | |||
490 | key.mv_data = (void *)k.constData(); | 529 | key.mv_data = (void *)k.constData(); |
491 | key.mv_size = k.size(); | 530 | key.mv_size = k.size(); |
492 | } | 531 | } |
493 | MDB_cursor_op nextOp = (d->allowDuplicates && !findSubstringKeys) ? MDB_NEXT_DUP : MDB_NEXT; | 532 | MDB_cursor_op nextOp = (allowDuplicates && !findSubstringKeys) ? MDB_NEXT_DUP : MDB_NEXT; |
494 | while ((rc = mdb_cursor_get(cursor, &key, &data, nextOp)) == 0) { | 533 | while ((rc = mdb_cursor_get(cursor, &key, &data, nextOp)) == 0) { |
495 | const auto current = QByteArray::fromRawData((char *)key.mv_data, key.mv_size); | 534 | const auto current = QByteArray::fromRawData((char *)key.mv_data, key.mv_size); |
496 | // Every consequitive lookup simply iterates through the list | 535 | // Every consequitive lookup simply iterates through the list |
@@ -529,6 +568,18 @@ int DataStore::NamedDatabase::scan(const QByteArray &k, const std::function<bool | |||
529 | return numberOfRetrievedValues; | 568 | return numberOfRetrievedValues; |
530 | } | 569 | } |
531 | 570 | ||
571 | |||
572 | void DataStore::NamedDatabase::findLatest(size_t key, | ||
573 | const std::function<void(size_t key, const QByteArray &value)> &resultHandler, | ||
574 | const std::function<void(const DataStore::Error &error)> &errorHandler) const | ||
575 | { | ||
576 | return findLatest(sizeTToByteArray(key), | ||
577 | [&resultHandler](const QByteArray &key, const QByteArray &value) { | ||
578 | resultHandler(byteArrayToSizeT(value), value); | ||
579 | }, | ||
580 | errorHandler); | ||
581 | } | ||
582 | |||
532 | void DataStore::NamedDatabase::findLatest(const QByteArray &k, const std::function<void(const QByteArray &key, const QByteArray &value)> &resultHandler, | 583 | void DataStore::NamedDatabase::findLatest(const QByteArray &k, const std::function<void(const QByteArray &key, const QByteArray &value)> &resultHandler, |
533 | const std::function<void(const DataStore::Error &error)> &errorHandler) const | 584 | const std::function<void(const DataStore::Error &error)> &errorHandler) const |
534 | { | 585 | { |
@@ -602,6 +653,17 @@ void DataStore::NamedDatabase::findLatest(const QByteArray &k, const std::functi | |||
602 | return; | 653 | return; |
603 | } | 654 | } |
604 | 655 | ||
656 | int DataStore::NamedDatabase::findAllInRange(const size_t lowerBound, const size_t upperBound, | ||
657 | const std::function<void(size_t key, const QByteArray &value)> &resultHandler, | ||
658 | const std::function<void(const DataStore::Error &error)> &errorHandler) const | ||
659 | { | ||
660 | return findAllInRange(sizeTToByteArray(lowerBound), sizeTToByteArray(upperBound), | ||
661 | [&resultHandler](const QByteArray &key, const QByteArray &value) { | ||
662 | resultHandler(byteArrayToSizeT(value), value); | ||
663 | }, | ||
664 | errorHandler); | ||
665 | } | ||
666 | |||
605 | int DataStore::NamedDatabase::findAllInRange(const QByteArray &lowerBound, const QByteArray &upperBound, | 667 | int DataStore::NamedDatabase::findAllInRange(const QByteArray &lowerBound, const QByteArray &upperBound, |
606 | const std::function<void(const QByteArray &key, const QByteArray &value)> &resultHandler, | 668 | const std::function<void(const QByteArray &key, const QByteArray &value)> &resultHandler, |
607 | const std::function<void(const DataStore::Error &error)> &errorHandler) const | 669 | const std::function<void(const DataStore::Error &error)> &errorHandler) const |
@@ -839,30 +901,8 @@ void DataStore::Transaction::abort() | |||
839 | d->transaction = nullptr; | 901 | d->transaction = nullptr; |
840 | } | 902 | } |
841 | 903 | ||
842 | //Ensure that we opened the correct database by comparing the expected identifier with the one | 904 | DataStore::NamedDatabase DataStore::Transaction::openDatabase(const QByteArray &db, |
843 | //we write to the database on first open. | 905 | const std::function<void(const DataStore::Error &error)> &errorHandler, int flags) const |
844 | static bool ensureCorrectDb(DataStore::NamedDatabase &database, const QByteArray &db, bool readOnly) | ||
845 | { | ||
846 | bool openedTheWrongDatabase = false; | ||
847 | auto count = database.scan("__internal_dbname", [db, &openedTheWrongDatabase](const QByteArray &key, const QByteArray &value) ->bool { | ||
848 | if (value != db) { | ||
849 | SinkWarning() << "Opened the wrong database, got " << value << " instead of " << db; | ||
850 | openedTheWrongDatabase = true; | ||
851 | } | ||
852 | return false; | ||
853 | }, | ||
854 | [&](const DataStore::Error &) { | ||
855 | }, false); | ||
856 | //This is the first time we open this database in a write transaction, write the db name | ||
857 | if (!count) { | ||
858 | if (!readOnly) { | ||
859 | database.write("__internal_dbname", db); | ||
860 | } | ||
861 | } | ||
862 | return !openedTheWrongDatabase; | ||
863 | } | ||
864 | |||
865 | DataStore::NamedDatabase DataStore::Transaction::openDatabase(const QByteArray &db, const std::function<void(const DataStore::Error &error)> &errorHandler, bool allowDuplicates) const | ||
866 | { | 906 | { |
867 | if (!d) { | 907 | if (!d) { |
868 | SinkError() << "Tried to open database on invalid transaction: " << db; | 908 | SinkError() << "Tried to open database on invalid transaction: " << db; |
@@ -871,7 +911,8 @@ DataStore::NamedDatabase DataStore::Transaction::openDatabase(const QByteArray & | |||
871 | Q_ASSERT(d->transaction); | 911 | Q_ASSERT(d->transaction); |
872 | // We don't now if anything changed | 912 | // We don't now if anything changed |
873 | d->implicitCommit = true; | 913 | d->implicitCommit = true; |
874 | auto p = new DataStore::NamedDatabase::Private(db, allowDuplicates, d->defaultErrorHandler, d->name, d->transaction); | 914 | auto p = new DataStore::NamedDatabase::Private( |
915 | db, flags, d->defaultErrorHandler, d->name, d->transaction); | ||
875 | auto ret = p->openDatabase(d->requestedRead, errorHandler); | 916 | auto ret = p->openDatabase(d->requestedRead, errorHandler); |
876 | if (!ret) { | 917 | if (!ret) { |
877 | delete p; | 918 | delete p; |
@@ -883,11 +924,6 @@ DataStore::NamedDatabase DataStore::Transaction::openDatabase(const QByteArray & | |||
883 | } | 924 | } |
884 | 925 | ||
885 | auto database = DataStore::NamedDatabase(p); | 926 | auto database = DataStore::NamedDatabase(p); |
886 | if (!ensureCorrectDb(database, db, d->requestedRead)) { | ||
887 | SinkWarning() << "Failed to open the database correctly" << db; | ||
888 | Q_ASSERT(false); | ||
889 | return DataStore::NamedDatabase(); | ||
890 | } | ||
891 | return database; | 927 | return database; |
892 | } | 928 | } |
893 | 929 | ||
@@ -1049,11 +1085,11 @@ public: | |||
1049 | 1085 | ||
1050 | //Create dbis from the given layout. | 1086 | //Create dbis from the given layout. |
1051 | for (auto it = layout.tables.constBegin(); it != layout.tables.constEnd(); it++) { | 1087 | for (auto it = layout.tables.constBegin(); it != layout.tables.constEnd(); it++) { |
1052 | const bool allowDuplicates = it.value(); | 1088 | const int flags = it.value(); |
1053 | MDB_dbi dbi = 0; | 1089 | MDB_dbi dbi = 0; |
1054 | const auto db = it.key(); | 1090 | const auto db = it.key(); |
1055 | const auto dbiName = name + db; | 1091 | const auto dbiName = name + db; |
1056 | if (createDbi(transaction, db, readOnly, allowDuplicates, dbi)) { | 1092 | if (createDbi(transaction, db, readOnly, flags, dbi)) { |
1057 | sDbis.insert(dbiName, dbi); | 1093 | sDbis.insert(dbiName, dbi); |
1058 | } | 1094 | } |
1059 | } | 1095 | } |
@@ -1063,8 +1099,8 @@ public: | |||
1063 | MDB_dbi dbi = 0; | 1099 | MDB_dbi dbi = 0; |
1064 | const auto dbiName = name + db; | 1100 | const auto dbiName = name + db; |
1065 | //We're going to load the flags anyways. | 1101 | //We're going to load the flags anyways. |
1066 | bool allowDuplicates = false; | 1102 | const int flags = 0; |
1067 | if (createDbi(transaction, db, readOnly, allowDuplicates, dbi)) { | 1103 | if (createDbi(transaction, db, readOnly, flags, dbi)) { |
1068 | sDbis.insert(dbiName, dbi); | 1104 | sDbis.insert(dbiName, dbi); |
1069 | } | 1105 | } |
1070 | } | 1106 | } |
diff --git a/common/utils.cpp b/common/utils.cpp index 3c54db4..f6c6798 100644 --- a/common/utils.cpp +++ b/common/utils.cpp | |||
@@ -23,3 +23,13 @@ QByteArray Sink::createUuid() | |||
23 | { | 23 | { |
24 | return QUuid::createUuid().toByteArray(); | 24 | return QUuid::createUuid().toByteArray(); |
25 | } | 25 | } |
26 | |||
27 | const QByteArray Sink::sizeTToByteArray(const size_t &value) | ||
28 | { | ||
29 | return QByteArray::fromRawData(reinterpret_cast<const char *>(&value), sizeof(size_t)); | ||
30 | } | ||
31 | |||
32 | size_t Sink::byteArrayToSizeT(const QByteArray &value) | ||
33 | { | ||
34 | return *reinterpret_cast<const size_t *>(value.constData()); | ||
35 | } | ||
diff --git a/common/utils.h b/common/utils.h index 7066d79..8565f17 100644 --- a/common/utils.h +++ b/common/utils.h | |||
@@ -26,6 +26,20 @@ namespace Sink { | |||
26 | 26 | ||
27 | QByteArray createUuid(); | 27 | QByteArray createUuid(); |
28 | 28 | ||
29 | // No copy is done on this functions. Therefore, the caller must not use the | ||
30 | // returned QByteArray after the size_t has been destroyed. | ||
31 | const QByteArray sizeTToByteArray(const size_t &); | ||
32 | size_t byteArrayToSizeT(const QByteArray &); | ||
33 | |||
34 | template <typename T> | ||
35 | static QByteArray padNumber(T number); | ||
36 | |||
37 | template <> | ||
38 | QByteArray padNumber<size_t>(size_t number) | ||
39 | { | ||
40 | return padNumber<qint64>(number); | ||
41 | } | ||
42 | |||
29 | template <typename T> | 43 | template <typename T> |
30 | static QByteArray padNumber(T number) | 44 | static QByteArray padNumber(T number) |
31 | { | 45 | { |