4 files changed, 29 insertions, 23 deletions

diff --git a/common/datastorequery.cpp b/common/datastorequery.cpp
index 0db59e1..2e0c348 100644
--- a/common/datastorequery.cpp
+++ b/common/datastorequery.cpp
@@ -144,7 +144,7 @@ public:
             const auto property = entity.getProperty(filterProperty);
             const auto comparator = propertyFilter.value(filterProperty);
             if (!comparator.matches(property)) {
-                SinkTraceCtx(mDatastore->mLogCtx) << "Filtering entity due to property mismatch on filter: " << filterProperty << property << ":" << comparator.value;
+                SinkTraceCtx(mDatastore->mLogCtx) << "Filtering entity due to property mismatch on filter: " << entity.identifier() << "Property: " << filterProperty << property << " Filter:" << comparator.value;
                 return false;
             }
         }
@@ -367,19 +367,22 @@ public:
 DataStoreQuery::DataStoreQuery(const Sink::QueryBase &query, const QByteArray &type, EntityStore &store)
     : mType(type), mStore(store), mLogCtx(store.logContext().subContext("datastorequery"))
 {
+    //This is what we use during a new query
     setupQuery(query);
 }
 
-DataStoreQuery::DataStoreQuery(const DataStoreQuery::State &state, const QByteArray &type, Sink::Storage::EntityStore &store)
+DataStoreQuery::DataStoreQuery(const DataStoreQuery::State &state, const QByteArray &type, Sink::Storage::EntityStore &store, bool incremental)
     : mType(type), mStore(store), mLogCtx(store.logContext().subContext("datastorequery"))
 {
+    //This is what we use when fetching more data, without having a new revision with incremental=false
+    //And this is what we use when the data changed and we want to update with incremental = true
     mCollector = state.mCollector;
     mSource = state.mSource;
 
     auto source = mCollector;
     while (source) {
         source->mDatastore = this;
-        source->mIncremental = true;
+        source->mIncremental = incremental;
         source = source->mSource;
     }
 }
diff --git a/common/datastorequery.h b/common/datastorequery.h
index a797782..ee5f99e 100644
--- a/common/datastorequery.h
+++ b/common/datastorequery.h
@@ -46,7 +46,7 @@ public:
     };
 
     DataStoreQuery(const Sink::QueryBase &query, const QByteArray &type, Sink::Storage::EntityStore &store);
-    DataStoreQuery(const DataStoreQuery::State &state, const QByteArray &type, Sink::Storage::EntityStore &store);
+    DataStoreQuery(const DataStoreQuery::State &state, const QByteArray &type, Sink::Storage::EntityStore &store, bool incremental);
     ~DataStoreQuery();
     ResultSet execute();
     ResultSet update(qint64 baseRevision);
diff --git a/common/queryrunner.cpp b/common/queryrunner.cpp
index 802fc48..43f48c0 100644
--- a/common/queryrunner.cpp
+++ b/common/queryrunner.cpp
@@ -51,7 +51,7 @@ public:
     virtual ~QueryWorker();
 
     ReplayResult executeIncrementalQuery(const Sink::Query &query, Sink::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, DataStoreQuery::State::Ptr state);
-    ReplayResult executeInitialQuery(const Sink::Query &query, const typename DomainType::Ptr &parent, Sink::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, int offset, int batchsize);
+    ReplayResult executeInitialQuery(const Sink::Query &query, const typename DomainType::Ptr &parent, Sink::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, int batchsize, DataStoreQuery::State::Ptr state);
 
 private:
     void resultProviderCallback(const Sink::Query &query, Sink::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, const ResultSet::Result &result);
