feat: add ReadOption.ReadTime to support timestamp reads. (#712)

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
- [ ] Make sure to open an issue as a [bug/issue](https://github.com/googleapis/java-datastore/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [ ] Ensure the tests and linter pass
- [ ] Code coverage does not decrease (if any source code was changed)
- [ ] Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> ☕️

If you write sample code, please follow the [samples format](
https://github.com/GoogleCloudPlatform/java-docs-samples/blob/main/SAMPLE_FORMAT.md).

Author: yixiaoshen

google-cloud-datastore/src/main/java/com/google/cloud/datastore/DatastoreImpl.java

@@ -23,6 +23,7 @@
 import com.google.cloud.RetryHelper.RetryHelperException;
 import com.google.cloud.ServiceOptions;
 import com.google.cloud.datastore.ReadOption.EventualConsistency;
+import com.google.cloud.datastore.ReadOption.ReadTime;
 import com.google.cloud.datastore.spi.v1.DatastoreRpc;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Preconditions;
@@ -338,12 +339,29 @@ public Iterator<Entity> get(Iterable<Key> keys, ReadOption... options) {
 
   private static com.google.datastore.v1.ReadOptions toReadOptionsPb(ReadOption... options) {
     com.google.datastore.v1.ReadOptions readOptionsPb = null;
-    if (options != null
-        && ReadOption.asImmutableMap(options).containsKey(EventualConsistency.class)) {
-      readOptionsPb =
-          com.google.datastore.v1.ReadOptions.newBuilder()
-              .setReadConsistency(ReadConsistency.EVENTUAL)
-              .build();
+    if (options != null) {
+      Map<Class<? extends ReadOption>, ReadOption> optionsByType =
+          ReadOption.asImmutableMap(options);
+
+      if (optionsByType.containsKey(EventualConsistency.class)
+          && optionsByType.containsKey(ReadTime.class)) {
+        throw DatastoreException.throwInvalidRequest(
+            "Can not use eventual consistency read with read time.");
+      }
+
+      if (optionsByType.containsKey(EventualConsistency.class)) {
+        readOptionsPb =
+            com.google.datastore.v1.ReadOptions.newBuilder()
+                .setReadConsistency(ReadConsistency.EVENTUAL)
+                .build();
+      }
+
+      if (optionsByType.containsKey(ReadTime.class)) {
+        readOptionsPb =
+            com.google.datastore.v1.ReadOptions.newBuilder()
+                .setReadTime(((ReadTime) optionsByType.get(ReadTime.class)).time().toProto())
+                .build();
+      }
     }
     return readOptionsPb;
   }

google-cloud-datastore/src/main/java/com/google/cloud/datastore/ReadOption.java

@@ -16,6 +16,8 @@
 
 package com.google.cloud.datastore;
 
+import com.google.api.core.BetaApi;
+import com.google.cloud.Timestamp;
 import com.google.common.collect.ImmutableMap;
 import java.io.Serializable;
 import java.util.Map;
@@ -47,6 +49,25 @@ public boolean isEventual() {
     }
   }
 
+  /**
+   * Reads entities as they were at the given time. This may not be older than 270 seconds. This
+   * value is only supported for Cloud Firestore in Datastore mode.
+   */
+  public static final class ReadTime extends ReadOption {
+
+    private static final long serialVersionUID = -6780321449114616067L;
+
+    private final Timestamp time;
+
+    private ReadTime(Timestamp time) {
+      this.time = time;
+    }
+
+    public Timestamp time() {
+      return time;
+    }
+  }
+
   private ReadOption() {}
 
   /**
@@ -57,6 +78,16 @@ public static EventualConsistency eventualConsistency() {
     return new EventualConsistency(true);
   }
 
+  /**
+   * Returns a {@code ReadOption} that specifies read time, allowing Datastore to return results
+   * from lookups and queries at a particular timestamp. This feature is currently in private
+   * preview.
+   */
+  @BetaApi
+  public static ReadTime readTime(Timestamp time) {
+    return new ReadTime(time);
+  }
+
   static Map<Class<? extends ReadOption>, ReadOption> asImmutableMap(ReadOption... options) {
     ImmutableMap.Builder<Class<? extends ReadOption>, ReadOption> builder = ImmutableMap.builder();
     for (ReadOption option : options) {

google-cloud-datastore/src/test/java/com/google/cloud/datastore/DatastoreTest.java

@@ -789,6 +789,26 @@ public void testEventualConsistencyQuery() {
     EasyMock.verify(rpcFactoryMock, rpcMock);
   }
 
+  @Test
+  public void testReadTimeQuery() {
+    Timestamp timestamp = Timestamp.now();
+    ReadOptions readOption = ReadOptions.newBuilder().setReadTime(timestamp.toProto()).build();
+    com.google.datastore.v1.GqlQuery query =
+        com.google.datastore.v1.GqlQuery.newBuilder().setQueryString("FROM * SELECT *").build();
+    RunQueryRequest.Builder expectedRequest =
+        RunQueryRequest.newBuilder()
+            .setReadOptions(readOption)
+            .setGqlQuery(query)
+            .setPartitionId(PartitionId.newBuilder().setProjectId(PROJECT_ID).build());
+    EasyMock.expect(rpcMock.runQuery(expectedRequest.build()))
+        .andReturn(RunQueryResponse.newBuilder().build());
+    EasyMock.replay(rpcFactoryMock, rpcMock);
+    Datastore datastore = rpcMockOptions.getService();
+    datastore.run(
+        Query.newGqlQueryBuilder("FROM * SELECT *").build(), ReadOption.readTime(timestamp));
+    EasyMock.verify(rpcFactoryMock, rpcMock);
+  }
+
   @Test
   public void testToUrlSafe() {
     byte[][] invalidUtf8 =
@@ -921,6 +941,34 @@ public void testLookupEventualConsistency() {
     EasyMock.verify(rpcFactoryMock, rpcMock);
   }
 
+  @Test
+  public void testLookupReadTime() {
+    Timestamp timestamp = Timestamp.now();
+    ReadOptions readOption = ReadOptions.newBuilder().setReadTime(timestamp.toProto()).build();
+    com.google.datastore.v1.Key key =
+        com.google.datastore.v1.Key.newBuilder()
+            .setPartitionId(PartitionId.newBuilder().setProjectId(PROJECT_ID).build())
+            .addPath(
+                com.google.datastore.v1.Key.PathElement.newBuilder()
+                    .setKind("kind1")
+                    .setName("name")
+                    .build())
+            .build();
+    LookupRequest lookupRequest =
+        LookupRequest.newBuilder().setReadOptions(readOption).addKeys(key).build();
+    EasyMock.expect(rpcMock.lookup(lookupRequest))
+        .andReturn(LookupResponse.newBuilder().build())
+        .times(3);
+    EasyMock.replay(rpcFactoryMock, rpcMock);
+    com.google.cloud.datastore.Datastore datastore = rpcMockOptions.getService();
+    datastore.get(KEY1, com.google.cloud.datastore.ReadOption.readTime(timestamp));
+    datastore.get(
+        ImmutableList.of(KEY1), com.google.cloud.datastore.ReadOption.readTime(timestamp));
+    datastore.fetch(
+        ImmutableList.of(KEY1), com.google.cloud.datastore.ReadOption.readTime(timestamp));
+    EasyMock.verify(rpcFactoryMock, rpcMock);
+  }
+
   @Test
   public void testGetArrayNoDeferredResults() {
     datastore.put(ENTITY3);

google-cloud-datastore/src/test/java/com/google/cloud/datastore/it/ITDatastoreTest.java

@@ -50,6 +50,7 @@
 import com.google.cloud.datastore.Query;
 import com.google.cloud.datastore.Query.ResultType;
 import com.google.cloud.datastore.QueryResults;
+import com.google.cloud.datastore.ReadOption;
 import com.google.cloud.datastore.StringValue;
 import com.google.cloud.datastore.StructuredQuery;
 import com.google.cloud.datastore.StructuredQuery.OrderBy;
@@ -647,6 +648,31 @@ public void testGet() {
     assertFalse(entity.contains("bla"));
   }
 
+  @Test
+  public void testGetWithReadTime() throws InterruptedException {
+    Key key = Key.newBuilder(PROJECT_ID, "new_kind", "name").setNamespace(NAMESPACE).build();
+
+    try {
+      DATASTORE.put(Entity.newBuilder(key).set("str", "old_str_value").build());
+
+      Thread.sleep(1000);
+      Timestamp now = Timestamp.now();
+      Thread.sleep(1000);
+
+      DATASTORE.put(Entity.newBuilder(key).set("str", "new_str_value").build());
+
+      Entity entity = DATASTORE.get(key);
+      StringValue value1 = entity.getValue("str");
+      assertEquals(StringValue.of("new_str_value"), value1);
+
+      entity = DATASTORE.get(key, ReadOption.readTime(now));
+      value1 = entity.getValue("str");
+      assertEquals(StringValue.of("old_str_value"), value1);
+    } finally {
+      DATASTORE.delete(key);
+    }
+  }
+
   @Test
   public void testGetArrayNoDeferredResults() {
     DATASTORE.put(ENTITY3);
@@ -920,4 +946,48 @@ public void testQueryWithStartCursor() {
     assertEquals(cursor2, cursor1);
     DATASTORE.delete(entity1.getKey(), entity2.getKey(), entity3.getKey());
   }
+
+  @Test
+  public void testQueryWithReadTime() throws InterruptedException {
+    Entity entity1 =
+        Entity.newBuilder(
+                Key.newBuilder(PROJECT_ID, "new_kind", "name-01").setNamespace(NAMESPACE).build())
+            .build();
+    Entity entity2 =
+        Entity.newBuilder(
+                Key.newBuilder(PROJECT_ID, "new_kind", "name-02").setNamespace(NAMESPACE).build())
+            .build();
+    Entity entity3 =
+        Entity.newBuilder(
+                Key.newBuilder(PROJECT_ID, "new_kind", "name-03").setNamespace(NAMESPACE).build())
+            .build();
+
+    DATASTORE.put(entity1, entity2);
+    Thread.sleep(1000);
+    Timestamp now = Timestamp.now();
+    Thread.sleep(1000);
+    DATASTORE.put(entity3);
+
+    try {
+      Query<Entity> query = Query.newEntityQueryBuilder().setKind("new_kind").build();
+
+      QueryResults<Entity> withoutReadTime = DATASTORE.run(query);
+      assertTrue(withoutReadTime.hasNext());
+      assertEquals(entity1, withoutReadTime.next());
+      assertTrue(withoutReadTime.hasNext());
+      assertEquals(entity2, withoutReadTime.next());
+      assertTrue(withoutReadTime.hasNext());
+      assertEquals(entity3, withoutReadTime.next());
+      assertFalse(withoutReadTime.hasNext());
+
+      QueryResults<Entity> withReadTime = DATASTORE.run(query, ReadOption.readTime(now));
+      assertTrue(withReadTime.hasNext());
+      assertEquals(entity1, withReadTime.next());
+      assertTrue(withReadTime.hasNext());
+      assertEquals(entity2, withReadTime.next());
+      assertFalse(withReadTime.hasNext());
+    } finally {
+      DATASTORE.delete(entity1.getKey(), entity2.getKey(), entity3.getKey());
+    }
+  }
 }