feat(spanner): support directed reads

src/spanner/src/client.rs

@@ -20,6 +20,8 @@ pub use crate::database_client::DatabaseClient;
 pub use crate::error::SpannerInternalError;
 pub use crate::from_value::{ConvertError, FromValue};
 pub use crate::key::{Key, KeyRange, KeySet, KeySetBuilder};
+pub use crate::model::DirectedReadOptions;
+pub use crate::model::directed_read_options;
 pub use crate::model::transaction_options::IsolationLevel;
 pub use crate::model::transaction_options::read_write::ReadLockMode;
 pub use crate::mutation::{Mutation, ValueBinder, WriteBuilder};

src/spanner/src/read.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use crate::key::KeySet;
+use crate::model::DirectedReadOptions;
 
 /// Represents an incomplete read operation that requires specifying keys.
 ///
@@ -61,6 +62,7 @@ impl ReadRequestBuilder {
             columns: self.columns,
             limit: None,
             request_options: None,
+            directed_read_options: None,
         }
     }
 
@@ -87,6 +89,7 @@ impl ReadRequestBuilder {
             columns: self.columns,
             limit: None,
             request_options: None,
+            directed_read_options: None,
         }
     }
 }
@@ -100,6 +103,7 @@ pub struct ConfiguredReadRequestBuilder {
     columns: Vec<String>,
     limit: Option<i64>,
     request_options: Option<crate::model::RequestOptions>,
+    directed_read_options: Option<DirectedReadOptions>,
 }
 
 impl ConfiguredReadRequestBuilder {
@@ -139,6 +143,26 @@ impl ConfiguredReadRequestBuilder {
         self
     }
 
+    /// Sets the directed read options for this request.
+    ///
+    /// ```
+    /// # use google_cloud_spanner::client::ReadRequest;
+    /// # use google_cloud_spanner::client::KeySet;
+    /// # use google_cloud_spanner::client::DirectedReadOptions;
+    /// let dro = DirectedReadOptions::default();
+    /// let req = ReadRequest::builder("MyTable", vec!["col1"])
+    ///     .with_keys(KeySet::all())
+    ///     .with_directed_read_options(dro)
+    ///     .build();
+    /// ```
+    ///
+    /// DirectedReadOptions can only be specified for a read-only transaction,
+    /// otherwise Spanner returns an INVALID_ARGUMENT error.
+    pub fn with_directed_read_options(mut self, options: DirectedReadOptions) -> Self {
+        self.directed_read_options = Some(options);
+        self
+    }
+
     /// Builds the configured `ReadRequest`.
     pub fn build(self) -> ReadRequest {
         ReadRequest {
@@ -148,6 +172,7 @@ impl ConfiguredReadRequestBuilder {
             columns: self.columns,
             limit: self.limit,
             request_options: self.request_options,
+            directed_read_options: self.directed_read_options,
         }
     }
 }
@@ -164,6 +189,7 @@ pub struct ReadRequest {
     pub(crate) columns: Vec<String>,
     pub(crate) limit: Option<i64>,
     pub(crate) request_options: Option<crate::model::RequestOptions>,
+    pub(crate) directed_read_options: Option<DirectedReadOptions>,
 }
 
 impl ReadRequest {
@@ -186,44 +212,23 @@ impl ReadRequest {
         }
     }
 
-    fn into_parts(
-        self,
-    ) -> (
-        String,
-        Option<String>,
-        crate::model::KeySet,
-        Vec<String>,
-        Option<i64>,
-        Option<crate::model::RequestOptions>,
-    ) {
-        (
-            self.table,
-            self.index,
-            self.keys.into_proto(),
-            self.columns,
-            self.limit,
-            self.request_options,
-        )
-    }
-
     pub(crate) fn into_request(self) -> crate::model::ReadRequest {
-        let (table, index, keys, columns, limit, request_options) = self.into_parts();
         crate::model::ReadRequest::default()
-            .set_table(table)
-            .set_columns(columns)
-            .set_key_set(keys)
-            .set_index(index.unwrap_or_default())
-            .set_limit(limit.unwrap_or_default())
-            .set_or_clear_request_options(request_options)
+            .set_table(self.table)
+            .set_columns(self.columns)
+            .set_key_set(self.keys.into_proto())
+            .set_index(self.index.unwrap_or_default())
+            .set_limit(self.limit.unwrap_or_default())
+            .set_or_clear_request_options(self.request_options)
+            .set_or_clear_directed_read_options(self.directed_read_options)
     }
 
     pub(crate) fn into_partition_read_request(self) -> crate::model::PartitionReadRequest {
-        let (table, index, keys, columns, _limit, _request_options) = self.into_parts();
         crate::model::PartitionReadRequest::default()
-            .set_table(table)
-            .set_columns(columns)
-            .set_key_set(keys)
-            .set_index(index.unwrap_or_default())
+            .set_table(self.table)
+            .set_columns(self.columns)
+            .set_key_set(self.keys.into_proto())
+            .set_index(self.index.unwrap_or_default())
     }
 }
 
