@@ -453,7 +453,7 @@
 <className>com/google/cloud/spanner/TransactionContext</className>
 <method>com.google.api.core.ApiFuture executeUpdateAsync(com.google.cloud.spanner.Statement)</method>
 </difference>
-
+
 <!-- Support for CommitStats added -->
 <difference>
 <differenceType>7012</differenceType>
@@ -475,5 +475,16 @@
 <className>com/google/cloud/spanner/TransactionRunner</className>
 <method>com.google.cloud.spanner.CommitResponse getCommitResponse()</method>
 </difference>
-
+
+<!-- PITR -->
+<difference>
+<differenceType>7013</differenceType>
+<className>com/google/cloud/spanner/BackupInfo$Builder</className>
+<method>com.google.cloud.spanner.BackupInfo$Builder setVersionTime(com.google.cloud.Timestamp)</method>
+</difference>
+<difference>
+<differenceType>7012</differenceType>
+<className>com/google/cloud/spanner/DatabaseAdminClient</className>
+<method>com.google.api.gax.longrunning.OperationFuture createBackup(com.google.cloud.spanner.Backup)</method>
+</difference>
 </differences>

@@ -65,7 +65,7 @@ public OperationFuture<Backup, CreateBackupMetadata> create() {
 getExpireTime() != null, "Cannot create a backup without an expire time");
 Preconditions.checkState(
 getDatabase() != null, "Cannot create a backup without a source database");
-return dbClient.createBackup(instance(), backup(), sourceDatabase(), getExpireTime());
+return dbClient.createBackup(this);
 }
 
 /**
@@ -182,6 +182,7 @@ static Backup fromProto(
 .setState(fromProtoState(proto.getState()))
 .setSize(proto.getSizeBytes())
 .setExpireTime(Timestamp.fromProto(proto.getExpireTime()))
+.setVersionTime(Timestamp.fromProto(proto.getVersionTime()))
 .setDatabase(DatabaseId.of(proto.getDatabase()))
 .setProto(proto)
 .build();

@@ -18,6 +18,7 @@
 
 import com.google.api.client.util.Preconditions;
 import com.google.cloud.Timestamp;
+import com.google.spanner.admin.database.v1.Database;
 import java.util.Objects;
 import javax.annotation.Nullable;
 
@@ -40,6 +41,17 @@ public abstract static class Builder {
 */
 public abstract Builder setExpireTime(Timestamp expireTime);
 
+/**
+* Optional for creating a new backup.
+*
+* <p>Specifies the timestamp to have an externally consistent copy of the database. If no
+* version time is specified, it will be automatically set to the backup create time.
+*
+* <p>The version time can be as far in the past as specified by the database earliest version
+* time (see {@link Database#getEarliestVersionTime()}).
+*/
+public abstract Builder setVersionTime(Timestamp versionTime);
+
 /**
 * Required for creating a new backup.
 *
@@ -55,6 +67,7 @@ abstract static class BuilderImpl extends Builder {
 protected final BackupId id;
 private State state = State.UNSPECIFIED;
 private Timestamp expireTime;
+private Timestamp versionTime;
 private DatabaseId database;
 private long size;
 private com.google.spanner.admin.database.v1.Backup proto;
@@ -67,6 +80,7 @@ abstract static class BuilderImpl extends Builder {
 this.id = other.id;
 this.state = other.state;
 this.expireTime = other.expireTime;
+this.versionTime = other.versionTime;
 this.database = other.database;
 this.size = other.size;
 this.proto = other.proto;
@@ -84,6 +98,12 @@ public Builder setExpireTime(Timestamp expireTime) {
 return this;
 }
 
+@Override
+public Builder setVersionTime(Timestamp versionTime) {
+this.versionTime = versionTime;
+return this;
+}
+
 @Override
 public Builder setDatabase(DatabaseId database) {
 Preconditions.checkArgument(
@@ -119,6 +139,7 @@ public enum State {
 private final BackupId id;
 private final State state;
 private final Timestamp expireTime;
+private final Timestamp versionTime;
 private final DatabaseId database;
 private final long size;
 private final com.google.spanner.admin.database.v1.Backup proto;
@@ -128,6 +149,7 @@ public enum State {
 this.state = builder.state;
 this.size = builder.size;
 this.expireTime = builder.expireTime;
+this.versionTime = builder.versionTime;
 this.database = builder.database;
 this.proto = builder.proto;
 }
@@ -157,6 +179,11 @@ public Timestamp getExpireTime() {
 return expireTime;
 }
 
+/** Returns the version time of the backup. */
+public Timestamp getVersionTime() {
+return versionTime;
+}
+
 /** Returns the id of the database that was used to create the backup. */
 public DatabaseId getDatabase() {
 return database;
@@ -180,17 +207,19 @@ public boolean equals(Object o) {
 && state == that.state
 && size == that.size
 && Objects.equals(expireTime, that.expireTime)
+&& Objects.equals(versionTime, that.versionTime)
 && Objects.equals(database, that.database);
 }
 
 @Override
 public int hashCode() {
-return Objects.hash(id, state, size, expireTime, database);
+return Objects.hash(id, state, size, expireTime, versionTime, database);
 }
 
 @Override
 public String toString() {
 return String.format(
-"Backup[%s, %s, %d, %s, %s]", id.getName(), state, size, expireTime, database);
+"Backup[%s, %s, %d, %s, %s, %s]",
+id.getName(), state, size, expireTime, versionTime, database);
 }
 }

