feat(spanner): add support for Proto Columns in Connection API

README.md

@@ -57,13 +57,13 @@ implementation 'com.google.cloud:google-cloud-spanner'
 If you are using Gradle without BOM, add this to your dependencies:
 
 ```Groovy
-implementation 'com.google.cloud:google-cloud-spanner:6.66.0'
+implementation 'com.google.cloud:google-cloud-spanner:6.67.0'
 ```
 
 If you are using SBT, add this to your dependencies:
 
 ```Scala
-libraryDependencies += "com.google.cloud" % "google-cloud-spanner" % "6.66.0"
+libraryDependencies += "com.google.cloud" % "google-cloud-spanner" % "6.67.0"
 ```
 <!-- {x-version-update-end} -->
 
@@ -671,7 +671,7 @@ Java is a registered trademark of Oracle and/or its affiliates.
 [kokoro-badge-link-5]: http://storage.googleapis.com/cloud-devrel-public/java/badges/java-spanner/java11.html
 [stability-image]: https://img.shields.io/badge/stability-stable-green
 [maven-version-image]: https://img.shields.io/maven-central/v/com.google.cloud/google-cloud-spanner.svg
-[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-spanner/6.66.0
+[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-spanner/6.67.0
 [authentication]: https://github.com/googleapis/google-cloud-java#authentication
 [auth-scopes]: https://developers.google.com/identity/protocols/oauth2/scopes
 [predefined-iam-roles]: https://cloud.google.com/iam/docs/understanding-roles#predefined_roles

google-cloud-spanner/clirr-ignored-differences.xml

@@ -337,6 +337,12 @@
     <className>com/google/cloud/spanner/spi/v1/GapicSpannerRpc</className>
     <method>com.google.cloud.spanner.spi.v1.SpannerRpc$StreamingCall read(com.google.spanner.v1.ReadRequest, com.google.cloud.spanner.spi.v1.SpannerRpc$ResultStreamConsumer, java.util.Map)</method>
   </difference>
+  <difference>
+    <differenceType>7005</differenceType>
+    <className>com/google/cloud/spanner/spi/v1/GapicSpannerRpc</className>
+    <method>com.google.api.gax.longrunning.OperationFuture updateDatabaseDdl(java.lang.String, java.lang.Iterable, java.lang.String)</method>
+    <to>com.google.api.gax.longrunning.OperationFuture updateDatabaseDdl(com.google.cloud.spanner.Database, java.lang.Iterable, java.lang.String)</to>
+  </difference>
   <difference>
     <differenceType>7005</differenceType>
     <className>com/google/cloud/spanner/spi/v1/GapicSpannerRpc</className>
@@ -387,6 +393,12 @@
     <className>com/google/cloud/spanner/spi/v1/SpannerRpc</className>
     <method>com.google.cloud.spanner.spi.v1.SpannerRpc$StreamingCall read(com.google.spanner.v1.ReadRequest, com.google.cloud.spanner.spi.v1.SpannerRpc$ResultStreamConsumer, java.util.Map)</method>
   </difference>
+  <difference>
+    <differenceType>7005</differenceType>
+    <className>com/google/cloud/spanner/spi/v1/SpannerRpc</className>
+    <method>com.google.api.gax.longrunning.OperationFuture updateDatabaseDdl(java.lang.String, java.lang.Iterable, java.lang.String)</method>
+    <to>com.google.api.gax.longrunning.OperationFuture updateDatabaseDdl(com.google.cloud.spanner.Database, java.lang.Iterable, java.lang.String)</to>
+  </difference>
   <difference>
     <differenceType>7005</differenceType>
     <className>com/google/cloud/spanner/spi/v1/SpannerRpc</className>
@@ -656,7 +668,7 @@
     <className>com/google/cloud/spanner/connection/Connection</className>
     <method>com.google.cloud.spanner.Spanner getSpanner()</method>
   </difference>
-  
+
   <!-- Add DdlInTransactionMode -->
   <difference>
     <differenceType>7012</differenceType>
@@ -668,7 +680,7 @@
     <className>com/google/cloud/spanner/connection/Connection</className>
     <method>com.google.cloud.spanner.connection.DdlInTransactionMode getDdlInTransactionMode()</method>
   </difference>
-  
+
   <!-- Added extended tracing option -->
   <difference>
     <differenceType>7012</differenceType>
@@ -688,4 +700,16 @@
     <method>void setExcludeTxnFromChangeStreams(boolean)</method>
   </difference>
 
+  <!-- Added Proto descriptors for proto columns -->
+  <difference>
+    <differenceType>7012</differenceType>
+    <className>com/google/cloud/spanner/connection/Connection</className>
+    <method>byte[] getProtoDescriptors()</method>
+  </difference>
+  <difference>
+    <differenceType>7012</differenceType>
+    <className>com/google/cloud/spanner/connection/Connection</className>
+    <method>void setProtoDescriptors(byte[])</method>
+  </difference>
+
 </differences>

google-cloud-spanner/src/main/java/com/google/cloud/spanner/Type.java

@@ -147,7 +147,7 @@ public static Type pgOid() {
   /**
    * To get the descriptor for the {@code PROTO} type.
    *
-   * @param protoTypeFqn Proto fully qualified name (ex: "spanner.examples.music.SingerInfo").
+   * @param protoTypeFqn Proto fully qualified name (ex: "examples.spanner.music.SingerInfo").
    */
   public static Type proto(String protoTypeFqn) {
     return new Type(Code.PROTO, protoTypeFqn);
@@ -156,7 +156,7 @@ public static Type proto(String protoTypeFqn) {
   /**
    * To get the descriptor for the {@code ENUM} type.
    *
-   * @param protoTypeFqn Proto ENUM fully qualified name (ex: "spanner.examples.music.Genre")
+   * @param protoTypeFqn Proto ENUM fully qualified name (ex: "examples.spanner.music.Genre")
    */
   public static Type protoEnum(String protoTypeFqn) {
     return new Type(Code.ENUM, protoTypeFqn);

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ClientSideStatementValueConverters.java

@@ -26,10 +26,12 @@
 import com.google.cloud.spanner.connection.PgTransactionMode.IsolationLevel;
 import com.google.common.base.Function;
 import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
 import com.google.protobuf.Duration;
 import com.google.protobuf.util.Durations;
 import com.google.spanner.v1.DirectedReadOptions;
 import com.google.spanner.v1.RequestOptions.Priority;
+import java.util.Base64;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.Locale;
@@ -563,4 +565,46 @@ public String convert(String value) {
       return value.substring(7).trim();
     }
   }
+
+  /** Converter for converting Base64 encoded string to byte[] */
+  static class ProtoDescriptorsConverter implements ClientSideStatementValueConverter<byte[]> {
+
+    public ProtoDescriptorsConverter(String allowedValues) {}
+
+    @Override
+    public Class<byte[]> getParameterClass() {
+      return byte[].class;
+    }
+
+    @Override
+    public byte[] convert(String value) {
+      if (value == null || value.length() == 0 || value.equalsIgnoreCase("null")) {
+        return null;
+      }
+      try {
+        return Base64.getDecoder().decode(value);
+      } catch (IllegalArgumentException e) {
+        return null;
+      }
+    }
+  }
+
+  /** Converter for converting String that take in file path as input to String */
+  static class ProtoDescriptorsFileConverter implements ClientSideStatementValueConverter<String> {
+
+    public ProtoDescriptorsFileConverter(String allowedValues) {}
+
+    @Override
+    public Class<String> getParameterClass() {
+      return String.class;
+    }
+
+    @Override
+    public String convert(String filePath) {
+      if (Strings.isNullOrEmpty(filePath)) {
+        return null;
+      }
+      return filePath;
+    }
+  }
 }

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/Connection.java

@@ -47,6 +47,7 @@
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
 
 /**
  * Internal connection API for Google Cloud Spanner. This interface may introduce breaking changes
@@ -403,6 +404,25 @@ default boolean isExcludeTxnFromChangeStreams() {
     throw new UnsupportedOperationException();
   }
 
+  /**
+   * Sets the proto descriptors to use for the next DDL statement (single or batch) that will be
+   * executed. The proto descriptor is automatically cleared after the statement is executed.
+   *
+   * @param protoDescriptors The proto descriptors to use with the next DDL statement (single or
+   *     batch) that will be executed on this connection.
+   */
+  default void setProtoDescriptors(@Nonnull byte[] protoDescriptors) {
+    throw new UnsupportedOperationException();
+  }
+
+  /**
+   * @return The proto descriptor that will be used with the next DDL statement (single or batch)
+   *     that is executed on this connection.
+   */
+  default byte[] getProtoDescriptors() {
+    throw new UnsupportedOperationException();
+  }
+
   /**
    * @return <code>true</code> if this connection will automatically retry read/write transactions
    *     that abort. This method may only be called when the connection is in read/write

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionImpl.java

@@ -22,6 +22,7 @@
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
 import com.google.api.gax.core.GaxProperties;
+import com.google.cloud.ByteArray;
 import com.google.cloud.Timestamp;
 import com.google.cloud.spanner.AsyncResultSet;
 import com.google.cloud.spanner.BatchClient;
@@ -65,6 +66,9 @@
 import io.opentelemetry.api.common.AttributesBuilder;
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.Tracer;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.InputStream;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -81,6 +85,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import org.threeten.bp.Instant;
 
@@ -113,7 +118,7 @@ private LeakedConnectionException() {
     }
   }
 
-  private volatile LeakedConnectionException leakedException;;
+  private volatile LeakedConnectionException leakedException;
   private final SpannerPool spannerPool;
   private AbstractStatementParser statementParser;
   /**
@@ -268,10 +273,11 @@ static UnitOfWorkType of(TransactionMode transactionMode) {
 
   private String transactionTag;
   private String statementTag;
-
   private boolean excludeTxnFromChangeStreams;
 
   private Duration maxCommitDelay;
+  private byte[] protoDescriptors;
+  private String protoDescriptorsFilePath;
 
   /** Create a connection and register it in the SpannerPool. */
   ConnectionImpl(ConnectionOptions options) {
@@ -353,6 +359,7 @@ public Spanner getSpanner() {
   private DdlClient createDdlClient() {
     return DdlClient.newBuilder()
         .setDatabaseAdminClient(spanner.getDatabaseAdminClient())
+        .setProjectId(options.getProjectId())
         .setInstanceId(options.getInstanceId())
         .setDatabaseName(options.getDatabaseName())
         .build();
@@ -763,6 +770,52 @@ public void setExcludeTxnFromChangeStreams(boolean excludeTxnFromChangeStreams)
     this.excludeTxnFromChangeStreams = excludeTxnFromChangeStreams;
   }
 
+  @Override
+  public byte[] getProtoDescriptors() {
+    ConnectionPreconditions.checkState(!isClosed(), CLOSED_ERROR_MSG);
+    if (this.protoDescriptors == null && this.protoDescriptorsFilePath != null) {
+      // Read from file if filepath is valid
+      try {
+        File protoDescriptorsFile = new File(this.protoDescriptorsFilePath);
+        if (!protoDescriptorsFile.isFile()) {
+          throw SpannerExceptionFactory.newSpannerException(
+              ErrorCode.INVALID_ARGUMENT,
+              String.format(
+                  "File %s is not a valid proto descriptors file", this.protoDescriptorsFilePath));
+        }
+        InputStream pdStream = new FileInputStream(protoDescriptorsFile);
+        this.protoDescriptors = ByteArray.copyFrom(pdStream).toByteArray();
+      } catch (Exception exception) {
+        throw SpannerExceptionFactory.newSpannerException(exception);
+      }
+    }
+    return this.protoDescriptors;
+  }
+
+  @Override
+  public void setProtoDescriptors(@Nonnull byte[] protoDescriptors) {
+    Preconditions.checkNotNull(protoDescriptors);
+    ConnectionPreconditions.checkState(!isClosed(), CLOSED_ERROR_MSG);
+    ConnectionPreconditions.checkState(
+        !isBatchActive(), "Proto descriptors cannot be set when a batch is active");
+    this.protoDescriptors = protoDescriptors;
+    this.protoDescriptorsFilePath = null;
+  }
+
+  void setProtoDescriptorsFilePath(@Nonnull String protoDescriptorsFilePath) {
+    Preconditions.checkNotNull(protoDescriptorsFilePath);
+    ConnectionPreconditions.checkState(!isClosed(), CLOSED_ERROR_MSG);
+    ConnectionPreconditions.checkState(
+        !isBatchActive(), "Proto descriptors file path cannot be set when a batch is active");
+    this.protoDescriptorsFilePath = protoDescriptorsFilePath;
+    this.protoDescriptors = null;
+  }
+
+  String getProtoDescriptorsFilePath() {
+    ConnectionPreconditions.checkState(!isClosed(), CLOSED_ERROR_MSG);
+    return this.protoDescriptorsFilePath;
+  }
+
   /**
    * Throws an {@link SpannerException} with code {@link ErrorCode#FAILED_PRECONDITION} if the
    * current state of this connection does not allow changing the setting for retryAbortsInternally.
@@ -1806,6 +1859,7 @@ UnitOfWork createNewUnitOfWork(
               .setSpan(
                   createSpanForUnitOfWork(
                       statementType == StatementType.DDL ? DDL_STATEMENT : SINGLE_USE_TRANSACTION))
+              .setProtoDescriptors(getProtoDescriptors())
               .build();
       if (!isInternalMetadataQuery && !forceSingleUse) {
         // Reset the transaction options after starting a single-use transaction.
@@ -1862,6 +1916,7 @@ UnitOfWork createNewUnitOfWork(
               .setStatementTimeout(statementTimeout)
               .withStatementExecutor(statementExecutor)
               .setSpan(createSpanForUnitOfWork(DDL_BATCH))
+              .setProtoDescriptors(getProtoDescriptors())
               .build();
         default:
       }
@@ -1885,7 +1940,11 @@ private void popUnitOfWorkFromTransactionStack() {
   }
 
   private ApiFuture<Void> executeDdlAsync(CallType callType, ParsedStatement ddl) {
-    return getOrStartDdlUnitOfWork().executeDdlAsync(callType, ddl);
+    ApiFuture<Void> result = getOrStartDdlUnitOfWork().executeDdlAsync(callType, ddl);
+    // reset proto descriptors after executing a DDL statement
+    this.protoDescriptors = null;
+    this.protoDescriptorsFilePath = null;
+    return result;
   }
 
   @Override
@@ -1985,6 +2044,11 @@ public ApiFuture<long[]> runBatchAsync() {
       }
       return ApiFutures.immediateFuture(new long[0]);
     } finally {
+      if (isDdlBatchActive()) {
+        // reset proto descriptors after executing a DDL batch
+        this.protoDescriptors = null;
+        this.protoDescriptorsFilePath = null;
+      }
       this.batchMode = BatchMode.NONE;
       setDefaultTransactionOptions();
     }

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionStatementExecutor.java

@@ -138,6 +138,14 @@ StatementResult statementSetPgSessionCharacteristicsTransactionMode(
 
   StatementResult statementShowTransactionIsolationLevel();
 
+  StatementResult statementSetProtoDescriptors(byte[] protoDescriptors);
+
+  StatementResult statementSetProtoDescriptorsFilePath(String filePath);
+
+  StatementResult statementShowProtoDescriptors();
+
+  StatementResult statementShowProtoDescriptorsFilePath();
+
   StatementResult statementExplain(String sql);
 
   StatementResult statementShowDataBoostEnabled();