Differential D14099 Diff 38608 common/datastorequery.cpp

Changeset View

Standalone View

View Options

common/datastorequery.cpp

Show All 35 Lines28{
36 } 36​ }
37 return ""; 37​ return "";
38} 38​}
39 39​
40class Source : public FilterBase { 40​class Source : public FilterBase {
41 public: 41​ public:
42 typedef QSharedPointer<Source> Ptr; 42​ typedef QSharedPointer<Source> Ptr;
43 43​
44 QVector<QByteArray> mIds; 44​ QVector<Identifier> mIds;
45 QVector<QByteArray>::ConstIterator mIt; 45​ QVector<Identifier>::ConstIterator mIt;
46 QVector<QByteArray> mIncrementalIds; 46​ QVector<Identifier> mIncrementalIds;
47 QVector<QByteArray>::ConstIterator mIncrementalIt; 47​ QVector<Identifier>::ConstIterator mIncrementalIt;
48 48​
49 Source (const QVector<QByteArray> &ids, DataStoreQuery *store) 49​ Source (const QVector<QByteArray> &ids, DataStoreQuery *store)
50 : FilterBase(store), 50​ : FilterBase(store),
51 mIds(ids), 51​ mIds()
52 mIt(mIds.constBegin()) ​
53 { 52​ {
54 53​ mIds.reserve(ids.size());
cmollekopfUnsubmitted
Done

Should we perhaps replace default construction with a static createIdentifier() method?

cmollekopf: Should we perhaps replace default construction with a static createIdentifier() method?
rnicoleAuthorUnsubmitted
Done

Replacing the behavior of the default construction is definitely a consideration. I initially wanted to avoid "null" Keys in the database: in the case something goes wrong and the default constructor is called without our knowledge, we will have new unique keys for each elements, which is arguably better than multiple elements with the same "null" key.

rnicole: Replacing the behavior of the default construction is definitely a consideration. I initially…
cmollekopfUnsubmitted
Done

Empty keys are much easier to catch though. We can easily build the safety net to make sure no empty keys ever end up in the database (I think lmdb doesn't even support that, so it will fail anyways). We can't distinguish between a legitimate uid and one that is accidentally created, so worst case that would even hide the root cause from us.

cmollekopf: Empty keys are much easier to catch though. We can easily build the safety net to make sure no…
54​ for (const auto &id : ids) {
55​ mIds.append(Identifier::fromDisplayByteArray(id));
56​ }
57​ mIt = mIds.constBegin();
55 } 58​ }
56 59​
57 virtual ~Source(){} 60​ virtual ~Source(){}
58 61​
59 virtual void skip() Q_DECL_OVERRIDE 62​ virtual void skip() Q_DECL_OVERRIDE
60 { 63​ {
61 if (mIt != mIds.constEnd()) { 64​ if (mIt != mIds.constEnd()) {
62 mIt++; 65​ mIt++;
63 } 66​ }
64 }; 67​ };
65 68​
66 void add(const QVector<QByteArray> &ids) 69​ void add(const QVector<Key> &keys)
67 { 70​ {
68 mIncrementalIds = ids; 71​ mIncrementalIds.clear();
72​ mIncrementalIds.reserve(keys.size());
73​ for (const auto &key : keys) {
74​ mIncrementalIds.append(key.identifier());
75​ }
69 mIncrementalIt = mIncrementalIds.constBegin(); 76​ mIncrementalIt = mIncrementalIds.constBegin();
70 } 77​ }
71 78​
72 bool next(const std::function<void(const ResultSet::Result &result)> &callback) Q_DECL_OVERRIDE 79​ bool next(const std::function<void(const ResultSet::Result &result)> &callback) Q_DECL_OVERRIDE
73 { 80​ {
74 if (!mIncrementalIds.isEmpty()) { 81​ if (!mIncrementalIds.isEmpty()) {
75 if (mIncrementalIt == mIncrementalIds.constEnd()) { 82​ if (mIncrementalIt == mIncrementalIds.constEnd()) {
76 return false; 83​ return false;
▲ Show 20 Lines • Show All 140 Lines • ▼ Show 20 Line(s)223 {
217 return mResult; 224​ return mResult;
218 } 225​ }
219 private: 226​ private:
220 QVariant mResult; 227​ QVariant mResult;
221 }; 228​ };
222 229​
223 QSet<QByteArray> mReducedValues; 230​ QSet<QByteArray> mReducedValues;
224 QSet<QByteArray> mIncrementallyReducedValues; 231​ QSet<QByteArray> mIncrementallyReducedValues;
225 QHash<QByteArray, QByteArray> mSelectedValues; 232​ QHash<QByteArray, Identifier> mSelectedValues;
226 QByteArray mReductionProperty; 233​ QByteArray mReductionProperty;
227 QByteArray mSelectionProperty; 234​ QByteArray mSelectionProperty;
228 QueryBase::Reduce::Selector::Comparator mSelectionComparator; 235​ QueryBase::Reduce::Selector::Comparator mSelectionComparator;
229 QList<Aggregator> mAggregators; 236​ QList<Aggregator> mAggregators;
230 237​
231 Reduce(const QByteArray &reductionProperty, const QByteArray &selectionProperty, QueryBase::Reduce::Selector::Comparator comparator, FilterBase::Ptr source, DataStoreQuery *store) 238​ Reduce(const QByteArray &reductionProperty, const QByteArray &selectionProperty, QueryBase::Reduce::Selector::Comparator comparator, FilterBase::Ptr source, DataStoreQuery *store)
232 : Filter(source, store), 239​ : Filter(source, store),
233 mReductionProperty(reductionProperty), 240​ mReductionProperty(reductionProperty),
Show All 24 Lines
258 static bool compare(const QVariant &left, const QVariant &right, QueryBase::Reduce::Selector::Comparator comparator) { 265​ static bool compare(const QVariant &left, const QVariant &right, QueryBase::Reduce::Selector::Comparator comparator) {
259 if (comparator == QueryBase::Reduce::Selector::Max) { 266​ if (comparator == QueryBase::Reduce::Selector::Max) {
260 return left > right; 267​ return left > right;
261 } 268​ }
262 return false; 269​ return false;
263 } 270​ }
264 271​
265 struct ReductionResult { 272​ struct ReductionResult {
266 QByteArray selection; 273​ Identifier selection;
267 QVector<QByteArray> aggregateIds; 274​ QVector<QByteArray> aggregateIds;
268 QMap<QByteArray, QVariant> aggregateValues; 275​ QMap<QByteArray, QVariant> aggregateValues;
269 }; 276​ };
270 277​
271 ReductionResult reduceOnValue(const QVariant &reductionValue) 278​ ReductionResult reduceOnValue(const QVariant &reductionValue)
272 { 279​ {
273 QMap<QByteArray, QVariant> aggregateValues; 280​ QMap<QByteArray, QVariant> aggregateValues;
274 QVariant selectionResultValue; 281​ QVariant selectionResultValue;
275 QByteArray selectionResult; 282​ Identifier selectionResult;
276 const auto results = indexLookup(mReductionProperty, reductionValue); 283​ const auto results = indexLookup(mReductionProperty, reductionValue);
277 for (auto &aggregator : mAggregators) { 284​ for (auto &aggregator : mAggregators) {
278 aggregator.reset(); 285​ aggregator.reset();
279 } 286​ }
280 QVector<QByteArray> reducedAndFilteredResults; 287​ QVector<QByteArray> reducedAndFilteredResults;
281 for (const auto &r : results) { 288​ for (const auto &r : results) {
282 readEntity(r, [&, this](const Sink::ApplicationDomain::ApplicationDomainType &entity, Sink::Operation operation) { 289​ readEntity(r, [&, this](const Sink::ApplicationDomain::ApplicationDomainType &entity, Sink::Operation operation) {
283 //We need to apply all property filters that we have until the reduction, because the index lookup was unfiltered. 290​ //We need to apply all property filters that we have until the reduction, because the index lookup was unfiltered.
284 if (!matchesFilter(entity)) { 291​ if (!matchesFilter(entity)) {
285 return; 292​ return;
286 } 293​ }
287 reducedAndFilteredResults << r; 294​ reducedAndFilteredResults << r.toDisplayByteArray();
288 Q_ASSERT(operation != Sink::Operation_Removal); 295​ Q_ASSERT(operation != Sink::Operation_Removal);
289 for (auto &aggregator : mAggregators) { 296​ for (auto &aggregator : mAggregators) {
290 if (!aggregator.property.isEmpty()) { 297​ if (!aggregator.property.isEmpty()) {
291 aggregator.process(entity.getProperty(aggregator.property)); 298​ aggregator.process(entity.getProperty(aggregator.property));
292 } else { 299​ } else {
cmollekopfUnsubmitted
Not Done

Most places here should change as well to avoid unnecessarily converting back and forth.
This is not to say that you have to add this to the same patch, it's fine if you have it planned for another patch.

cmollekopf: Most places here should change as well to avoid unnecessarily converting back and forth. This…
rnicoleAuthorUnsubmitted
Not Done

This conversion is here since it will be forwarded to the callback in next, which takes a ResultSet::Result as argument. But ResultSet is used in many places, and here seemed like the most straightforward place to convert it. Should we plan on doing a patch for ResultSet?

rnicole: This conversion is here since it will be forwarded to the `callback` in `next`, which takes a…
cmollekopfUnsubmitted
Not Done

Yes. I think everything internal should be using the binary representation, and ResultSet is the primary means to "stream" results through the query that then get filtered.

cmollekopf: Yes. I think everything internal should be using the binary representation, and ResultSet is…
293 aggregator.process(QVariant{}); 300​ aggregator.process(QVariant{});
294 } 301​ }
295 } 302​ }
296 auto selectionValue = entity.getProperty(mSelectionProperty); 303​ auto selectionValue = entity.getProperty(mSelectionProperty);
297 if (!selectionResultValue.isValid() || compare(selectionValue, selectionResultValue, mSelectionComparator)) { 304​ if (!selectionResultValue.isValid() || compare(selectionValue, selectionResultValue, mSelectionComparator)) {
298 selectionResultValue = selectionValue; 305​ selectionResultValue = selectionValue;
299 selectionResult = entity.identifier(); 306​ selectionResult = Identifier::fromDisplayByteArray(entity.identifier());
300 } 307​ }
301 }); 308​ });
302 } 309​ }
303 310​
304 for (auto &aggregator : mAggregators) { 311​ for (auto &aggregator : mAggregators) {
305 aggregateValues.insert(aggregator.resultProperty, aggregator.result()); 312​ aggregateValues.insert(aggregator.resultProperty, aggregator.result());
306 } 313​ }
307 return {selectionResult, reducedAndFilteredResults, aggregateValues}; 314​ return {selectionResult, reducedAndFilteredResults, aggregateValues};
308 } 315​ }
309 316​
310 bool next(const std::function<void(const ResultSet::Result &)> &callback) Q_DECL_OVERRIDE { 317​ bool next(const std::function<void(const ResultSet::Result &)> &callback) Q_DECL_OVERRIDE {
311 bool foundValue = false; 318​ bool foundValue = false;
312 while(!foundValue && mSource->next([this, callback, &foundValue](const ResultSet::Result &result) { 319​ while(!foundValue && mSource->next([this, callback, &foundValue](const ResultSet::Result &result) {
313 const auto reductionValue = [&] { 320​ const auto reductionValue = [&] {
314 const auto v = result.entity.getProperty(mReductionProperty); 321​ const auto v = result.entity.getProperty(mReductionProperty);
315 //Because we also get Operation_Removal for filtered entities. We use the fact that actually removed entites 322​ //Because we also get Operation_Removal for filtered entities. We use the fact that actually removed entites
316 //won't have the property to reduce on. 323​ //won't have the property to reduce on.
317 //TODO: Perhaps find a cleaner solutoin than abusing Operation::Removed for filtered properties. 324​ //TODO: Perhaps find a cleaner solutoin than abusing Operation::Removed for filtered properties.
318 if (v.isNull() && result.operation == Sink::Operation_Removal) { 325​ if (v.isNull() && result.operation == Sink::Operation_Removal) {
319 //For removals we have to read the last revision to get a value, and thus be able to find the correct thread. 326​ //For removals we have to read the last revision to get a value, and thus be able to find the correct thread.
320 QVariant reductionValue; 327​ QVariant reductionValue;
321 readPrevious(result.entity.identifier(), [&] (const ApplicationDomain::ApplicationDomainType &prev) { 328​ const auto id = Identifier::fromDisplayByteArray(result.entity.identifier());
329​ readPrevious(id, [&] (const ApplicationDomain::ApplicationDomainType &prev) {
322 Q_ASSERT(result.entity.identifier() == prev.identifier()); 330​ Q_ASSERT(result.entity.identifier() == prev.identifier());
323 reductionValue = prev.getProperty(mReductionProperty); 331​ reductionValue = prev.getProperty(mReductionProperty);
324 }); 332​ });
325 return reductionValue; 333​ return reductionValue;
326 } else { 334​ } else {
327 return v; 335​ return v;
328 } 336​ }
329 }(); 337​ }();
330 if (reductionValue.isNull()) { 338​ if (reductionValue.isNull()) {
331 SinkTraceCtx(mDatastore->mLogCtx) << "No reduction value: " << result.entity.identifier(); 339​ SinkTraceCtx(mDatastore->mLogCtx) << "No reduction value: " << result.entity.identifier();
332 //We failed to find a value to reduce on, so ignore this entity. 340​ //We failed to find a value to reduce on, so ignore this entity.
333 //Can happen if the entity was already removed and we have no previous revision. 341​ //Can happen if the entity was already removed and we have no previous revision.
334 return; 342​ return;
335 } 343​ }
336 const auto reductionValueBa = getByteArray(reductionValue); 344​ const auto reductionValueBa = getByteArray(reductionValue);
337 if (!mReducedValues.contains(reductionValueBa)) { 345​ if (!mReducedValues.contains(reductionValueBa)) {
338 //Only reduce every value once. 346​ //Only reduce every value once.
339 mReducedValues.insert(reductionValueBa); 347​ mReducedValues.insert(reductionValueBa);
340 SinkTraceCtx(mDatastore->mLogCtx) << "Reducing new value: " << result.entity.identifier() << reductionValueBa; 348​ SinkTraceCtx(mDatastore->mLogCtx) << "Reducing new value: " << result.entity.identifier() << reductionValueBa;
341 auto reductionResult = reduceOnValue(reductionValue); 349​ auto reductionResult = reduceOnValue(reductionValue);
342 350​
343 //This can happen if we get a removal message from a filtered entity and all entites of the reduction are filtered. 351​ //This can happen if we get a removal message from a filtered entity and all entites of the reduction are filtered.
344 if (reductionResult.selection.isEmpty()) { 352​ if (reductionResult.selection.isNull()) {
345 return; 353​ return;
346 } 354​ }
347 mSelectedValues.insert(reductionValueBa, reductionResult.selection); 355​ mSelectedValues.insert(reductionValueBa, reductionResult.selection);
348 readEntity(reductionResult.selection, [&](const Sink::ApplicationDomain::ApplicationDomainType &entity, Sink::Operation operation) { 356​ readEntity(reductionResult.selection, [&](const Sink::ApplicationDomain::ApplicationDomainType &entity, Sink::Operation operation) {
349 callback({entity, operation, reductionResult.aggregateValues, reductionResult.aggregateIds}); 357​ callback({entity, operation, reductionResult.aggregateValues, reductionResult.aggregateIds});
350 foundValue = true; 358​ foundValue = true;
351 }); 359​ });
352 } else { 360​ } else {
353 //During initial query, do nothing. The lookup above will take care of it. 361​ //During initial query, do nothing. The lookup above will take care of it.
354 //During updates adjust the reduction according to the modification/addition or removal 362​ //During updates adjust the reduction according to the modification/addition or removal
355 //We have to redo the reduction for every element, because of the aggregation values. 363​ //We have to redo the reduction for every element, because of the aggregation values.
356 if (mIncremental && !mIncrementallyReducedValues.contains(reductionValueBa)) { 364​ if (mIncremental && !mIncrementallyReducedValues.contains(reductionValueBa)) {
357 SinkTraceCtx(mDatastore->mLogCtx) << "Incremental reduction update: " << result.entity.identifier() << reductionValueBa; 365​ SinkTraceCtx(mDatastore->mLogCtx) << "Incremental reduction update: " << result.entity.identifier() << reductionValueBa;
358 mIncrementallyReducedValues.insert(reductionValueBa); 366​ mIncrementallyReducedValues.insert(reductionValueBa);
359 //Redo the reduction to find new aggregated values 367​ //Redo the reduction to find new aggregated values
360 auto selectionResult = reduceOnValue(reductionValue); 368​ auto selectionResult = reduceOnValue(reductionValue);
361 369​
362 //If mSelectedValues did not contain the value, oldSelectionResult will be empty.(Happens if entites have been filtered) 370​ //If mSelectedValues did not contain the value, oldSelectionResult will be empty.(Happens if entites have been filtered)
363 auto oldSelectionResult = mSelectedValues.take(reductionValueBa); 371​ auto oldSelectionResult = mSelectedValues.take(reductionValueBa);
364 SinkTraceCtx(mDatastore->mLogCtx) << "Old selection result: " << oldSelectionResult << " New selection result: " << selectionResult.selection; 372​ SinkTraceCtx(mDatastore->mLogCtx) << "Old selection result: " << oldSelectionResult << " New selection result: " << selectionResult.selection;
365 if (selectionResult.selection.isEmpty() && oldSelectionResult.isEmpty()) { 373​ if (selectionResult.selection.isNull() && oldSelectionResult.isNull()) {
366 //Nothing to do, the item was filtered before, and still is. 374​ //Nothing to do, the item was filtered before, and still is.
367 } else if (oldSelectionResult == selectionResult.selection) { 375​ } else if (oldSelectionResult == selectionResult.selection) {
368 mSelectedValues.insert(reductionValueBa, selectionResult.selection); 376​ mSelectedValues.insert(reductionValueBa, selectionResult.selection);
369 Q_ASSERT(!selectionResult.selection.isEmpty()); 377​ Q_ASSERT(!selectionResult.selection.isNull());
370 readEntity(selectionResult.selection, [&](const Sink::ApplicationDomain::ApplicationDomainType &entity, Sink::Operation) { 378​ readEntity(selectionResult.selection, [&](const Sink::ApplicationDomain::ApplicationDomainType &entity, Sink::Operation) {
371 callback({entity, Sink::Operation_Modification, selectionResult.aggregateValues, selectionResult.aggregateIds}); 379​ callback({entity, Sink::Operation_Modification, selectionResult.aggregateValues, selectionResult.aggregateIds});
372 }); 380​ });
373 } else { 381​ } else {
374 //remove old result 382​ //remove old result
375 if (!oldSelectionResult.isEmpty()) { 383​ if (!oldSelectionResult.isNull()) {
376 readEntity(oldSelectionResult, [&](const Sink::ApplicationDomain::ApplicationDomainType &entity, Sink::Operation) { 384​ readEntity(oldSelectionResult, [&](const Sink::ApplicationDomain::ApplicationDomainType &entity, Sink::Operation) {
377 callback({entity, Sink::Operation_Removal}); 385​ callback({entity, Sink::Operation_Removal});
378 }); 386​ });
379 } 387​ }
380 388​
381 //If the last item has been removed, then there's nothing to add 389​ //If the last item has been removed, then there's nothing to add
382 if (!selectionResult.selection.isEmpty()) { 390​ if (!selectionResult.selection.isNull()) {
383 //add new result 391​ //add new result
384 mSelectedValues.insert(reductionValueBa, selectionResult.selection); 392​ mSelectedValues.insert(reductionValueBa, selectionResult.selection);
385 Q_ASSERT(!selectionResult.selection.isEmpty()); 393​ Q_ASSERT(!selectionResult.selection.isNull());
386 readEntity(selectionResult.selection, [&](const Sink::ApplicationDomain::ApplicationDomainType &entity, Sink::Operation) { 394​ readEntity(selectionResult.selection, [&](const Sink::ApplicationDomain::ApplicationDomainType &entity, Sink::Operation) {
387 callback({entity, Sink::Operation_Creation, selectionResult.aggregateValues, selectionResult.aggregateIds}); 395​ callback({entity, Sink::Operation_Creation, selectionResult.aggregateValues, selectionResult.aggregateIds});
388 }); 396​ });
389 } 397​ }
390 } 398​ }
391 } 399​ }
392 } 400​ }
393 })) 401​ }))
▲ Show 20 Lines • Show All 78 Lines • ▼ Show 20 Line(s)
472DataStoreQuery::State::Ptr DataStoreQuery::getState() 480​DataStoreQuery::State::Ptr DataStoreQuery::getState()
473{ 481​{
474 auto state = State::Ptr::create(); 482​ auto state = State::Ptr::create();
475 state->mSource = mSource; 483​ state->mSource = mSource;
476 state->mCollector = mCollector; 484​ state->mCollector = mCollector;
477 return state; 485​ return state;
478} 486​}
479 487​
480void DataStoreQuery::readEntity(const QByteArray &key, const BufferCallback &resultCallback) 488​void DataStoreQuery::readEntity(const Identifier &id, const BufferCallback &resultCallback)
481{ 489​{
482 mStore.readLatest(mType, key, resultCallback); 490​ mStore.readLatest(mType, id, resultCallback);
483} 491​}
484 492​
485void DataStoreQuery::readPrevious(const QByteArray &key, const std::function<void (const ApplicationDomain::ApplicationDomainType &)> &callback) 493​void DataStoreQuery::readPrevious(const Identifier &id, const std::function<void (const ApplicationDomain::ApplicationDomainType &)> &callback)
486{ 494​{
487 mStore.readPrevious(mType, key, mStore.maxRevision(), callback); 495​ mStore.readPrevious(mType, id, mStore.maxRevision(), callback);
488} 496​}
489 497​
490QVector<QByteArray> DataStoreQuery::indexLookup(const QByteArray &property, const QVariant &value) 498​QVector<Identifier> DataStoreQuery::indexLookup(const QByteArray &property, const QVariant &value)
491{ 499​{
492 return mStore.indexLookup(mType, property, value); 500​ return mStore.indexLookup(mType, property, value);
493} 501​}
494 502​
495/* ResultSet DataStoreQuery::filterAndSortSet(ResultSet &resultSet, const FilterFunction &filter, const QByteArray &sortProperty) */ 503​/* ResultSet DataStoreQuery::filterAndSortSet(ResultSet &resultSet, const FilterFunction &filter, const QByteArray &sortProperty) */
496/* { */ 504​/* { */
497/* const bool sortingRequired = !sortProperty.isEmpty(); */ 505​/* const bool sortingRequired = !sortProperty.isEmpty(); */
498/* if (mInitialQuery && sortingRequired) { */ 506​/* if (mInitialQuery && sortingRequired) { */
▲ Show 20 Lines • Show All 150 Lines • ▼ Show 20 Line(s)645 for (const auto &stage : query.getFilterStages()) {
649 } else if (auto filter = stage.dynamicCast<Query::Bloom>()) { 657​ } else if (auto filter = stage.dynamicCast<Query::Bloom>()) {
650 baseSet = Bloom::Ptr::create(filter->property, baseSet, this); 658​ baseSet = Bloom::Ptr::create(filter->property, baseSet, this);
651 } 659​ }
652 } 660​ }
653 661​
654 mCollector = Collector::Ptr::create(baseSet, this); 662​ mCollector = Collector::Ptr::create(baseSet, this);
655} 663​}
656 664​
657QVector<QByteArray> DataStoreQuery::loadIncrementalResultSet(qint64 baseRevision) 665​QVector<Key> DataStoreQuery::loadIncrementalResultSet(qint64 baseRevision)
658{ 666​{
659 QVector<QByteArray> changedKeys; 667​ QVector<Key> changedKeys;
660 mStore.readRevisions(baseRevision, mType, [&](const QByteArray &key) { 668​ mStore.readRevisions(baseRevision, mType, [&](const Key &key) {
661 changedKeys << key; 669​ changedKeys << key;
662 }); 670​ });
663 return changedKeys; 671​ return changedKeys;
664} 672​}
665 673​
666ResultSet DataStoreQuery::update(qint64 baseRevision) 674​ResultSet DataStoreQuery::update(qint64 baseRevision)
667{ 675​{
668 SinkTraceCtx(mLogCtx) << "Executing query update from revision " << baseRevision << " to revision " << mStore.maxRevision(); 676​ SinkTraceCtx(mLogCtx) << "Executing query update from revision " << baseRevision << " to revision " << mStore.maxRevision();
▲ Show 20 Lines • Show All 45 Lines • Show Last 20 Lines