@@ -118,8 +118,14 @@ public OperationFuture<Backup, CreateBackupMetadata> backup(Backup backup) {
 Preconditions.checkArgument(
 backup.getInstanceId().equals(getId().getInstanceId()),
 "The instance of the backup must be equal to the instance of this database.");
+
 return dbClient.createBackup(
-instance(), backup.getId().getBackup(), database(), backup.getExpireTime());
+dbClient
+.newBackupBuilder(backup.getId())
+.setDatabase(getId())
+.setExpireTime(backup.getExpireTime())
+.setVersionTime(backup.getVersionTime())
+.build());
 }
 
 /**
@@ -177,6 +183,8 @@ static Database fromProto(
 .setState(fromProtoState(proto.getState()))
 .setCreateTime(Timestamp.fromProto(proto.getCreateTime()))
 .setRestoreInfo(RestoreInfo.fromProtoOrNullIfDefaultInstance(proto.getRestoreInfo()))
+.setVersionRetentionPeriod(proto.getVersionRetentionPeriod())
+.setEarliestVersionTime(Timestamp.fromProto(proto.getEarliestVersionTime()))
 .setProto(proto)
 .build();
 }

@@ -101,6 +101,32 @@ OperationFuture<Backup, CreateBackupMetadata> createBackup(
 String sourceInstanceId, String backupId, String databaseId, Timestamp expireTime)
 throws SpannerException;
 
+/**
+* Creates a new backup from a database in a Cloud Spanner instance.
+*
+* <p>Example to create a backup.
+*
+* <pre>{@code
+* BackupId backupId = BackupId.of("project", "instance", "backup-id");
+* DatabaseId databaseId = DatabaseId.of("project", "instance", "database-id");
+* Timestamp expireTime = Timestamp.ofTimeMicroseconds(expireTimeMicros);
+* Timestamp versionTime = Timestamp.ofTimeMicroseconds(versionTimeMicros);
+*
+* Backup backupToCreate = dbAdminClient
+* .newBackupBuilder(backupId)
+* .setDatabase(databaseId)
+* .setExpireTime(expireTime)
+* .setVersionTime(versionTime)
+* .build();
+*
+* OperationFuture<Backup, CreateBackupMetadata> op = dbAdminClient.createBackup(backupToCreate);
+* Backup createdBackup = op.get();
+* }</pre>
+*
+* @param backup the backup to be created
+*/
+OperationFuture<Backup, CreateBackupMetadata> createBackup(Backup backup) throws SpannerException;
+
 /**
 * Restore a database from a backup. The database that is restored will be created and may not
 * already exist.

@@ -114,16 +114,31 @@ public Database apply(Exception e) {
 public OperationFuture<Backup, CreateBackupMetadata> createBackup(
 String instanceId, String backupId, String databaseId, Timestamp expireTime)
 throws SpannerException {
-com.google.spanner.admin.database.v1.Backup backup =
+final Backup backup =
+newBackupBuilder(BackupId.of(projectId, instanceId, backupId))
+.setDatabase(DatabaseId.of(projectId, instanceId, databaseId))
+.setExpireTime(expireTime)
+.build();
+return createBackup(backup);
+}
+
+@Override
+public OperationFuture<Backup, CreateBackupMetadata> createBackup(final Backup backup) {
+final String instanceId = backup.getInstanceId().getInstance();
+final String databaseId = backup.getDatabase().getDatabase();
+final String backupId = backup.getId().getBackup();
+final com.google.spanner.admin.database.v1.Backup.Builder backupBuilder =
 com.google.spanner.admin.database.v1.Backup.newBuilder()
 .setDatabase(getDatabaseName(instanceId, databaseId))
-.setExpireTime(expireTime.toProto())
-.build();
-String instanceName = getInstanceName(instanceId);
-OperationFuture<com.google.spanner.admin.database.v1.Backup, CreateBackupMetadata>
-rawOperationFuture = rpc.createBackup(instanceName, backupId, backup);
+.setExpireTime(backup.getExpireTime().toProto());
+if (backup.getVersionTime() != null) {
+backupBuilder.setVersionTime(backup.getVersionTime().toProto());
+}
+final String instanceName = getInstanceName(instanceId);
+final OperationFuture<com.google.spanner.admin.database.v1.Backup, CreateBackupMetadata>
+rawOperationFuture = rpc.createBackup(instanceName, backupId, backupBuilder.build());
 
-return new OperationFutureImpl<Backup, CreateBackupMetadata>(
+return new OperationFutureImpl<>(
 rawOperationFuture.getPollingFuture(),
 rawOperationFuture.getInitialFuture(),
 new ApiFunction<OperationSnapshot, Backup>() {
@@ -137,6 +152,7 @@ public Backup apply(OperationSnapshot snapshot) {
 com.google.spanner.admin.database.v1.Backup.newBuilder(proto)
 .setName(proto.getName())
 .setExpireTime(proto.getExpireTime())
+.setVersionTime(proto.getVersionTime())
 .setState(proto.getState())
 .build(),
 DatabaseAdminClientImpl.this);

@@ -30,6 +30,10 @@ public abstract static class Builder {
 
 abstract Builder setRestoreInfo(RestoreInfo restoreInfo);
 
+abstract Builder setVersionRetentionPeriod(String versionRetentionPeriod);
+
+abstract Builder setEarliestVersionTime(Timestamp earliestVersionTime);
+
 abstract Builder setProto(com.google.spanner.admin.database.v1.Database proto);
 
 /** Builds the database from this builder. */
