chore: address PR comments.

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

@@ -1165,6 +1165,19 @@ public Builder setAsyncExecutorProvider(CloseableExecutorProvider provider) {
      *
      * <p>DirectedReadOptions set at the request level will take precedence over the options set
      * using this method.
+     *
+     * <p>An example below of how {@link DirectedReadOptions} can be constructed by including a
+     * replica.
+     *
+     * <pre><code>
+     * DirectedReadOptions.newBuilder()
+     *           .setIncludeReplicas(
+     *               IncludeReplicas.newBuilder()
+     *                   .addReplicaSelections(
+     *                       ReplicaSelection.newBuilder().setLocation("us-east1").build()))
+     *           .build();
+     *           }
+     * </code></pre>
      */
     public Builder setDirectedReadOptions(DirectedReadOptions directedReadOptions) {
       this.directedReadOptions =

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

@@ -2976,6 +2976,7 @@ public void testBackendPartitionQueryOptions() {
             .setChannelProvider(channelProvider)
             .setCredentials(NoCredentials.getInstance())
             .setSessionPoolOption(SessionPoolOptions.newBuilder().setMinSessions(0).build())
+            .setDirectedReadOptions(DIRECTED_READ_OPTIONS2)
             .build()
             .getService()) {
       BatchClient client =
@@ -3015,6 +3016,56 @@ public void testBackendPartitionQueryOptions() {
     }
   }
 
+  @Test
+  public void
+      testBackendPartitionQueryOptions_whenDirectedReadOptionsViaSpannerOptions_assertOptions() {
+    // Use a Spanner instance with MinSession=0 to prevent background requests
+    // from the session pool interfering with the test case.
+    try (Spanner spanner =
+        SpannerOptions.newBuilder()
+            .setProjectId("[PROJECT]")
+            .setChannelProvider(channelProvider)
+            .setCredentials(NoCredentials.getInstance())
+            .setSessionPoolOption(SessionPoolOptions.newBuilder().setMinSessions(0).build())
+            .setDirectedReadOptions(DIRECTED_READ_OPTIONS2)
+            .build()
+            .getService()) {
+      BatchClient client =
+          spanner.getBatchClient(DatabaseId.of("[PROJECT]", "[INSTANCE]", "[DATABASE"));
+      BatchReadOnlyTransaction transaction =
+          client.batchReadOnlyTransaction(TimestampBound.strong());
+      List<Partition> partitions =
+          transaction.partitionQuery(
+              PartitionOptions.newBuilder().setMaxPartitions(10L).build(),
+              Statement.newBuilder(SELECT1.getSql())
+                  .withQueryOptions(
+                      QueryOptions.newBuilder()
+                          .setOptimizerVersion("1")
+                          .setOptimizerStatisticsPackage("custom-package")
+                          .build())
+                  .build());
+      try (ResultSet rs = transaction.execute(partitions.get(0))) {
+        // Just iterate over the results to execute the query.
+        while (rs.next()) {}
+      } finally {
+        transaction.cleanup();
+      }
+      // Check if the last query executed is a DeleteSessionRequest and the second last query
+      // executed is a ExecuteSqlRequest and was executed using a custom optimizer version,
+      // statistics package and directed read options.
+      List<AbstractMessage> requests = mockSpanner.getRequests();
+      assert requests.size() >= 2 : "required to have at least 2 requests";
+      assertThat(requests.get(requests.size() - 1)).isInstanceOf(DeleteSessionRequest.class);
+      assertThat(requests.get(requests.size() - 2)).isInstanceOf(ExecuteSqlRequest.class);
+      ExecuteSqlRequest executeSqlRequest = (ExecuteSqlRequest) requests.get(requests.size() - 2);
+      assertThat(executeSqlRequest.getQueryOptions()).isNotNull();
+      assertThat(executeSqlRequest.getQueryOptions().getOptimizerVersion()).isEqualTo("1");
+      assertThat(executeSqlRequest.getQueryOptions().getOptimizerStatisticsPackage())
+          .isEqualTo("custom-package");
+      assertThat(executeSqlRequest.getDirectedReadOptions()).isEqualTo(DIRECTED_READ_OPTIONS2);
+    }
+  }
+
   @Test
   public void testBackendPartitionReadOptions() {
     // Use a Spanner instance with MinSession=0 to prevent background requests
@@ -3025,6 +3076,7 @@ public void testBackendPartitionReadOptions() {
             .setChannelProvider(channelProvider)
             .setCredentials(NoCredentials.getInstance())
             .setSessionPoolOption(SessionPoolOptions.newBuilder().setMinSessions(0).build())
+            .setDirectedReadOptions(DIRECTED_READ_OPTIONS2)
             .build()
             .getService()) {
       BatchClient client =
@@ -3056,6 +3108,48 @@ public void testBackendPartitionReadOptions() {
     }
   }
 
+  @Test
+  public void
+      testBackendPartitionReadOptions_whenDirectedReadOptionsViaSpannerOptions_assertOptions() {
+    // Use a Spanner instance with MinSession=0 to prevent background requests
+    // from the session pool interfering with the test case.
+    try (Spanner spanner =
+        SpannerOptions.newBuilder()
+            .setProjectId("[PROJECT]")
+            .setChannelProvider(channelProvider)
+            .setCredentials(NoCredentials.getInstance())
+            .setSessionPoolOption(SessionPoolOptions.newBuilder().setMinSessions(0).build())
+            .setDirectedReadOptions(DIRECTED_READ_OPTIONS2)
+            .build()
+            .getService()) {
+      BatchClient client =
+          spanner.getBatchClient(DatabaseId.of("[PROJECT]", "[INSTANCE]", "[DATABASE"));
+      BatchReadOnlyTransaction transaction =
+          client.batchReadOnlyTransaction(TimestampBound.strong());
+      List<Partition> partitions =
+          transaction.partitionRead(
+              PartitionOptions.newBuilder().setMaxPartitions(10L).build(),
+              "FOO",
+              KeySet.all(),
+              Lists.newArrayList("1"));
+      try (ResultSet rs = transaction.execute(partitions.get(0))) {
+        // Just iterate over the results to execute the query.
+        while (rs.next()) {}
+      } finally {
+        transaction.cleanup();
+      }
+      // Check if the last query executed is a DeleteSessionRequest and the second last query
+      // executed is a ExecuteSqlRequest and was executed using a custom optimizer version,
+      // statistics package and directed read options.
+      List<AbstractMessage> requests = mockSpanner.getRequests();
+      assert requests.size() >= 2 : "required to have at least 2 requests";
+      assertThat(requests.get(requests.size() - 1)).isInstanceOf(DeleteSessionRequest.class);
+      assertThat(requests.get(requests.size() - 2)).isInstanceOf(ReadRequest.class);
+      ReadRequest readRequest = (ReadRequest) requests.get(requests.size() - 2);
+      assertThat(readRequest.getDirectedReadOptions()).isEqualTo(DIRECTED_READ_OPTIONS2);
+    }
+  }
+
   @Test
   public void testAsyncQuery() throws Exception {
     final int EXPECTED_ROW_COUNT = 10;