fix(spanner): Fix the start timestamp > end timestamp issue for change

stream initial query.

Author: jiangzzhu

java-spanner/google-cloud-spanner-executor/src/main/java/com/google/cloud/executor/spanner/CloudClientExecutor.java

@@ -2462,14 +2462,19 @@ private Status executeExecuteChangeStreamQuery(
       // For initial partition query (no partition token) we simulate precision of the timestamp
       // in nanoseconds as that's closer inlined with the production client code.
 
-      String startTime =
-          timestampToString(
+      com.google.protobuf.Timestamp startTimestamp =
+          timestampFromMicrosOrNanos(
               !action.hasPartitionToken(), Timestamps.toMicros(action.getStartTime()));
+      String startTime = String.format("\"%s\"", Timestamps.toString(startTimestamp));
       String endTime = "null";
       if (action.hasEndTime()) {
-        endTime =
-            timestampToString(
+        com.google.protobuf.Timestamp endTimestamp =
+            timestampFromMicrosOrNanos(
                 !action.hasPartitionToken(), Timestamps.toMicros(action.getEndTime()));
+        // Guarantees that the end timestamp is not before the start timestamp.
+        endTimestamp =
+            Timestamps.compare(endTimestamp, startTimestamp) >= 0 ? endTimestamp : startTimestamp;
+        endTime = String.format("\"%s\"", Timestamps.toString(endTimestamp));
       }
       String heartbeat = "null";
       if (action.hasHeartbeatMilliseconds()) {
@@ -3948,6 +3953,20 @@ private <T extends Serializable> ByteString marshall(T object) throws IOExceptio
     return output.toByteString();
   }
 
+  /**
+   * Converts timestamp microseconds to Timestamp. If useNanosPrecision is set to true, it pads
+   * input timestamp with 3 random digits (less than 1 microsecond) treating it as timestamp
+   * nanoseconds. This simulates the production client code send Timestamp in nanoseconds precision.
+   */
+  private static com.google.protobuf.Timestamp timestampFromMicrosOrNanos(
+      boolean useNanosPrecision, long timestampInMicros) {
+    com.google.protobuf.Timestamp timestamp =
+        useNanosPrecision
+            ? Timestamps.fromNanos(timestampInMicros * 1000 + System.nanoTime() % 1000)
+            : Timestamps.fromMicros(timestampInMicros);
+    return timestamp;
+  }
+
   /** Build Timestamp from micros. */
   private Timestamp timestampFromMicros(long micros) {
     long seconds = TimeUnit.MICROSECONDS.toSeconds(micros);