@@ -41,6 +45,8 @@ abstract static class BuilderImpl extends Builder {
 private State state = State.UNSPECIFIED;
 private Timestamp createTime;
 private RestoreInfo restoreInfo;
+private String versionRetentionPeriod;
+private Timestamp earliestVersionTime;
 private com.google.spanner.admin.database.v1.Database proto;
 
 BuilderImpl(DatabaseId id) {
@@ -52,6 +58,8 @@ abstract static class BuilderImpl extends Builder {
 this.state = other.state;
 this.createTime = other.createTime;
 this.restoreInfo = other.restoreInfo;
+this.versionRetentionPeriod = other.versionRetentionPeriod;
+this.earliestVersionTime = other.earliestVersionTime;
 this.proto = other.proto;
 }
 
@@ -73,6 +81,18 @@ Builder setRestoreInfo(@Nullable RestoreInfo restoreInfo) {
 return this;
 }
 
+@Override
+Builder setVersionRetentionPeriod(String versionRetentionPeriod) {
+this.versionRetentionPeriod = versionRetentionPeriod;
+return this;
+}
+
+@Override
+Builder setEarliestVersionTime(Timestamp earliestVersionTime) {
+this.earliestVersionTime = earliestVersionTime;
+return this;
+}
+
 @Override
 Builder setProto(@Nullable com.google.spanner.admin.database.v1.Database proto) {
 this.proto = proto;
@@ -96,13 +116,17 @@ public enum State {
 private final State state;
 private final Timestamp createTime;
 private final RestoreInfo restoreInfo;
+private final String versionRetentionPeriod;
+private final Timestamp earliestVersionTime;
 private final com.google.spanner.admin.database.v1.Database proto;
 
 public DatabaseInfo(DatabaseId id, State state) {
 this.id = id;
 this.state = state;
 this.createTime = null;
 this.restoreInfo = null;
+this.versionRetentionPeriod = null;
+this.earliestVersionTime = null;
 this.proto = null;
 }
 
@@ -111,6 +135,8 @@ public DatabaseInfo(DatabaseId id, State state) {
 this.state = builder.state;
 this.createTime = builder.createTime;
 this.restoreInfo = builder.restoreInfo;
+this.versionRetentionPeriod = builder.versionRetentionPeriod;
+this.earliestVersionTime = builder.earliestVersionTime;
 this.proto = builder.proto;
 }
 
@@ -129,6 +155,23 @@ public Timestamp getCreateTime() {
 return createTime;
 }
 
+/**
+* Returns the version retention period of the database. This is the period for which Cloud
+* Spanner retains all versions of data for the database. For instance, if set to 3 days, Cloud
+* Spanner will retain data versions that are up to 3 days old.
+*/
+public String getVersionRetentionPeriod() {
+return versionRetentionPeriod;
+}
+
+/**
+* Returns the earliest version time of the database. This is the oldest timestamp that can be
+* used to read old versions of the data.
+*/
+public Timestamp getEarliestVersionTime() {
+return earliestVersionTime;
+}
+
 /**
 * Returns the {@link RestoreInfo} of the database if any is available, or <code>null</code> if no
 * {@link RestoreInfo} is available for this database.
@@ -154,16 +197,21 @@ public boolean equals(Object o) {
 return id.equals(that.id)
 && state == that.state
 && Objects.equals(createTime, that.createTime)
-&& Objects.equals(restoreInfo, that.restoreInfo);
+&& Objects.equals(restoreInfo, that.restoreInfo)
+&& Objects.equals(versionRetentionPeriod, that.versionRetentionPeriod)
+&& Objects.equals(earliestVersionTime, that.earliestVersionTime);
 }
 
 @Override
 public int hashCode() {
-return Objects.hash(id, state, createTime, restoreInfo);
+return Objects.hash(
+id, state, createTime, restoreInfo, versionRetentionPeriod, earliestVersionTime);
 }
 
 @Override
 public String toString() {
-return String.format("Database[%s, %s, %s, %s]", id.getName(), state, createTime, restoreInfo);
+return String.format(
+"Database[%s, %s, %s, %s, %s, %s]",
+id.getName(), state, createTime, restoreInfo, versionRetentionPeriod, earliestVersionTime);
 }
 }