@@ -286,4 +291,14 @@ mod tests {
             "tag1"
         );
     }
+
+    #[test]
+    fn with_directed_read_options() {
+        let dro = DirectedReadOptions::default();
+        let req = ReadRequest::builder("MyTable", vec!["col1"])
+            .with_keys(KeySet::all())
+            .with_directed_read_options(dro.clone())
+            .build();
+        assert_eq!(req.directed_read_options, Some(dro));
+    }
 }

src/spanner/src/statement.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::model::DirectedReadOptions;
 use crate::to_value::ToValue;
 use crate::types::Type;
 use crate::value::Value;
@@ -32,6 +33,7 @@ pub struct StatementBuilder {
     params: BTreeMap<String, Value>,
     param_types: BTreeMap<String, Type>,
     request_options: Option<crate::model::RequestOptions>,
+    directed_read_options: Option<DirectedReadOptions>,
 }
 
 impl StatementBuilder {
@@ -41,6 +43,7 @@ impl StatementBuilder {
             params: BTreeMap::new(),
             param_types: BTreeMap::new(),
             request_options: None,
+            directed_read_options: None,
         }
     }
 
@@ -90,13 +93,32 @@ impl StatementBuilder {
         self
     }
 
+    /// Sets the directed read options for this statement.
+    ///
+    /// ```
+    /// # use google_cloud_spanner::client::Statement;
+    /// # use google_cloud_spanner::client::DirectedReadOptions;
+    /// let dro = DirectedReadOptions::default();
+    /// let stmt = Statement::builder("SELECT * FROM users")
+    ///     .with_directed_read_options(dro)
+    ///     .build();
+    /// ```
+    ///
+    /// DirectedReadOptions can only be specified for a read-only transaction,
+    /// otherwise Spanner returns an INVALID_ARGUMENT error.
+    pub fn with_directed_read_options(mut self, options: DirectedReadOptions) -> Self {
+        self.directed_read_options = Some(options);
+        self
+    }
+
     /// Builds and returns the finalized Statement object.
     pub fn build(self) -> Statement {
         Statement {
             sql: self.sql,
             params: self.params,
             param_types: self.param_types,
             request_options: self.request_options,
+            directed_read_options: self.directed_read_options,
         }
     }
 }
