Index-Free (4/6): Filter document queries by read time

firebase-firestore/src/main/java/com/google/firebase/firestore/local/IndexedQueryEngine.java

@@ -29,6 +29,7 @@
 import com.google.firebase.firestore.model.DocumentKey;
 import com.google.firebase.firestore.model.FieldPath;
 import com.google.firebase.firestore.model.MaybeDocument;
+import com.google.firebase.firestore.model.SnapshotVersion;
 import com.google.firebase.firestore.model.value.ArrayValue;
 import com.google.firebase.firestore.model.value.BooleanValue;
 import com.google.firebase.firestore.model.value.DoubleValue;
@@ -99,7 +100,7 @@ public IndexedQueryEngine(
   @Override
   public ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingQuery(Query query) {
     return query.isDocumentQuery()
-        ? localDocuments.getDocumentsMatchingQuery(query)
+        ? localDocuments.getDocumentsMatchingQuery(query, SnapshotVersion.NONE)
         : performCollectionQuery(query);
   }
 
@@ -118,7 +119,7 @@ private ImmutableSortedMap<DocumentKey, Document> performCollectionQuery(Query q
           "If there are any filters, we should be able to use an index.");
       // TODO: Call overlay.getCollectionDocuments(query.getPath()) and filter the
       // results (there may still be startAt/endAt bounds that apply).
-      filteredResults = localDocuments.getDocumentsMatchingQuery(query);
+      filteredResults = localDocuments.getDocumentsMatchingQuery(query, SnapshotVersion.NONE);
     }
 
     return filteredResults;

firebase-firestore/src/main/java/com/google/firebase/firestore/local/LocalDocumentsView.java

