fix(datastore): fix selective sync expressions to run DDB query when possible (#1100)

* fix(datastore): wrap provided syncexpression in 'and' operator to query items

* fix(datastore): refactor InitialSyncOperation tests

* fix(datastore): update docs

* fix(datastore): move optimization logic down to sync query

* fix(datastore): optimized sync, add more tests

* fix(datastore): optimized sync, querygroup filter test

Author: diegocstn

AmplifyPlugins/Core/AWSPluginsCore/Model/GraphQLRequest/GraphQLRequest+AnyModelWithSync.swift

@@ -170,7 +170,7 @@ extension GraphQLRequest: ModelSyncGraphQLRequestFactory {
         var documentBuilder = ModelBasedGraphQLDocumentBuilder(modelSchema: modelSchema,
                                                                operationType: .query)
         documentBuilder.add(decorator: DirectiveNameDecorator(type: .sync))
-        if let predicate = predicate {
+        if let predicate = optimizePredicate(predicate) {
             documentBuilder.add(decorator: FilterDecorator(filter: predicate.graphQLFilter(for: modelSchema)))
         }
         documentBuilder.add(decorator: PaginationDecorator(limit: limit, nextToken: nextToken))
@@ -207,4 +207,22 @@ extension GraphQLRequest: ModelSyncGraphQLRequestFactory {
                                                   responseType: MutationSyncResult.self,
                                                   decodePath: document.name)
     }
+
+    /// This function tries to optimize provided `QueryPredicate` to perform a DynamoDB query instead of a scan.
+    /// Wrapping the predicate with a group AND enables AppSync to perform the optimization.
+    /// If the provided predicate is already a QueryPredicateGroup, this is not needed.
+    /// If the provided group is of type AND, the optimization will occur.
+    /// If the top level group is OR or NOT, the optimization is not possible anyway.
+    private static func optimizePredicate(_ predicate: QueryPredicate?) -> QueryPredicate? {
+        guard let predicate = predicate else {
+            return nil
+        }
+        if predicate as? QueryPredicateGroup != nil {
+            return predicate
+        } else if let predicate = predicate as? QueryPredicateConstant,
+                  predicate == .all {
+            return predicate
+        }
+        return QueryPredicateGroup(type: .and, predicates: [predicate])
+    }
 }

AmplifyPlugins/Core/AWSPluginsCoreTests/Model/GraphQLRequest/GraphQLRequestAnyModelWithSyncTests.swift

@@ -275,6 +275,65 @@ class GraphQLRequestAnyModelWithSyncTests: XCTestCase {
         XCTAssertEqual(variables["lastSync"] as? Int, lastSync)
     }
 
+    func testOptimizedSyncQueryGraphQLRequestWithFilter() {
+        let modelType = Post.self as Model.Type
+        let nextToken = "nextToken"
+        let limit = 100
+        let lastSync = 123
+        let postId = "123"
+        let predicate = Post.CodingKeys.id.eq(postId)
+        let request = GraphQLRequest<SyncQueryResult>.syncQuery(
+            modelSchema: modelType.schema,
+            where: predicate,
+            limit: limit,
+            nextToken: nextToken,
+            lastSync: lastSync)
+
+        guard let variables = request.variables else {
+            XCTFail("The request doesn't contain variables")
+            return
+        }
+        guard variables["filter"] != nil, let filter = variables["filter"] as? [String: Any] else {
+            XCTFail("The request doesn't contain a filter")
+            return
+        }
+
+        XCTAssertEqual(variables["limit"] as? Int, limit)
+        XCTAssertEqual(variables["nextToken"] as? String, nextToken)
+        XCTAssertNotNil(filter)
+        XCTAssertNotNil(filter["and"])
+    }
+
+    func testSyncQueryGraphQLRequestWithPredicateGroupFilter() {
+        let modelType = Post.self as Model.Type
+        let nextToken = "nextToken"
+        let limit = 100
+        let lastSync = 123
+        let postId = "123"
+        let altPostId = "456"
+        let predicate = Post.CodingKeys.id.eq(postId) || Post.CodingKeys.id.eq(altPostId)
+        let request = GraphQLRequest<SyncQueryResult>.syncQuery(
+            modelSchema: modelType.schema,
+            where: predicate,
+            limit: limit,
+            nextToken: nextToken,
+            lastSync: lastSync)
+
+        guard let variables = request.variables else {
+            XCTFail("The request doesn't contain variables")
+            return
+        }
+        guard variables["filter"] != nil, let filter = variables["filter"] as? [String: Any] else {
+            XCTFail("The request doesn't contain a filter")
+            return
+        }
+
+        XCTAssertEqual(variables["limit"] as? Int, limit)
+        XCTAssertEqual(variables["nextToken"] as? String, nextToken)
+        XCTAssertNotNil(filter)
+        XCTAssertNotNil(filter["or"])
+    }
+
     func testUpdateMutationWithEmptyFilter() {
         let post = Post(title: "title", content: "content", createdAt: .now())
         let documentStringValue = """

AmplifyPlugins/DataStore/AWSDataStoreCategoryPluginTests/Sync/InitialSync/InitialSyncOperationSyncExpressionTests.swift

@@ -0,0 +1,247 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import XCTest
+import SQLite
+import Combine
+
+@testable import Amplify
+@testable import AmplifyTestCommon
+@testable import AWSDataStoreCategoryPlugin
+@testable import AWSPluginsCore
+
+class InitialSyncOperationSyncExpressionTests: XCTestCase {
+    typealias APIPluginQueryResponder = QueryRequestListenerResponder<PaginatedList<AnyModel>>
+
+    var storageAdapter: StorageEngineAdapter!
+    var apiPlugin = MockAPICategoryPlugin()
+    let reconciliationQueue = MockReconciliationQueue()
+    var syncStartedReceived: XCTestExpectation!
+    var syncCompletionReceived: XCTestExpectation!
+    var finishedReceived: XCTestExpectation!
+    var apiWasQueried: XCTestExpectation!
+
+    override func setUpWithError() throws {
+        storageAdapter = try SQLiteStorageEngineAdapter(connection: Connection(.inMemory))
+        try storageAdapter.setUp(modelSchemas: StorageEngine.systemModelSchemas + [MockSynced.schema])
+        syncStartedReceived = expectation(description: "Sync started received, sync operation started")
+        syncCompletionReceived = expectation(description: "Sync completion received, sync operation is complete")
+        finishedReceived = expectation(description: "InitialSyncOperation finishe offering items")
+        apiWasQueried = expectation(description: "API was queried with sync expression")
+    }
+
+    func initialSyncOperation(withSyncExpression syncExpression: DataStoreSyncExpression,
+                              responder: APIPluginQueryResponder) -> InitialSyncOperation {
+        apiPlugin.responders[.queryRequestListener] = responder
+        let configuration  = DataStoreConfiguration.custom(syncPageSize: 10, syncExpressions: [syncExpression])
+        return InitialSyncOperation(
+            modelSchema: MockSynced.schema,
+            api: apiPlugin,
+            reconciliationQueue: reconciliationQueue,
+            storageAdapter: storageAdapter,
+            dataStoreConfiguration: configuration)
+    }
+
+    func testBaseQueryWithBasicSyncExpression() throws {
+        let responder = APIPluginQueryResponder { request, listener in
+            XCTAssertEqual(request.document, """
+            query SyncMockSynceds($filter: ModelMockSyncedFilterInput, $limit: Int) {
+              syncMockSynceds(filter: $filter, limit: $limit) {
+                items {
+                  id
+                  __typename
+                  _version
+                  _deleted
+                  _lastChangedAt
+                }
+                nextToken
+                startedAt
+              }
+            }
+            """)
+            guard let filter = request.variables?["filter"] as? [String: Any?] else {
+                XCTFail("Unable to get filter")
+                return nil
+            }
+            guard let group = filter["and"] as? [[String: Any?]] else {
+                XCTFail("Unable to find 'and' group")
+                return nil
+            }
+
+            guard let key = group[0]["id"] as? [String: Any?] else {
+                XCTFail("Unable to get id from filter")
+                return nil
+            }
+            guard let value = key["eq"] as? String else {
+                XCTFail("Unable to get eq from key")
+                return nil
+            }
+            XCTAssertEqual(value, "id-123")
+
+            let list = PaginatedList<AnyModel>(items: [], nextToken: nil, startedAt: nil)
+            let event: GraphQLOperation<PaginatedList<AnyModel>>.OperationResult = .success(.success(list))
+            listener?(event)
+            self.apiWasQueried.fulfill()
+            return nil
+        }
+
+        let syncExpression = DataStoreSyncExpression.syncExpression(MockSynced.schema, where: {
+            MockSynced.keys.id == "id-123"
+        })
+        let operation = initialSyncOperation(withSyncExpression: syncExpression, responder: responder)
+
+        let sink = operation
+            .publisher
+            .sink(receiveCompletion: { _ in
+                self.syncCompletionReceived.fulfill()
+            }, receiveValue: { [self] value in
+                switch value {
+                case .started(modelName: let modelName, syncType: let syncType):
+                    XCTAssertEqual(modelName, "MockSynced")
+                    XCTAssertEqual(syncType, .fullSync)
+                    syncStartedReceived.fulfill()
+                case .finished(modelName: let modelName):
+                    XCTAssertEqual(modelName, "MockSynced")
+                    finishedReceived.fulfill()
+                default:
+                    break
+                }
+            })
+
+        operation.main()
+
+        waitForExpectations(timeout: 1)
+        sink.cancel()
+    }
+
+    func testBaseQueryWithFilterSyncExpression() throws {
+        let responder = APIPluginQueryResponder { request, listener in
+            XCTAssertEqual(request.document, """
+            query SyncMockSynceds($filter: ModelMockSyncedFilterInput, $limit: Int) {
+              syncMockSynceds(filter: $filter, limit: $limit) {
+                items {
+                  id
+                  __typename
+                  _version
+                  _deleted
+                  _lastChangedAt
+                }
+                nextToken
+                startedAt
+              }
+            }
+            """)
+            guard let filter = request.variables?["filter"] as? [String: Any?] else {
+                XCTFail("Unable to get filter")
+                return nil
+            }
+            guard let group = filter["or"] as? [[String: Any?]] else {
+                XCTFail("Unable to find 'or' group")
+                return nil
+            }
+
+            guard let key = group[0]["id"] as? [String: Any?] else {
+                XCTFail("Unable to get id from filter")
+                return nil
+            }
+            guard let value = key["eq"] as? String else {
+                XCTFail("Unable to get eq from key")
+                return nil
+            }
+            XCTAssertEqual(value, "id-123")
+
+            let list = PaginatedList<AnyModel>(items: [], nextToken: nil, startedAt: nil)
+            let event: GraphQLOperation<PaginatedList<AnyModel>>.OperationResult = .success(.success(list))
+            listener?(event)
+            self.apiWasQueried.fulfill()
+            return nil
+        }
+
+        let syncExpression = DataStoreSyncExpression.syncExpression(MockSynced.schema, where: {
+            MockSynced.keys.id == "id-123" || MockSynced.keys.id == "id-456"
+        })
+        let operation = initialSyncOperation(withSyncExpression: syncExpression, responder: responder)
+
+        let sink = operation
+            .publisher
+            .sink(receiveCompletion: { _ in
+                self.syncCompletionReceived.fulfill()
+            }, receiveValue: { [self] value in
+                switch value {
+                case .started(modelName: let modelName, syncType: let syncType):
+                    XCTAssertEqual(modelName, "MockSynced")
+                    XCTAssertEqual(syncType, .fullSync)
+                    syncStartedReceived.fulfill()
+                case .finished(modelName: let modelName):
+                    XCTAssertEqual(modelName, "MockSynced")
+                    finishedReceived.fulfill()
+                default:
+                    break
+                }
+            })
+
+        operation.main()
+
+        waitForExpectations(timeout: 1)
+        sink.cancel()
+    }
+
+    func testBaseQueryWithSyncExpressionConstantAll() throws {
+        let responder = APIPluginQueryResponder { request, listener in
+            XCTAssertEqual(request.document, """
+            query SyncMockSynceds($limit: Int) {
+              syncMockSynceds(limit: $limit) {
+                items {
+                  id
+                  __typename
+                  _version
+                  _deleted
+                  _lastChangedAt
+                }
+                nextToken
+                startedAt
+              }
+            }
+            """)
+            XCTAssertNil(request.variables?["filter"])
+
+            let list = PaginatedList<AnyModel>(items: [], nextToken: nil, startedAt: nil)
+            let event: GraphQLOperation<PaginatedList<AnyModel>>.OperationResult = .success(.success(list))
+            listener?(event)
+            self.apiWasQueried.fulfill()
+            return nil
+        }
+
+        let syncExpression = DataStoreSyncExpression.syncExpression(MockSynced.schema, where: {
+            QueryPredicateConstant.all
+        })
+        let operation = initialSyncOperation(withSyncExpression: syncExpression, responder: responder)
+
+        let sink = operation
+            .publisher
+            .sink(receiveCompletion: { _ in
+                self.syncCompletionReceived.fulfill()
+            }, receiveValue: { [self] value in
+                switch value {
+                case .started(modelName: let modelName, syncType: let syncType):
+                    XCTAssertEqual(modelName, "MockSynced")
+                    XCTAssertEqual(syncType, .fullSync)
+                    syncStartedReceived.fulfill()
+                case .finished(modelName: let modelName):
+                    XCTAssertEqual(modelName, "MockSynced")
+                    finishedReceived.fulfill()
+                default:
+                    break
+                }
+            })
+
+        operation.main()
+
+        waitForExpectations(timeout: 1)
+        sink.cancel()
+    }
+}