feat: enable client to server compression

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

@@ -1056,7 +1056,8 @@ public Builder setCallCredentialsProvider(CallCredentialsProvider callCredential
 
     /**
      * Sets the compression to use for all gRPC calls. The compressor must be a valid name known in
-     * the {@link CompressorRegistry}.
+     * the {@link CompressorRegistry}. This will enable compression both from the client to the
+     * server and from the server to the client.
      *
      * <p>Supported values are:
      *

google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java

@@ -389,6 +389,7 @@ public GapicSpannerRpc(final SpannerOptions options) {
                           MoreObjects.firstNonNull(
                               options.getInterceptorProvider(),
                               SpannerInterceptorProvider.createDefault()))
+                      // This sets the response compressor (Server -> Client).
                       .withEncoding(compressorName))
               .setHeaderProvider(headerProviderWithUserAgent)
               // Attempts direct access to spanner service over gRPC to improve throughput,
@@ -1901,6 +1902,10 @@ <ReqT, RespT> GrpcCallContext newCallContext(
     if (options != null) {
       context = context.withChannelAffinity(Option.CHANNEL_HINT.getLong(options).intValue());
     }
+    if (compressorName != null) {
+      // This sets the compressor for Client -> Server.
+      context = context.withCallOptions(context.getCallOptions().withCompression(compressorName));
+    }
     context = context.withExtraHeaders(metadataProvider.newExtraHeaders(resource, projectName));
     if (callCredentialsProvider != null) {
       CallCredentials callCredentials = callCredentialsProvider.getCallCredentials();

google-cloud-spanner/src/test/java/com/google/cloud/spanner/ChannelUsageTest.java

@@ -130,6 +130,12 @@ public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
                       ServerCall<ReqT, RespT> call,
                       Metadata headers,
                       ServerCallHandler<ReqT, RespT> next) {
+                    // Verify that the compressor name header is set.
+                    assertEquals(
+                        "gzip",
+                        headers.get(
+                            Metadata.Key.of(
+                                "x-response-encoding", Metadata.ASCII_STRING_MARSHALLER)));
                     Attributes attributes = call.getAttributes();
                     @SuppressWarnings({"unchecked", "deprecation"})
                     Attributes.Key<InetSocketAddress> key =
@@ -179,6 +185,7 @@ private SpannerOptions createSpannerOptions() {
                   return input;
                 })
             .setNumChannels(numChannels)
+            .setCompressorName("gzip")
             .setSessionPoolOption(
                 SessionPoolOptions.newBuilder()
                     .setMinSessions(numChannels * 2)

google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpcTest.java

@@ -19,6 +19,7 @@
 import static com.google.common.truth.Truth.assertThat;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assume.assumeTrue;
 
@@ -259,6 +260,58 @@ public void testNoCallCredentials() {
     rpc.shutdown();
   }
 
+  @Test
+  public void testClientCompressorGzip() {
+    SpannerOptions options =
+        SpannerOptions.newBuilder().setProjectId("some-project").setCompressorName("gzip").build();
+    GapicSpannerRpc rpc = new GapicSpannerRpc(options, false);
+    assertEquals(
+        "gzip",
+        rpc.newCallContext(
+                optionsMap,
+                "/some/resource",
+                GetSessionRequest.getDefaultInstance(),
+                SpannerGrpc.getGetSessionMethod())
+            .getCallOptions()
+            .getCompressor());
+    rpc.shutdown();
+  }
+
+  @Test
+  public void testClientCompressorIdentity() {
+    SpannerOptions options =
+        SpannerOptions.newBuilder()
+            .setProjectId("some-project")
+            .setCompressorName("identity")
+            .build();
+    GapicSpannerRpc rpc = new GapicSpannerRpc(options, false);
+    assertEquals(
+        "identity",
+        rpc.newCallContext(
+                optionsMap,
+                "/some/resource",
+                GetSessionRequest.getDefaultInstance(),
+                SpannerGrpc.getGetSessionMethod())
+            .getCallOptions()
+            .getCompressor());
+    rpc.shutdown();
+  }
+
+  @Test
+  public void testClientCompressorDefault() {
+    SpannerOptions options = SpannerOptions.newBuilder().setProjectId("some-project").build();
+    GapicSpannerRpc rpc = new GapicSpannerRpc(options, false);
+    assertNull(
+        rpc.newCallContext(
+                optionsMap,
+                "/some/resource",
+                GetSessionRequest.getDefaultInstance(),
+                SpannerGrpc.getGetSessionMethod())
+            .getCallOptions()
+            .getCompressor());
+    rpc.shutdown();
+  }
+
   private static final class TimeoutHolder {
 
     private Duration timeout;