@@ -131,15 +131,22 @@ ImmutableSortedMap<DocumentKey, MaybeDocument> getLocalViewOfDocuments(
   // documents in a given collection so that SimpleQueryEngine can do that and then filter in
   // memory.
 
-  /** Performs a query against the local view of all documents. */
-  ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingQuery(Query query) {
+  /**
+   * Performs a query against the local view of all documents.
+   *
+   * @param query The query to match documents against.
+   * @param sinceReadTime If not set to SnapshotVersion.MIN, return only documents that have been
+   *     read since this snapshot version (exclusive).
+   */
+  ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingQuery(
+      Query query, SnapshotVersion sinceReadTime) {
     ResourcePath path = query.getPath();
     if (query.isDocumentQuery()) {
       return getDocumentsMatchingDocumentQuery(path);
     } else if (query.isCollectionGroupQuery()) {
-      return getDocumentsMatchingCollectionGroupQuery(query);
+      return getDocumentsMatchingCollectionGroupQuery(query, sinceReadTime);
     } else {
-      return getDocumentsMatchingCollectionQuery(query);
+      return getDocumentsMatchingCollectionQuery(query, sinceReadTime);
     }
   }
 
@@ -156,7 +163,7 @@ private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingDocumentQu
   }
 
   private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingCollectionGroupQuery(
-      Query query) {
+      Query query, SnapshotVersion sinceReadTime) {
     hardAssert(
         query.getPath().isEmpty(),
         "Currently we only support collection group queries at the root.");
@@ -169,7 +176,7 @@ private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingCollection
     for (ResourcePath parent : parents) {
       Query collectionQuery = query.asCollectionQueryAtPath(parent.append(collectionId));
       ImmutableSortedMap<DocumentKey, Document> collectionResults =
-          getDocumentsMatchingCollectionQuery(collectionQuery);
+          getDocumentsMatchingCollectionQuery(collectionQuery, sinceReadTime);
       for (Map.Entry<DocumentKey, Document> docEntry : collectionResults) {
         results = results.insert(docEntry.getKey(), docEntry.getValue());
       }
@@ -179,9 +186,9 @@ private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingCollection
 
   /** Queries the remote documents and overlays mutations. */
   private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingCollectionQuery(
-      Query query) {
+      Query query, SnapshotVersion sinceReadTime) {
     ImmutableSortedMap<DocumentKey, Document> results =
-        remoteDocumentCache.getAllDocumentsMatchingQuery(query);
+        remoteDocumentCache.getAllDocumentsMatchingQuery(query, sinceReadTime);
 
     List<MutationBatch> matchingBatches = mutationQueue.getAllMutationBatchesAffectingQuery(query);
 

firebase-firestore/src/main/java/com/google/firebase/firestore/local/MemoryRemoteDocumentCache.java

@@ -48,6 +48,9 @@ final class MemoryRemoteDocumentCache implements RemoteDocumentCache {
 
   @Override
   public void add(MaybeDocument document, SnapshotVersion readTime) {
+    hardAssert(
+        !readTime.equals(SnapshotVersion.NONE),
+        "Cannot add document to the RemoteDocumentCache with a read time of zero");
     docs = docs.insert(document.getKey(), new Pair<>(document, readTime));
 
     persistence.getIndexManager().addToCollectionParentIndex(document.getKey().getPath().popLast());
@@ -82,7 +85,8 @@ public Map<DocumentKey, MaybeDocument> getAll(Iterable<DocumentKey> keys) {
   }
 
   @Override
-  public ImmutableSortedMap<DocumentKey, Document> getAllDocumentsMatchingQuery(Query query) {
+  public ImmutableSortedMap<DocumentKey, Document> getAllDocumentsMatchingQuery(
+      Query query, SnapshotVersion sinceReadTime) {
     hardAssert(
         !query.isCollectionGroupQuery(),
         "CollectionGroup queries should be handled in LocalDocumentsView");
@@ -112,6 +116,11 @@ public ImmutableSortedMap<DocumentKey, Document> getAllDocumentsMatchingQuery(Qu
         continue;
       }
 
+      SnapshotVersion readTime = entry.getValue().second;
+      if (readTime.compareTo(sinceReadTime) <= 0) {
+        continue;
+      }
+
       Document doc = (Document) maybeDoc;
       if (query.matches(doc)) {
         result = result.insert(doc.getKey(), doc);

firebase-firestore/src/main/java/com/google/firebase/firestore/local/RemoteDocumentCache.java

@@ -75,7 +75,10 @@ interface RemoteDocumentCache {
    * <p>Cached NoDocument entries have no bearing on query results.
    *
    * @param query The query to match documents against.
+   * @param sinceReadTime If not set to SnapshotVersion.MIN, return only documents that have been
+   *     read since this snapshot version (exclusive).
    * @return The set of matching documents.
    */
-  ImmutableSortedMap<DocumentKey, Document> getAllDocumentsMatchingQuery(Query query);
+  ImmutableSortedMap<DocumentKey, Document> getAllDocumentsMatchingQuery(
+      Query query, SnapshotVersion sinceReadTime);
 }

firebase-firestore/src/main/java/com/google/firebase/firestore/local/SQLitePersistence.java

@@ -77,7 +77,7 @@ public static String databaseName(String persistenceKey, DatabaseId databaseId)
     }
   }
 
-  private final OpenHelper opener;
+  private final SQLiteOpenHelper opener;
   private final LocalSerializer serializer;
   private final StatsCollector statsCollector;
   private final SQLiteQueryCache queryCache;
@@ -125,8 +125,19 @@ public SQLitePersistence(
       LocalSerializer serializer,
       StatsCollector statsCollector,
       LruGarbageCollector.Params params) {
-    String databaseName = databaseName(persistenceKey, databaseId);
-    this.opener = new OpenHelper(context, databaseName);
+    this(
+        serializer,
+        statsCollector,
+        params,
+        new OpenHelper(context, databaseName(persistenceKey, databaseId)));
+  }
+
+  public SQLitePersistence(
+      LocalSerializer serializer,
+      StatsCollector statsCollector,
+      LruGarbageCollector.Params params,
+      SQLiteOpenHelper openHelper) {
+    this.opener = openHelper;
     this.serializer = serializer;
     this.statsCollector = statsCollector;
     this.queryCache = new SQLiteQueryCache(this, this.serializer);

firebase-firestore/src/main/java/com/google/firebase/firestore/local/SQLiteRemoteDocumentCache.java

@@ -52,6 +52,10 @@ final class SQLiteRemoteDocumentCache implements RemoteDocumentCache {
 
   @Override
   public void add(MaybeDocument maybeDocument, SnapshotVersion readTime) {
+    hardAssert(
+        !readTime.equals(SnapshotVersion.NONE),
+        "Cannot add document to the RemoteDocumentCache with a read time of zero");
+
     String path = pathForKey(maybeDocument.getKey());
     Timestamp timestamp = readTime.getTimestamp();
     MessageLite message = serializer.encodeMaybeDocument(maybeDocument);
@@ -131,7 +135,8 @@ public Map<DocumentKey, MaybeDocument> getAll(Iterable<DocumentKey> documentKeys
   }
 
   @Override
-  public ImmutableSortedMap<DocumentKey, Document> getAllDocumentsMatchingQuery(Query query) {
+  public ImmutableSortedMap<DocumentKey, Document> getAllDocumentsMatchingQuery(
+      Query query, SnapshotVersion sinceReadTime) {
     hardAssert(
         !query.isCollectionGroupQuery(),
         "CollectionGroup queries should be handled in LocalDocumentsView");
@@ -142,6 +147,7 @@ public ImmutableSortedMap<DocumentKey, Document> getAllDocumentsMatchingQuery(Qu
 
     String prefixPath = EncodedPath.encode(prefix);
     String prefixSuccessorPath = EncodedPath.prefixSuccessor(prefixPath);
+    Timestamp readTime = sinceReadTime.getTimestamp();
 
     BackgroundQueue backgroundQueue = new BackgroundQueue();
 
@@ -150,8 +156,16 @@ public ImmutableSortedMap<DocumentKey, Document> getAllDocumentsMatchingQuery(Qu
             new ImmutableSortedMap[] {DocumentCollections.emptyDocumentMap()};
 
     int rowsProcessed =
-        db.query("SELECT path, contents FROM remote_documents WHERE path >= ? AND path < ?")
-            .binding(prefixPath, prefixSuccessorPath)
+        db.query(
+                "SELECT path, contents FROM remote_documents WHERE path >= ? AND path < ? "
+                    + "AND (read_time_seconds IS NULL OR read_time_seconds > ? "
+                    + "OR (read_time_seconds = ? AND read_time_nanos > ?))")
+            .binding(
+                prefixPath,
+                prefixSuccessorPath,
+                readTime.getSeconds(),
+                readTime.getSeconds(),
+                readTime.getNanoseconds())
             .forEach(
                 row -> {
                   // TODO: Actually implement a single-collection query

firebase-firestore/src/main/java/com/google/firebase/firestore/local/SQLiteSchema.java

@@ -46,9 +46,9 @@ class SQLiteSchema {
    * The version of the schema. Increase this by one for each migration added to runMigrations
    * below.
    *
-   * <p>TODO(index-free): The migration to schema version 9 doesn't backfill `update_time` as this
+   * <p>TODO(index-free): The migration to schema version 9 doesn't backfill `read_time` as this
    * requires rewriting the RemoteDocumentCache. For index-free queries to efficiently handle
-   * existing documents, we still need to populate update_time for all existing entries, drop the
+   * existing documents, we still need to populate read_time for all existing entries, drop the
    * RemoteDocumentCache or ask users to invoke `clearPersistence()` manually. If we decide to
    * backfill or drop the contents of the RemoteDocumentCache, we need to perform an additional
    * schema migration.

firebase-firestore/src/main/java/com/google/firebase/firestore/local/SimpleQueryEngine.java

@@ -19,6 +19,7 @@
 import com.google.firebase.firestore.model.Document;
 import com.google.firebase.firestore.model.DocumentKey;
 import com.google.firebase.firestore.model.MaybeDocument;
+import com.google.firebase.firestore.model.SnapshotVersion;
 
 /**
  * A naive implementation of QueryEngine that just loads all the documents in the queried collection
@@ -36,7 +37,7 @@ public SimpleQueryEngine(LocalDocumentsView localDocumentsView) {
   public ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingQuery(Query query) {
     // TODO: Once LocalDocumentsView provides a getCollectionDocuments() method, we
     // should call that here and then filter the results.
-    return localDocumentsView.getDocumentsMatchingQuery(query);
+    return localDocumentsView.getDocumentsMatchingQuery(query, SnapshotVersion.NONE);
   }
 
   @Override

firebase-firestore/src/test/java/com/google/firebase/firestore/local/LocalStoreTestCase.java

@@ -727,12 +727,12 @@ public void testCollectsGarbageAfterAcknowledgedMutation() {
     Query query = Query.atPath(ResourcePath.fromString("foo"));
     int targetId = allocateQuery(query);
     applyRemoteEvent(
-        updateRemoteEvent(doc("foo/bar", 0, map("foo", "old")), asList(targetId), emptyList()));
+        updateRemoteEvent(doc("foo/bar", 1, map("foo", "old")), asList(targetId), emptyList()));
     writeMutation(patchMutation("foo/bar", map("foo", "bar")));
     releaseQuery(query);
     writeMutation(setMutation("foo/bah", map("foo", "bah")));
     writeMutation(deleteMutation("foo/baz"));
-    assertContains(doc("foo/bar", 0, map("foo", "bar"), Document.DocumentState.LOCAL_MUTATIONS));
+    assertContains(doc("foo/bar", 1, map("foo", "bar"), Document.DocumentState.LOCAL_MUTATIONS));
     assertContains(doc("foo/bah", 0, map("foo", "bah"), Document.DocumentState.LOCAL_MUTATIONS));
     assertContains(deletedDoc("foo/baz", 0));
 
@@ -761,13 +761,13 @@ public void testCollectsGarbageAfterRejectedMutation() {
     Query query = Query.atPath(ResourcePath.fromString("foo"));
     int targetId = allocateQuery(query);
     applyRemoteEvent(
-        updateRemoteEvent(doc("foo/bar", 0, map("foo", "old")), asList(targetId), emptyList()));
+        updateRemoteEvent(doc("foo/bar", 1, map("foo", "old")), asList(targetId), emptyList()));
     writeMutation(patchMutation("foo/bar", map("foo", "bar")));
     // Release the query so that our target count goes back to 0 and we are considered up-to-date.
     releaseQuery(query);
     writeMutation(setMutation("foo/bah", map("foo", "bah")));
     writeMutation(deleteMutation("foo/baz"));
-    assertContains(doc("foo/bar", 0, map("foo", "bar"), Document.DocumentState.LOCAL_MUTATIONS));
+    assertContains(doc("foo/bar", 1, map("foo", "bar"), Document.DocumentState.LOCAL_MUTATIONS));
     assertContains(doc("foo/bah", 0, map("foo", "bah"), Document.DocumentState.LOCAL_MUTATIONS));
     assertContains(deletedDoc("foo/baz", 0));
 

firebase-firestore/src/test/java/com/google/firebase/firestore/local/RemoteDocumentCacheTestCase.java

@@ -177,19 +177,48 @@ public void testDocumentsMatchingQuery() {
 
     Query query = Query.atPath(path("b"));
     ImmutableSortedMap<DocumentKey, Document> results =
-        remoteDocumentCache.getAllDocumentsMatchingQuery(query);
+        remoteDocumentCache.getAllDocumentsMatchingQuery(query, SnapshotVersion.NONE);
     List<Document> expected = asList(doc("b/1", 42, docData), doc("b/2", 42, docData));
     assertEquals(expected, values(results));
   }
 
+  @Test
+  public void testDocumentsMatchingQuerySinceReadTime() {
+    Map<String, Object> docData = map("data", 2);
+    addTestDocumentAtPath("b/old", /* updateTime= */ 1, /* readTime= */ 11);
+    addTestDocumentAtPath("b/current", /* updateTime= */ 2, /*  readTime= = */ 12);
+    addTestDocumentAtPath("b/new", /* updateTime= */ 3, /*  readTime= = */ 13);
+
+    Query query = Query.atPath(path("b"));
+    ImmutableSortedMap<DocumentKey, Document> results =
+        remoteDocumentCache.getAllDocumentsMatchingQuery(query, version(12));
+    List<Document> expected = asList(doc("b/new", 3, docData));
+    assertEquals(expected, values(results));
+  }
+
+  @Test
+  public void testDocumentsMatchingUsesReadTimeNotUpdateTime() {
+    Map<String, Object> docData = map("data", 2);
+    addTestDocumentAtPath("b/old", /* updateTime= */ 1, /* readTime= */ 2);
+    addTestDocumentAtPath("b/new", /* updateTime= */ 2, /* readTime= */ 1);
+
+    Query query = Query.atPath(path("b"));
+    ImmutableSortedMap<DocumentKey, Document> results =
+        remoteDocumentCache.getAllDocumentsMatchingQuery(query, version(1));
+    List<Document> expected = asList(doc("b/old", 1, docData));
+    assertEquals(expected, values(results));
+  }
+
   private Document addTestDocumentAtPath(String path) {
-    Document doc = doc(path, 42, map("data", 2));
-    add(doc, version(42));
+    return addTestDocumentAtPath(path, 42, 42);
+  }
+
+  private Document addTestDocumentAtPath(String path, int updateTime, int readTime) {
+    Document doc = doc(path, updateTime, map("data", 2));
+    add(doc, version(readTime));
     return doc;
   }
 
-  // TODO(mrschmidt): Add a test uses different update and read times and verifies that we correctly
-  // filter by read time
   private void add(MaybeDocument doc, SnapshotVersion readTime) {
     persistence.runTransaction("add entry", () -> remoteDocumentCache.add(doc, readTime));
   }