@@ -130,6 +152,7 @@ pub struct Statement {
     pub(crate) params: BTreeMap<String, Value>,
     pub(crate) param_types: BTreeMap<String, Type>,
     pub(crate) request_options: Option<crate::model::RequestOptions>,
+    pub(crate) directed_read_options: Option<DirectedReadOptions>,
 }
 
 impl Statement {
@@ -165,12 +188,14 @@ impl Statement {
 
     pub(crate) fn into_request(self) -> crate::model::ExecuteSqlRequest {
         let request_options = self.request_options.clone();
+        let directed_read_options = self.directed_read_options.clone();
         let (sql, params, param_types) = self.into_parts();
         crate::model::ExecuteSqlRequest::default()
             .set_sql(sql)
             .set_or_clear_params(params)
             .set_param_types(param_types)
             .set_or_clear_request_options(request_options)
+            .set_or_clear_directed_read_options(directed_read_options)
     }
 
     pub(crate) fn into_batch_statement(self) -> crate::model::execute_batch_dml_request::Statement {
@@ -324,4 +349,13 @@ mod tests {
             "tag1"
         );
     }
+
+    #[test]
+    fn with_directed_read_options() {
+        let dro = DirectedReadOptions::default();
+        let stmt = Statement::builder("SELECT * FROM users")
+            .with_directed_read_options(dro.clone())
+            .build();
+        assert_eq!(stmt.directed_read_options, Some(dro));
+    }
 }

tests/spanner/src/directed_read.rs

@@ -0,0 +1,83 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use directed_read_options::{IncludeReplicas, Replicas};
+use google_cloud_spanner::client::{
+    DatabaseClient, DirectedReadOptions, KeySet, ReadRequest, directed_read_options,
+};
+
+pub async fn read_only_with_directed_read(db_client: &DatabaseClient) -> anyhow::Result<()> {
+    let mut dro = DirectedReadOptions::default();
+    let mut include = IncludeReplicas::default();
+    include.auto_failover_disabled = true;
+    dro.replicas = Some(Replicas::IncludeReplicas(Box::new(include)));
+
+    let read = ReadRequest::builder("AllTypes", vec!["Id"])
+        .with_keys(KeySet::all())
+        .with_directed_read_options(dro)
+        .build();
+
+    let mut result_set = db_client
+        .single_use()
+        .build()
+        .execute_read(read)
+        .await
+        .expect("Failed to execute read with DirectedReadOptions in RO transaction");
+
+    // We don't need to check rows, just that the call succeeded.
+    let _ = result_set.next().await;
+    Ok(())
+}
+
+pub async fn read_write_with_directed_read_error(db_client: &DatabaseClient) -> anyhow::Result<()> {
+    let mut dro = DirectedReadOptions::default();
+    let mut include = IncludeReplicas::default();
+    include.auto_failover_disabled = true;
+    dro.replicas = Some(Replicas::IncludeReplicas(Box::new(include)));
+
+    let read = ReadRequest::builder("AllTypes", vec!["Id"])
+        .with_keys(KeySet::all())
+        .with_directed_read_options(dro)
+        .build();
+
+    // Read-write transaction runner
+    let runner = db_client.read_write_transaction().build().await?;
+
+    let result: google_cloud_spanner::Result<()> = runner
+        .run(async |tx| {
+            let read = read.clone();
+            let mut rs = tx.execute_read(read).await?;
+            let _ = rs.next().await;
+            Ok(())
+        })
+        .await;
+
+    assert!(
+        result.is_err(),
+        "Expected read-write transaction with DirectedReadOptions to fail"
+    );
+
+    let err = result.unwrap_err();
+    let err_str = format!("{:?}", err);
+    // The proto documentation states that an INVALID_ARGUMENT error should be returned,
+    // but the emulator returns FailedPrecondition with a specific message.
+    assert!(
+        err_str.contains("FailedPrecondition")
+            || err_str.contains("Directed reads can only be performed in a read-only transaction"),
+        "Expected FailedPrecondition error, got: {}",
+        err_str
+    );
+
+    Ok(())
+}

tests/spanner/src/lib.rs

@@ -14,6 +14,7 @@
 
 pub mod batch_read_only_transaction;
 pub mod client;
+pub mod directed_read;
 pub mod partitioned_dml;
 pub mod query;
 pub mod read;

tests/spanner/tests/driver.rs

@@ -114,4 +114,18 @@ mod spanner {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn run_directed_read_tests() -> anyhow::Result<()> {
+        let db_client = match integration_tests_spanner::client::create_database_client().await {
+            Some(c) => c,
+            None => return Ok(()),
+        };
+
+        integration_tests_spanner::directed_read::read_only_with_directed_read(&db_client).await?;
+        integration_tests_spanner::directed_read::read_write_with_directed_read_error(&db_client)
+            .await?;
+
+        Ok(())
+    }
 }