@@ -72,18 +72,18 @@ QueryRunner<DomainType>::QueryRunner(const Sink::Query &query, const Sink::Resou
     auto guardPtr = QPointer<QObject>(&guard);
     auto fetcher = [=](const typename DomainType::Ptr &parent) {
         const QByteArray parentId = parent ? parent->identifier() : QByteArray();
-        SinkTraceCtx(mLogCtx) << "Running fetcher. Offset: " << mOffset[parentId] << " Batchsize: " << mBatchSize;
+        SinkTraceCtx(mLogCtx) << "Running fetcher. Batchsize: " << mBatchSize;
         auto resultProvider = mResultProvider;
         auto resultTransformation = mResultTransformation;
-        auto offset = mOffset[parentId];
         auto batchSize = mBatchSize;
         auto resourceContext = mResourceContext;
         auto logCtx = mLogCtx;
+        auto state = mQueryState.value(parentId);
         const bool runAsync = !query.synchronousQuery();
         //The lambda will be executed in a separate thread, so copy all arguments
-        async::run<ReplayResult>([resultTransformation, offset, batchSize, query, bufferType, resourceContext, resultProvider, parent, logCtx]() {
+        async::run<ReplayResult>([=]() {
             QueryWorker<DomainType> worker(query, resourceContext, bufferType, resultTransformation, logCtx);
-            return worker.executeInitialQuery(query, parent, *resultProvider, offset, batchSize);
+            return worker.executeInitialQuery(query, parent, *resultProvider, batchSize, state);
         }, runAsync)
             .then([this, parentId, query, parent, resultProvider, guardPtr](const ReplayResult &result) {
                 if (!guardPtr) {
@@ -91,8 +91,7 @@ QueryRunner<DomainType>::QueryRunner(const Sink::Query &query, const Sink::Resou
                     return;
                 }
                 mInitialQueryComplete = true;
-                mQueryState = result.queryState;
-                mOffset[parentId] += result.replayedEntities;
+                mQueryState[parentId] = result.queryState;
                 // Only send the revision replayed information if we're connected to the resource, there's no need to start the resource otherwise.
                 if (query.liveQuery()) {
                     mResourceAccess->sendRevisionReplayedCommand(result.newRevision);
@@ -111,10 +110,11 @@ QueryRunner<DomainType>::QueryRunner(const Sink::Query &query, const Sink::Resou
         Q_ASSERT(!query.synchronousQuery());
         // Incremental updates are always loaded directly, leaving it up to the result to discard the changes if they are not interesting
         setQuery([=]() -> KAsync::Job<void> {
+            const QByteArray parentId;
             auto resultProvider = mResultProvider;
             auto resourceContext = mResourceContext;
             auto logCtx = mLogCtx;
-            auto state = mQueryState;
+            auto state = mQueryState.value(parentId);
             if (!mInitialQueryComplete) {
                 SinkWarningCtx(mLogCtx) << "Can't start the incremental query before the initial query is complete";
                 fetcher({});
@@ -225,7 +225,7 @@ ReplayResult QueryWorker<DomainType>::executeIncrementalQuery(const Sink::Query
         SinkWarningCtx(mLogCtx) << "No previous query state.";
         return {0, 0, false, DataStoreQuery::State::Ptr{}};
     }
-    auto preparedQuery = DataStoreQuery{*state, ApplicationDomain::getTypeName<DomainType>(), entityStore};
+    auto preparedQuery = DataStoreQuery{*state, ApplicationDomain::getTypeName<DomainType>(), entityStore, true};
     auto resultSet = preparedQuery.update(baseRevision);
     SinkTraceCtx(mLogCtx) << "Filtered set retrieved. " << Log::TraceTime(time.elapsed());
     auto replayResult = resultSet.replaySet(0, 0, [this, query, &resultProvider](const ResultSet::Result &result) {
@@ -240,7 +240,7 @@ ReplayResult QueryWorker<DomainType>::executeIncrementalQuery(const Sink::Query
 
 template <class DomainType>
 ReplayResult QueryWorker<DomainType>::executeInitialQuery(
-    const Sink::Query &query, const typename DomainType::Ptr &parent, Sink::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, int offset, int batchsize)
+    const Sink::Query &query, const typename DomainType::Ptr &parent, Sink::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, int batchsize, DataStoreQuery::State::Ptr state)
 {
     QTime time;
     time.start();
@@ -257,11 +257,17 @@ ReplayResult QueryWorker<DomainType>::executeInitialQuery(
     }
 
     auto entityStore = EntityStore{mResourceContext, mLogCtx};
-    auto preparedQuery = DataStoreQuery{modifiedQuery, ApplicationDomain::getTypeName<DomainType>(), entityStore};
-    auto resultSet = preparedQuery.execute();
+    auto preparedQuery = [&] {
+        if (state) {
+            return DataStoreQuery{*state, ApplicationDomain::getTypeName<DomainType>(), entityStore, false};
+        } else {
+            return DataStoreQuery{modifiedQuery, ApplicationDomain::getTypeName<DomainType>(), entityStore};
+        }
+    }();
+    auto resultSet = preparedQuery.execute();;
 
-    SinkTraceCtx(mLogCtx) << "Filtered set retrieved. " << Log::TraceTime(time.elapsed());
-    auto replayResult = resultSet.replaySet(offset, batchsize, [this, query, &resultProvider](const ResultSet::Result &result) {
+    SinkTraceCtx(mLogCtx) << "Filtered set retrieved." << Log::TraceTime(time.elapsed());
+    auto replayResult = resultSet.replaySet(0, batchsize, [this, query, &resultProvider](const ResultSet::Result &result) {
         resultProviderCallback(query, resultProvider, result);
     });
 
@@ -269,9 +275,7 @@ ReplayResult QueryWorker<DomainType>::executeInitialQuery(
         << (replayResult.replayedAll ? "Replayed all available results.\n" : "")
         << "Initial query took: " << Log::TraceTime(time.elapsed());
 
-    auto state = preparedQuery.getState();
-
-    return {entityStore.maxRevision(), replayResult.replayedEntities, replayResult.replayedAll, state};
+    return {entityStore.maxRevision(), replayResult.replayedEntities, replayResult.replayedAll, preparedQuery.getState()};
 }
 
 #define REGISTER_TYPE(T) \
diff --git a/common/queryrunner.h b/common/queryrunner.h
index f5c7ead..5308eac 100644
--- a/common/queryrunner.h
+++ b/common/queryrunner.h
@@ -98,11 +98,10 @@ private:
     QSharedPointer<Sink::ResourceAccessInterface> mResourceAccess;
     QSharedPointer<Sink::ResultProvider<typename DomainType::Ptr>> mResultProvider;
     ResultTransformation mResultTransformation;
-    QHash<QByteArray, qint64> mOffset;
+    QHash<QByteArray, DataStoreQuery::State::Ptr> mQueryState;
     int mBatchSize;
     QObject guard;
     Sink::Log::Context mLogCtx;
-    DataStoreQuery::State::Ptr mQueryState;
     bool mInitialQueryComplete = false;
     bool mQueryInProgress = false;
 };