fix(datastore): base sync when sync expression changes

aws-api-appsync/src/main/java/com/amplifyframework/core/model/query/predicate/GsonPredicateAdapters.java

@@ -15,6 +15,7 @@
 
 package com.amplifyframework.core.model.query.predicate;
 
+import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import com.google.gson.JsonDeserializationContext;
 import com.google.gson.JsonDeserializer;
@@ -33,13 +34,15 @@ public final class GsonPredicateAdapters {
     private GsonPredicateAdapters() {}
 
     /**
-     * Registers the adapters into an {@link GsonBuilder}.
+     * Registers the QueryPredicate adapter into an {@link GsonBuilder}.
+     * registerTypeHierarchyAdapter enables objects assigned to concrete QueryPredicate classes
+     * (e.g., QueryPredicateOperation) to use this adapter.
+     * 
      * @param builder A GsonBuilder.
      */
     public static void register(GsonBuilder builder) {
         builder
-            .registerTypeAdapter(QueryOperator.class, new QueryOperatorAdapter())
-            .registerTypeAdapter(QueryPredicate.class, new QueryPredicateAdapter());
+            .registerTypeHierarchyAdapter(QueryPredicate.class, new QueryPredicateAdapter());
     }
 
     /**
@@ -127,10 +130,17 @@ public static final class QueryPredicateAdapter implements
             JsonDeserializer<QueryPredicate>, JsonSerializer<QueryPredicate> {
         private static final String TYPE = "_type";
 
+        // internal Gson instance for avoiding infinite loop
+        private final Gson gson = new GsonBuilder()
+                .registerTypeAdapter(QueryOperator.class, new QueryOperatorAdapter())
+                .serializeNulls()
+                .create();
+
         private enum PredicateType {
             OPERATION,
             GROUP,
-            ALL
+            ALL,
+            NONE
         }
 
         /**
@@ -147,11 +157,13 @@ public QueryPredicate deserialize(JsonElement json, Type type, JsonDeserializati
             String predicateType = jsonObject.get(TYPE).getAsString();
             switch (PredicateType.valueOf(predicateType)) {
                 case OPERATION:
-                    return context.deserialize(json, QueryPredicateOperation.class);
+                    return gson.fromJson(json, QueryPredicateOperation.class);
                 case GROUP:
-                    return context.deserialize(json, QueryPredicateGroup.class);
+                    return gson.fromJson(json, QueryPredicateGroup.class);
                 case ALL:
-                    return context.deserialize(json, MatchAllQueryPredicate.class);
+                    return gson.fromJson(json, MatchAllQueryPredicate.class);
+                case NONE:
+                    return gson.fromJson(json, MatchNoneQueryPredicate.class);
                 default:
                     throw new JsonParseException("Unable to deserialize " +
                             json.toString() + " to QueryPredicate instance.");
@@ -164,16 +176,15 @@ public QueryPredicate deserialize(JsonElement json, Type type, JsonDeserializati
         @Override
         public JsonElement serialize(QueryPredicate predicate, Type type, JsonSerializationContext context)
                 throws JsonParseException {
-            JsonElement json;
+            JsonElement json = gson.toJsonTree(predicate);
             PredicateType predicateType;
             if (predicate instanceof MatchAllQueryPredicate) {
                 predicateType = PredicateType.ALL;
-                json = context.serialize(predicate, MatchAllQueryPredicate.class);
+            } else if (predicate instanceof MatchNoneQueryPredicate) {
+                predicateType = PredicateType.NONE;
             } else if (predicate instanceof QueryPredicateOperation) {
-                json = context.serialize(predicate, QueryPredicateOperation.class);
                 predicateType = PredicateType.OPERATION;
             } else if (predicate instanceof QueryPredicateGroup) {
-                json = context.serialize(predicate, QueryPredicateGroup.class);
                 predicateType = PredicateType.GROUP;
             } else {
                 throw new JsonParseException("Unable to identify the predicate type.");

aws-datastore/src/main/java/com/amplifyframework/datastore/DataStoreConfiguration.java

@@ -512,6 +512,20 @@ public Builder syncExpression(@NonNull String modelName,
             return Builder.this;
         }
 
+        /**
+         * Sets sync expressions for models to filter which data is synced locally.
+         * The expression is evaluated each time DataStore is started.
+         * The QueryPredicate is applied on both sync and subscriptions.
+         * [NOTE: this will override the previous syncExpression configuration if there's any]
+         * @param syncExpressions the Map of {@link DataStoreSyncExpression}s to filter data
+         * @return Current builder
+         */
+        @NonNull
+        public Builder syncExpressions(@NonNull Map<String, DataStoreSyncExpression> syncExpressions) {
+            this.syncExpressions = Objects.requireNonNull(syncExpressions);
+            return Builder.this;
+        }
+
         private void populateSettingsFromJson() throws DataStoreException {
             if (pluginJson == null) {
                 return;

aws-datastore/src/main/java/com/amplifyframework/datastore/storage/sqlite/SQLiteStorageAdapter.java

@@ -934,7 +934,10 @@ private Completable updateModels() {
                     Objects.requireNonNull(databaseConnectionHandle);
                     sqliteStorageHelper.update(databaseConnectionHandle, oldVersion, newVersion);
                 } else {
-                    LOG.debug("Database up to date. Checking ModelMetadata.");
+                    // We only need to do the model migration here because the current implementation of
+                    // sqliteStorageHelper.update will drop all existing tables and recreate tables with new schemas,
+                    // However, this might be changed in the future
+                    LOG.debug("Database up to date. Checking System Models.");
                     new ModelMigrations(databaseConnectionHandle, modelsProvider).apply();
                 }
             }

aws-datastore/src/main/java/com/amplifyframework/datastore/storage/sqlite/migrations/AddSyncExpressionToLastSyncMetadata.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2024 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package com.amplifyframework.datastore.storage.sqlite.migrations;
+
+import android.database.Cursor;
+import android.database.sqlite.SQLiteDatabase;
+
+import com.amplifyframework.core.Amplify;
+import com.amplifyframework.core.category.CategoryType;
+import com.amplifyframework.logging.Logger;
+import com.amplifyframework.util.Wrap;
+
+/**
+ * Add SyncExpression (TEXT) column to LastSyncMetadata table.
+ */
+public final class AddSyncExpressionToLastSyncMetadata implements ModelMigration {
+    private static final Logger LOG = Amplify.Logging.logger(CategoryType.DATASTORE, "amplify:aws-datastore");
+    private final SQLiteDatabase database;
+    private final String newSyncExpColumnName = "syncExpression";
+
+    /**
+     * Constructor for the migration class.
+     * @param database Connection to the SQLite database.
+     */
+    public AddSyncExpressionToLastSyncMetadata(SQLiteDatabase database) {
+        this.database = database;
+    }
+
+    @Override
+    public void apply() {
+        if (!needsMigration()) {
+            LOG.debug("No LastSyncMetadata migration needed.");
+            return;
+        }
+        addNewSyncExpColumnName();
+    }
+
+    /**
+     * Alter LastSyncMetadata table with new column.
+     * Existing rows in LasySyncMetadata will have 'null' for ${newSyncExpColumnName} value,
+     * until the next sync/hydrate operation.
+     */
+    private void addNewSyncExpColumnName() {
+        try {
+            database.beginTransaction();
+            final String addColumnSql = "ALTER TABLE LastSyncMetadata ADD COLUMN " +
+                    newSyncExpColumnName + " TEXT";
+            database.execSQL(addColumnSql);
+            database.setTransactionSuccessful();
+            LOG.debug("Successfully upgraded LastSyncMetadata table with new field: " + newSyncExpColumnName);
+        } finally {
+            if (database.inTransaction()) {
+                database.endTransaction();
+            }
+        }
+    }
+
+    private boolean needsMigration() {
+        final String checkColumnSql = "SELECT COUNT(*) FROM pragma_table_info('LastSyncMetadata') " +
+            "WHERE name=" + Wrap.inSingleQuotes(newSyncExpColumnName);
+        try (Cursor queryResults = database.rawQuery(checkColumnSql, new String[]{})) {
+            if (queryResults.moveToNext()) {
+                int recordNum = queryResults.getInt(0);
+                return recordNum == 0; // needs to be upgraded if there's no column named ${newSyncExpColumnName}
+            }
+        }
+        return false;
+    }
+}

aws-datastore/src/main/java/com/amplifyframework/datastore/storage/sqlite/migrations/ModelMigrations.java

@@ -37,6 +37,7 @@ public class ModelMigrations {
     public ModelMigrations(SQLiteDatabase databaseConnectionHandle, ModelProvider modelsProvider) {
         List<ModelMigration> migrationClasses = new ArrayList<>();
         migrationClasses.add(new AddModelNameToModelMetadataKey(databaseConnectionHandle, modelsProvider));
+        migrationClasses.add(new AddSyncExpressionToLastSyncMetadata(databaseConnectionHandle));
         this.modelMigrations = Immutable.of(migrationClasses);
     }
 

aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/LastSyncMetadata.java

@@ -22,12 +22,13 @@
 import com.amplifyframework.core.model.Model;
 import com.amplifyframework.core.model.annotations.ModelConfig;
 import com.amplifyframework.core.model.annotations.ModelField;
+import com.amplifyframework.core.model.query.predicate.QueryPredicate;
 
 import java.util.Objects;
 import java.util.UUID;
 
 /**
- * Metadata about the last time that a model class was sync'd with AppSync backend.
+ * Metadata about the last time that a model class was sync'd with AppSync backend using a certain syncExpression.
  * This metadata is persisted locally as a system model. This metadata is inspected
  * whenever the Sync Engine starts up. The system consider the value of
  * {@link LastSyncMetadata#getLastSyncTime()} to decide whether or not it should
@@ -39,40 +40,48 @@ public final class LastSyncMetadata implements Model {
     private final @ModelField(targetType = "String", isRequired = true) String modelClassName;
     private final @ModelField(targetType = "AWSTimestamp", isRequired = true) Long lastSyncTime;
     private final @ModelField(targetType = "String", isRequired = true) String lastSyncType;
+    private final @ModelField(targetType = "String") QueryPredicate syncExpression;
 
     @SuppressWarnings("checkstyle:ParameterName") // The field is named "id" in the model; keep it consistent
-    private LastSyncMetadata(String id, String modelClassName, Long lastSyncTime, SyncType syncType) {
+    private LastSyncMetadata(String id, String modelClassName, Long lastSyncTime,
+                             SyncType syncType, @Nullable QueryPredicate syncExpression) {
         this.id = id;
         this.modelClassName = modelClassName;
         this.lastSyncTime = lastSyncTime;
         this.lastSyncType = syncType.name();
+        this.syncExpression = syncExpression;
     }
 
     /**
      * Creates an instance of an {@link LastSyncMetadata}, indicating that the provided
      * model has been base sync'd, and that the last sync occurred at the given time.
      * @param modelClassName Name of model
      * @param lastSyncTime Last time it was synced
+     * @param syncExpression the corresponding sync expression being used during last sync
      * @param <T> t type of Model.
      * @return {@link LastSyncMetadata} for the model class
      */
     public static <T extends Model> LastSyncMetadata baseSyncedAt(@NonNull String modelClassName,
-                                                           @Nullable long lastSyncTime) {
+                                                                  @Nullable long lastSyncTime,
+                                                                  @Nullable QueryPredicate syncExpression) {
         Objects.requireNonNull(modelClassName);
-        return create(modelClassName, lastSyncTime, SyncType.BASE);
+        return create(modelClassName, lastSyncTime, SyncType.BASE, syncExpression);
     }
 
     /**
      * Creates an instance of an {@link LastSyncMetadata}, indicating that the provided
      * model has been base delta sync'd, and that the last sync occurred at the given time.
      * @param modelClassName Name of model
      * @param lastSyncTime Last time it was synced
+     * @param syncExpression the corresponding sync expression being used during last sync
+     * @param <T> t type of Model.
      * @return {@link LastSyncMetadata} for the model class
      */
     static <T extends Model> LastSyncMetadata deltaSyncedAt(@NonNull String modelClassName,
-                                                           @Nullable long lastSyncTime) {
+                                                            @Nullable long lastSyncTime,
+                                                            @Nullable QueryPredicate syncExpression) {
         Objects.requireNonNull(modelClassName);
-        return create(modelClassName, lastSyncTime, SyncType.DELTA);
+        return create(modelClassName, lastSyncTime, SyncType.DELTA, syncExpression);
     }
 
     /**
@@ -84,22 +93,25 @@ static <T extends Model> LastSyncMetadata deltaSyncedAt(@NonNull String modelCla
      */
     public static <T extends Model> LastSyncMetadata neverSynced(@NonNull String modelClassName) {
         Objects.requireNonNull(modelClassName);
-        return create(modelClassName, null, SyncType.BASE);
+        return create(modelClassName, null, SyncType.BASE, null);
     }
 
     /**
      * Creates an {@link LastSyncMetadata} for the provided model class.
      * @param modelClassName Name of model class for which metadata pertains
      * @param lastSyncTime Time of last sync; null, if never.
      * @param syncType The type of sync (FULL or DELTA).
+     * @param syncExpression the corresponding sync expression being used during last sync
      * @param <T> Type of model
      * @return {@link LastSyncMetadata}
      */
     @SuppressWarnings("WeakerAccess")
-    static <T extends Model> LastSyncMetadata create(
-            @NonNull String modelClassName, @Nullable Long lastSyncTime, @NonNull SyncType syncType) {
+    static <T extends Model> LastSyncMetadata create(@NonNull String modelClassName,
+                                                     @Nullable Long lastSyncTime,
+                                                     @NonNull SyncType syncType,
+                                                     @Nullable QueryPredicate syncExpression) {
         Objects.requireNonNull(modelClassName);
-        return new LastSyncMetadata(hash(modelClassName), modelClassName, lastSyncTime, syncType);
+        return new LastSyncMetadata(hash(modelClassName), modelClassName, lastSyncTime, syncType, syncExpression);
     }
 
     @NonNull
@@ -144,6 +156,14 @@ public String getLastSyncType() {
         return lastSyncType;
     }
 
+    /**
+     * Returns the sync expression being used in the last sync.
+     * @return A serialized sync expression
+     */
+    public QueryPredicate getSyncExpression() {
+        return this.syncExpression;
+    }
+
     /**
      * Computes a stable hash for a model class, by its name.
      * Since {@link Model}s have to have unique IDs, we need an ID for this class.
@@ -175,6 +195,9 @@ public boolean equals(Object thatObject) {
         if (!ObjectsCompat.equals(lastSyncType, that.lastSyncType)) {
             return false;
         }
+        if (!ObjectsCompat.equals(syncExpression, that.syncExpression)) {
+            return false;
+        }
         return ObjectsCompat.equals(lastSyncTime, that.lastSyncTime);
     }
 
@@ -184,6 +207,7 @@ public int hashCode() {
         result = 31 * result + modelClassName.hashCode();
         result = 31 * result + lastSyncTime.hashCode();
         result = 31 * result + lastSyncType.hashCode();
+        result = 31 * result + syncExpression.hashCode();
         return result;
     }
 
@@ -194,6 +218,7 @@ public String toString() {
             ", modelClassName='" + modelClassName + '\'' +
             ", lastSyncTime=" + lastSyncTime +
             ", lastSyncType=" + lastSyncType +
+            ", syncExpression=" + syncExpression +
             '}';
     }
 }

aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/SyncProcessor.java

@@ -97,7 +97,6 @@ private SyncProcessor(Builder builder) {
         this.queryPredicateProvider = builder.queryPredicateProvider;
         this.requestRetry = builder.requestRetry;
         this.isSyncRetryEnabled = builder.isSyncRetryEnabled;
-
         if (!this.isSyncRetryEnabled) {
             LOG.warn("Disabling sync retries will be deprecated in a future version.");
         }
@@ -182,7 +181,8 @@ Completable hydrate() {
 
     private Completable createHydrationTask(ModelSchema schema) {
         ModelSyncMetricsAccumulator metricsAccumulator = new ModelSyncMetricsAccumulator(schema.getName());
-        return syncTimeRegistry.lookupLastSyncTime(schema.getName())
+        QueryPredicate currentSyncExpression = this.queryPredicateProvider.getPredicate(schema.getName());
+        return syncTimeRegistry.lookupLastSyncTime(schema.getName(), currentSyncExpression)
             .map(this::filterOutOldSyncTimes)
             // And for each, perform a sync. The network response will contain an Iterable<ModelWithMetadata<T>>
             .flatMap(lastSyncTime -> {
@@ -202,8 +202,8 @@ private Completable createHydrationTask(ModelSchema schema) {
             })
             .flatMapCompletable(syncType -> {
                 Completable syncTimeSaveCompletable = SyncType.DELTA.equals(syncType) ?
-                    syncTimeRegistry.saveLastDeltaSyncTime(schema.getName(), SyncTime.now()) :
-                    syncTimeRegistry.saveLastBaseSyncTime(schema.getName(), SyncTime.now());
+                    syncTimeRegistry.saveLastDeltaSync(schema.getName(), SyncTime.now(), currentSyncExpression) :
+                    syncTimeRegistry.saveLastBaseSync(schema.getName(), SyncTime.now(), currentSyncExpression);
                 return syncTimeSaveCompletable.andThen(Completable.fromAction(() ->
                     Amplify.Hub.publish(
                         HubChannel.DATASTORE, metricsAccumulator.toModelSyncedEvent(syncType).toHubEvent()